behave-html-formatter>=0.9.4;python_version>='3.6'
mock!=4.0.0,!=4.0.1 # BSD
PrettyTable<0.8 # BSD
-six # MIT
python-subunit # Apache-2.0/BSD
os-testr # Apache-2.0
junitxml
import enum
import pkg_resources
import prettytable
-import six
from stevedore import driver
import yaml
os.makedirs(constants.RESULTS_DIR)
except OSError as ex:
if ex.errno != errno.EEXIST:
- six.print_("{} {}".format("Cannot create", constants.RESULTS_DIR))
+ print("{} {}".format("Cannot create", constants.RESULTS_DIR))
return testcase.TestCase.EX_RUN_ERROR
if env.get('DEBUG').lower() == 'true':
logging.config.fileConfig(pkg_resources.resource_filename(
import time
import json
-import six
from behave.__main__ import main as behave_main
config = ['--tags='+','.join(tags),
'--junit', '--junit-directory={}'.format(self.res_dir),
'--format=json', '--outfile={}'.format(self.json_file)]
- if six.PY3:
- html_file = os.path.join(self.res_dir, 'output.html')
- config += ['--format=behave_html_formatter:HTMLFormatter',
- '--outfile={}'.format(html_file)]
+ html_file = os.path.join(self.res_dir, 'output.html')
+ config += ['--format=behave_html_formatter:HTMLFormatter',
+ '--outfile={}'.format(html_file)]
if kwargs.get("console", False):
config += ['--format=pretty', '--outfile=-']
for feature in suites:
import re
import zipfile
+from urllib.parse import urlparse
import boto3
from boto3.s3.transfer import TransferConfig
import botocore
import pkg_resources
import requests
-from six.moves import urllib
from xtesting.core import testcase
from xtesting.utils import env
multipart_threshold = 5 * 1024 ** 5 if "google" in os.environ[
"S3_ENDPOINT_URL"] else 8 * 1024 * 1024
config = TransferConfig(multipart_threshold=multipart_threshold)
- bucket_name = urllib.parse.urlparse(dst_s3_url).netloc
+ bucket_name = urlparse(dst_s3_url).netloc
s3path = re.search(
- '^/*(.*)/*$', urllib.parse.urlparse(dst_s3_url).path).group(1)
+ '^/*(.*)/*$', urlparse(dst_s3_url).path).group(1)
prefix = os.path.join(s3path, build_tag)
# pylint: disable=no-member
for s3_object in b3resource.Bucket(bucket_name).objects.filter(
multipart_threshold = 5 * 1024 ** 5 if "google" in os.environ[
"S3_ENDPOINT_URL"] else 8 * 1024 * 1024
config = TransferConfig(multipart_threshold=multipart_threshold)
- bucket_name = urllib.parse.urlparse(dst_s3_url).netloc
+ bucket_name = urlparse(dst_s3_url).netloc
mime_type = mimetypes.guess_type('{}.zip'.format(build_tag))
- path = urllib.parse.urlparse(dst_s3_url).path.strip("/")
+ path = urlparse(dst_s3_url).path.strip("/")
# pylint: disable=no-member
b3resource.Bucket(bucket_name).upload_file(
'{}.zip'.format(build_tag),
import sys
import time
-import six
from xtesting.core import testcase
__author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, "
"Cedric Ollivier <cedric.ollivier@orange.com>")
-@six.add_metaclass(abc.ABCMeta)
-class Feature(testcase.TestCase):
+class Feature(testcase.TestCase, metaclass=abc.ABCMeta):
"""Base model for single feature."""
__logger = logging.getLogger(__name__)
from lxml import etree
import prettytable
-import six
from xtesting.core import testcase
if console:
sys.stdout.write(line.decode("utf-8"))
f_stdout.write(line.decode("utf-8"))
- if six.PY3:
- try:
- process.wait(timeout=max_duration)
- except subprocess.TimeoutExpired:
- process.kill()
- self.__logger.info(
- "Killing MTS process after %d second(s).",
- max_duration)
- return 3
- else:
- process.wait()
+ try:
+ process.wait(timeout=max_duration)
+ except subprocess.TimeoutExpired:
+ process.kill()
+ self.__logger.info(
+ "Killing MTS process after %d second(s).",
+ max_duration)
+ return 3
with open(self.result_file, 'r') as f_stdin:
self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
return process.returncode
import logging
import os
+from io import StringIO
import robot.api
from robot.errors import RobotError
from robot.reporting import resultwriter
import robot.run
from robot.utils.robottime import timestamp_to_secs
-from six import StringIO
from xtesting.core import testcase
import mimetypes
import os
import re
-import sys
+from urllib.parse import urlparse
import boto3
from boto3.s3.transfer import TransferConfig
import botocore
import prettytable
import requests
-import six
-from six.moves import urllib
from xtesting.utils import decorators
from xtesting.utils import env
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
-@six.add_metaclass(abc.ABCMeta)
-class TestCase():
+class TestCase(metaclass=abc.ABCMeta):
# pylint: disable=too-many-instance-attributes
"""Base model for single test case."""
url, data=json.dumps(data, sort_keys=True),
headers=self.headers)
req.raise_for_status()
- if urllib.parse.urlparse(url).scheme != "file":
+ if urlparse(url).scheme != "file":
# href must be postprocessed as OPNFV testapi is misconfigured
# (localhost is returned)
uid = re.sub(r'^.*/api/v1/results/*', '', req.json()["href"])
multipart_threshold = 5 * 1024 ** 5 if "google" in os.environ[
"S3_ENDPOINT_URL"] else 8 * 1024 * 1024
config = TransferConfig(multipart_threshold=multipart_threshold)
- bucket_name = urllib.parse.urlparse(dst_s3_url).netloc
+ bucket_name = urlparse(dst_s3_url).netloc
try:
b3resource.meta.client.head_bucket(Bucket=bucket_name)
except botocore.exceptions.ClientError as exc:
# pylint: disable=no-member
b3resource.create_bucket(Bucket=bucket_name)
else:
- typ, value, traceback = sys.exc_info()
- six.reraise(typ, value, traceback)
- except Exception: # pylint: disable=broad-except
- typ, value, traceback = sys.exc_info()
- six.reraise(typ, value, traceback)
- path = urllib.parse.urlparse(dst_s3_url).path.strip("/")
+ raise exc
+ except Exception as exc: # pylint: disable=broad-except
+ raise exc
+ path = urlparse(dst_s3_url).path.strip("/")
dst_http_url = os.environ["HTTP_DST_URL"]
output_str = "\n"
self.details["links"] = []
"""Define the parent class to run unittest.TestSuite as TestCase."""
from __future__ import division
-
+from io import BytesIO
import logging
import os
import shutil
import unittest
from subunit.run import SubunitTestRunner
-import six
from xtesting.core import testcase
self.start_time = time.time()
if not os.path.isdir(self.res_dir):
os.makedirs(self.res_dir)
- stream = six.BytesIO()
+ stream = BytesIO()
result = SubunitTestRunner(
stream=stream, verbosity=2).run(self.suite).decorated
self.generate_stats(stream)
import time
-import six
-
from xtesting.core import testcase
def run(self, **kwargs):
self.start_time = time.time()
- six.print_("Hello World")
+ print("Hello World")
self.result = 100
self.stop_time = time.time()
# pylint: disable=missing-docstring
-import six
-
from xtesting.core import feature
class Test(feature.Feature):
def execute(self, **kwargs):
- six.print_("Hello World")
+ print("Hello World")
return 0
except Exception: # pylint: disable=broad-except
pass
envfile = 'rc_file'
- with mock.patch('six.moves.builtins.open',
+ with mock.patch('builtins.open',
mock.mock_open(read_data=msg)) as mock_method,\
mock.patch('os.path.isfile', return_value=True):
mock_method.return_value.__iter__ = lambda self: iter(
'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
def test_get_dict_by_test(self):
- with mock.patch('six.moves.builtins.open', mock.mock_open()), \
+ with mock.patch('builtins.open', mock.mock_open()), \
mock.patch('yaml.safe_load') as mock_yaml:
mock_obj = mock.Mock()
testcase_dict = {'case_name': 'testname',
with mock.patch('xtesting.ci.tier_builder.yaml.safe_load',
return_value=self.mock_yaml), \
- mock.patch('six.moves.builtins.open', mock.mock_open()):
+ mock.patch('builtins.open', mock.mock_open()):
os.environ["INSTALLER_TYPE"] = 'test_installer'
os.environ["DEPLOY_SCENARIO"] = 'test_scenario'
self.tierbuilder = tier_builder.TierBuilder('testcases_file')
import unittest
import mock
-import six
from xtesting.core import behaveframework
self.test = behaveframework.BehaveFramework(
case_name='behave', project_name='xtesting')
- @mock.patch('six.moves.builtins.open', side_effect=OSError)
+ @mock.patch('builtins.open', side_effect=OSError)
def test_raises_exc_open(self, *args): # pylint: disable=unused-argument
with self.assertRaises(OSError):
self.test.parse_results()
@mock.patch('json.load', return_value=[{'foo': 'bar'}])
- @mock.patch('six.moves.builtins.open', mock.mock_open())
+ @mock.patch('builtins.open', mock.mock_open())
def test_raises_exc_key(self, *args): # pylint: disable=unused-argument
with self.assertRaises(KeyError):
self.test.parse_results()
@mock.patch('json.load', return_value=[])
- @mock.patch('six.moves.builtins.open', mock.mock_open())
+ @mock.patch('builtins.open', mock.mock_open())
def test_raises_exe_zerodivision(self, *args):
# pylint: disable=unused-argument
with self.assertRaises(ZeroDivisionError):
self.test.parse_results()
def _test_result(self, response, result):
- with mock.patch('six.moves.builtins.open', mock.mock_open()), \
+ with mock.patch('builtins.open', mock.mock_open()), \
mock.patch('json.load', return_value=response):
self.test.parse_results()
self.assertEqual(self.test.result, result)
data = [{'status': 'passed'}, {'status': 'passed'}]
self._test_result(data, 100)
- @mock.patch('six.moves.builtins.open', mock.mock_open())
+ @mock.patch('builtins.open', mock.mock_open())
def test_count(self, *args): # pylint: disable=unused-argument
self._response.extend([{'status': 'failed'}, {'status': 'skipped'}])
with mock.patch('json.load', mock.Mock(return_value=self._response)):
args_list = [
'--tags=', '--junit',
'--junit-directory={}'.format(self.test.res_dir),
- '--format=json', '--outfile={}'.format(self.test.json_file)]
- if six.PY3:
- args_list += [
- '--format=behave_html_formatter:HTMLFormatter',
- '--outfile={}'.format(html_file)]
+ '--format=json', '--outfile={}'.format(self.test.json_file),
+ '--format=behave_html_formatter:HTMLFormatter',
+ '--outfile={}'.format(html_file)]
args_list.append('foo')
args[0].assert_called_once_with(args_list)
mock_method.assert_called_once_with()
args_list = [
'--tags=', '--junit',
'--junit-directory={}'.format(self.test.res_dir),
- '--format=json', '--outfile={}'.format(self.test.json_file)]
- if six.PY3:
- args_list += [
- '--format=behave_html_formatter:HTMLFormatter',
- '--outfile={}'.format(html_file)]
+ '--format=json', '--outfile={}'.format(self.test.json_file),
+ '--format=behave_html_formatter:HTMLFormatter',
+ '--outfile={}'.format(html_file)]
if console:
args_list += ['--format=pretty', '--outfile=-']
args_list.append('foo')
import os
+from io import BytesIO
import logging
import subprocess
import unittest
import mock
-import six
from xtesting.core import feature
from xtesting.core import testcase
# logging must be disabled else it calls time.time()
# what will break these unit tests.
logging.disable(logging.CRITICAL)
- with mock.patch('six.moves.builtins.open'):
+ with mock.patch('builtins.open'):
self.feature = FakeTestCase(
project_name=self._project_name, case_name=self._case_name)
# logging must be disabled else it calls time.time()
# what will break these unit tests.
logging.disable(logging.CRITICAL)
- with mock.patch('six.moves.builtins.open'):
+ with mock.patch('builtins.open'):
self.feature = feature.BashFeature(
project_name=self._project_name, case_name=self._case_name)
@mock.patch('subprocess.Popen',
side_effect=subprocess.CalledProcessError(0, '', ''))
def test_run_ko1(self, *args):
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run(testcase.TestCase.EX_RUN_ERROR)
mopen.assert_called_once_with(self._output_file, "w")
args[0].assert_called_once_with(
@mock.patch('os.path.isdir', return_value=True)
@mock.patch('subprocess.Popen')
def test_run_ko2(self, *args):
- stream = six.BytesIO()
+ stream = BytesIO()
stream.write(b"foo")
stream.seek(0)
attrs = {'return_value.stdout': stream, 'return_value.returncode': 1}
args[0].configure_mock(**attrs)
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run(testcase.TestCase.EX_RUN_ERROR)
self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
@mock.patch('os.path.isdir', return_value=True)
@mock.patch('subprocess.Popen')
def test_run1(self, *args):
- stream = six.BytesIO()
+ stream = BytesIO()
stream.write(b"foo")
stream.seek(0)
attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
args[0].configure_mock(**attrs)
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run(testcase.TestCase.EX_OK)
self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
@mock.patch('os.path.isdir', return_value=True)
@mock.patch('subprocess.Popen')
def test_run2(self, *args):
- stream = six.BytesIO()
+ stream = BytesIO()
stream.write(b"foo")
stream.seek(0)
attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
args[0].configure_mock(**attrs)
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run_console(True, testcase.TestCase.EX_OK)
self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
@mock.patch('os.path.isdir', return_value=True)
@mock.patch('subprocess.Popen')
def test_run3(self, *args):
- stream = six.BytesIO()
+ stream = BytesIO()
stream.write(b"foo")
stream.seek(0)
attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
args[0].configure_mock(**attrs)
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run_console(False, testcase.TestCase.EX_OK)
self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
@mock.patch('os.path.isdir', return_value=False)
@mock.patch('subprocess.Popen')
def test_run4(self, *args):
- stream = six.BytesIO()
+ stream = BytesIO()
stream.write(b"foo")
stream.seek(0)
attrs = {'return_value.stdout': stream, 'return_value.returncode': 0}
args[0].configure_mock(**attrs)
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen:
+ with mock.patch('builtins.open', mock.mock_open()) as mopen:
self._test_run_console(False, testcase.TestCase.EX_OK)
self.assertIn(mock.call(self._output_file, 'w'), mopen.mock_calls)
self.assertIn(mock.call(self._output_file, 'r'), mopen.mock_calls)
# pylint: disable=missing-docstring
+import io
import logging
import subprocess
import unittest
import mock
-import six
from xtesting.core import unit
from xtesting.core import testcase
@mock.patch('subprocess.Popen', side_effect=Exception)
def test_generate_stats_ko(self, *args):
- stream = six.StringIO()
+ stream = io.StringIO()
with self.assertRaises(Exception):
self.psrunner.generate_stats(stream)
args[0].assert_called_once_with(
return_value=mock.Mock(
communicate=mock.Mock(return_value=(b"foo", b"bar"))))
def test_generate_stats_ok(self, *args):
- stream = six.StringIO()
+ stream = io.StringIO()
self.psrunner.generate_stats(stream)
args[0].assert_called_once_with(
['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
- @mock.patch('six.moves.builtins.open', mock.mock_open())
+ @mock.patch('builtins.open', mock.mock_open())
@mock.patch('subprocess.Popen', side_effect=Exception)
def test_generate_xunit_ko(self, *args):
- stream = six.StringIO()
+ stream = io.StringIO()
with self.assertRaises(Exception), \
- mock.patch('six.moves.builtins.open',
+ mock.patch('builtins.open',
mock.mock_open()) as mock_open:
self.psrunner.generate_xunit(stream)
args[0].assert_called_once_with(
return_value=mock.Mock(
communicate=mock.Mock(return_value=(b"foo", b"bar"))))
def test_generate_xunit_ok(self, *args):
- stream = six.BytesIO()
- with mock.patch('six.moves.builtins.open',
+ stream = io.BytesIO()
+ with mock.patch('builtins.open',
mock.mock_open()) as mock_open:
self.psrunner.generate_xunit(stream)
args[0].assert_called_once_with(
@mock.patch('subunit.run.SubunitTestRunner.run')
def _test_run(self, mock_result, status, result, *args):
args[0].return_value = mock_result
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as m_open:
+ with mock.patch('builtins.open', mock.mock_open()) as m_open:
self.assertEqual(self.psrunner.run(), status)
m_open.assert_called_once_with(
'{}/subunit_stream'.format(self.psrunner.res_dir), 'wb')
@mock.patch('subunit.run.SubunitTestRunner.run')
def _test_run_name(self, name, mock_result, status, result, *args):
args[0].return_value = mock_result
- with mock.patch('six.moves.builtins.open', mock.mock_open()) as m_open:
+ with mock.patch('builtins.open', mock.mock_open()) as m_open:
self.assertEqual(self.psrunner.run(name=name), status)
m_open.assert_called_once_with(
'{}/subunit_stream'.format(self.psrunner.res_dir), 'wb')
decorated=mock.Mock(
testsRun=50, errors=[], failures=[]))
args[3].side_effect = exc
- with mock.patch('six.moves.builtins.open',
+ with mock.patch('builtins.open',
mock.mock_open()) as m_open:
self.assertEqual(
self.psrunner.run(), testcase.TestCase.EX_RUN_ERROR)
import logging
import os
import unittest
-
-from six.moves import reload_module
+import importlib
from xtesting.utils import env
def test_get_unknown_env(self):
self.assertEqual(env.get('FOO'), 'foo')
- reload_module(env)
+ importlib.reload(env)
def test_get_unset_env(self):
del os.environ['CI_LOOP']
import errno
import functools
import os
+from urllib.parse import urlparse
import mock
import requests.sessions
-from six.moves import urllib
def can_dump_request_to_file(method):
def dump_preparedrequest(request, **kwargs):
# pylint: disable=unused-argument
- parseresult = urllib.parse.urlparse(request.url)
+ parseresult = urlparse(request.url)
if parseresult.scheme == "file":
try:
dirname = os.path.dirname(parseresult.path)
def patch_request(method, url, **kwargs):
with requests.sessions.Session() as session:
- parseresult = urllib.parse.urlparse(url)
+ parseresult = urlparse(url)
if parseresult.scheme == "file":
with mock.patch.object(session, 'send',
side_effect=dump_preparedrequest):
Code Review / functest-xtesting.git / commitdiff