未验证 提交 5b7a2a6d 编写于 作者: C Cleber Rosa

Merge remote-tracking branch 'clebergnu/drop_python_2_v3'

Signed-off-by: NCleber Rosa <crosa@redhat.com>
......@@ -5,7 +5,6 @@ matrix:
- python: "nightly"
python:
- "2.7"
- "3.4"
- "3.5"
- "3.6"
......
......@@ -6,7 +6,7 @@ Avocado is a set of tools and libraries to help with automated testing.
One can call it a test framework with benefits. Native tests are
written in Python and they follow the unittest
(https://docs.python.org/2.7/library/unittest.html) pattern, but any
(https://docs.python.org/3.6/library/unittest.html) pattern, but any
executable can serve as a test.
Avocado is composed of:
......@@ -44,8 +44,8 @@ Installing with standard Python tools
-------------------------------------
The simplest installation method is through ``pip``. On most POSIX
systems with Python 2.7 and ``pip`` available, installation can be
performed with a single command::
systems with Python 3.4 (or later) and ``pip`` available, installation
can be performed with a single command::
pip install --user avocado-framework
......@@ -60,7 +60,7 @@ If you want even more isolation, Avocado can also be installed in a
Python virtual environment. with no additional steps besides creating
and activating the "venv" itself::
python -m virtualenv /path/to/new/virtual_environment
python -m venv /path/to/new/virtual_environment
. /path/to/new/virtual_environment/bin/activate
pip install avocado-framework
......
......@@ -31,8 +31,6 @@ import sys
import time
import tempfile
from six.moves import xrange as range
from . import job_id
from . import settings
from . import exit_codes
......
......@@ -27,9 +27,6 @@ import tempfile
import time
import traceback
from six import iteritems
from six.moves import xrange as range
from . import version
from . import data_dir
from . import dispatcher
......@@ -260,7 +257,7 @@ class Job(object):
def __stop_job_logging(self):
if self._stdout_stderr:
sys.stdout, sys.stderr = self._stdout_stderr
for handler, loggers in iteritems(self.__logging_handlers):
for handler, loggers in self.__logging_handlers.items():
for logger in loggers:
logging.getLogger(logger).removeHandler(handler)
......
......@@ -25,7 +25,6 @@ import shlex
import sys
from enum import Enum
from six import string_types, iteritems
from . import data_dir
from . import output
......@@ -229,7 +228,7 @@ class TestLoaderProxy(object):
# Using __func__ to avoid problem with different term_supp instances
healthy_func = getattr(output.TERM_SUPPORT.healthy_str, '__func__')
types = [mapping[_[0]]
for _ in iteritems(plugin.get_decorator_mapping())
for _ in plugin.get_decorator_mapping().items()
if _[1].__func__ is healthy_func]
return [name + '.' + _ for _ in types]
......@@ -390,7 +389,7 @@ class TestLoaderProxy(object):
test_path = test_parameters.pop('modulePath')
else:
test_path = None
if isinstance(test_class, string_types):
if isinstance(test_class, str):
module_name = os.path.basename(test_path).split('.')[0]
test_module_dir = os.path.abspath(os.path.dirname(test_path))
# Tests with local dir imports need this
......@@ -614,13 +613,13 @@ class FileLoader(TestLoader):
# Instrumented tests are defined as string and loaded at the
# execution time.
for tst in tests:
if not isinstance(tst[0], string_types):
if not isinstance(tst[0], str):
return None
else:
test_class = next(key for key, value in iteritems(mapping)
test_class = next(key for key, value in mapping.items()
if value == self.test_type)
for tst in tests:
if (isinstance(tst[0], string_types) or
if (isinstance(tst[0], str) or
not issubclass(tst[0], test_class)):
return None
return tests
......@@ -687,7 +686,7 @@ class FileLoader(TestLoader):
result = []
class_methods = safeloader.find_class_and_methods(test_path,
_RE_UNIT_TEST)
for klass, methods in iteritems(class_methods):
for klass, methods in class_methods.items():
if klass in disabled:
continue
if test_path.endswith(".py"):
......@@ -725,7 +724,7 @@ class FileLoader(TestLoader):
if avocado_tests:
test_factories = []
for test_class, info in avocado_tests.items():
if isinstance(test_class, string_types):
if isinstance(test_class, str):
for test_method, tags in info:
name = test_name + \
':%s.%s' % (test_class, test_method)
......
......@@ -22,8 +22,6 @@ import re
import sys
import traceback
from six import string_types, iterkeys
from . import exit_codes
from ..utils import path as utils_path
from .settings import settings
......@@ -433,7 +431,7 @@ def reconfigure(args):
disable_log_handler(LOG_UI.getChild("debug"))
# Add custom loggers
for name in [_ for _ in enabled if _ not in iterkeys(BUILTIN_STREAMS)]:
for name in [_ for _ in enabled if _ not in BUILTIN_STREAMS]:
stream_level = re.split(r'(?<!\\):', name, maxsplit=1)
name = stream_level[0]
if len(stream_level) == 1:
......@@ -569,11 +567,11 @@ def add_log_handler(logger, klass=logging.StreamHandler, stream=sys.stdout,
:param level: Log level (defaults to `INFO``)
:param fmt: Logging format (defaults to ``%(name)s: %(message)s``)
"""
if isinstance(logger, string_types):
if isinstance(logger, str):
logger = logging.getLogger(logger)
handler = klass(stream)
handler.setLevel(level)
if isinstance(fmt, string_types):
if isinstance(fmt, str):
fmt = logging.Formatter(fmt=fmt)
handler.setFormatter(fmt)
logger.addHandler(handler)
......@@ -582,7 +580,7 @@ def add_log_handler(logger, klass=logging.StreamHandler, stream=sys.stdout,
def disable_log_handler(logger):
if isinstance(logger, string_types):
if isinstance(logger, str):
logger = logging.getLogger(logger)
# Handlers might be reused elsewhere, can't delete them
while logger.handlers:
......
......@@ -18,9 +18,6 @@ Module related to test parameters
import logging
import re
from six import iterkeys, iteritems
from six.moves import xrange as range
class NoMatchError(KeyError):
pass
......@@ -62,9 +59,9 @@ class AvocadoParams(object):
self._logger_name = logger_name
def __eq__(self, other):
if set(iterkeys(self.__dict__)) != set(iterkeys(other.__dict__)):
if set(self.__dict__) != set(other.__dict__):
return False
for attr in iterkeys(self.__dict__):
for attr in self.__dict__:
if (getattr(self, attr) != getattr(other, attr)):
return False
return True
......@@ -265,5 +262,5 @@ class AvocadoParam(object):
which generates lots of duplicate entries due to inherited values.
"""
for leaf in self._leaves:
for key, value in iteritems(leaf.environment):
for key, value in leaf.environment.items():
yield (leaf.environment.origin[key].path, key, value)
......@@ -18,8 +18,6 @@ Avocado application command line parsing.
import argparse
from six import iteritems
from . import exit_codes
from . import varianter
from . import settings
......@@ -84,8 +82,8 @@ class Parser(object):
self.application.add_argument('--config', metavar='CONFIG_FILE',
nargs='?',
help='Use custom configuration from a file')
streams = (['"%s": %s' % _ for _ in iteritems(BUILTIN_STREAMS)] +
['"%s": %s' % _ for _ in iteritems(BUILTIN_STREAM_SETS)])
streams = (['"%s": %s' % _ for _ in BUILTIN_STREAMS.items()] +
['"%s": %s' % _ for _ in BUILTIN_STREAM_SETS.items()])
streams = "; ".join(streams)
self.application.add_argument('--show', action="store",
type=lambda value: value.split(","),
......
......@@ -27,7 +27,6 @@ except ImportError:
from pkg_resources import resource_filename
from pkg_resources import resource_isdir
from pkg_resources import resource_listdir
from six import string_types
from stevedore import ExtensionManager
from ..utils import path
......@@ -97,7 +96,7 @@ def convert_value_type(value, value_type):
except Exception:
sval = value
if isinstance(value_type, string_types):
if isinstance(value_type, str):
if value_type == 'str':
value_type = str
elif value_type == 'bool':
......
......@@ -30,7 +30,6 @@ import time
import unittest
from difflib import unified_diff
from six import string_types, iteritems
from . import data_dir
from . import defaults
......@@ -130,7 +129,7 @@ class TestID(object):
return repr(str(self))
def __eq__(self, other):
if isinstance(other, string_types):
if isinstance(other, str):
return str(self) == other
else:
return self.__dict__ == other.__dict__
......@@ -710,7 +709,7 @@ class Test(unittest.TestCase, TestData):
sys.stderr.rm_logger(LOG_JOB.getChild("stderr"))
if isinstance(sys.stdout, output.LoggingFile):
sys.stdout.rm_logger(LOG_JOB.getChild("stdout"))
for name, handler in iteritems(self._logging_handlers):
for name, handler in self._logging_handlers.items():
logging.getLogger(name).removeHandler(handler)
def _record_reference(self, produced_file_path, reference_file_name):
......@@ -848,7 +847,7 @@ class Test(unittest.TestCase, TestData):
test_exception = details
self.log.debug("Local variables:")
local_vars = inspect.trace()[1][0].f_locals
for key, value in iteritems(local_vars):
for key, value in local_vars.items():
self.log.debug(' -> %s %s: %s', key, type(value), value)
finally:
try:
......
......@@ -38,8 +38,6 @@ import copy
import itertools
import locale
from six import string_types, iteritems
from ..utils import astring
......@@ -96,19 +94,18 @@ class TreeEnvironment(dict):
:param sort: Sorted to provide stable output
:rtype: str
"""
def _iteritems_sorted(dictionary):
return sorted(iteritems(dictionary))
if sort:
sort_fn = sorted
else:
def sort_fn(x):
return x
# Use __str__ instead of __repr__ to improve readability
if self:
if sort:
_iteritems = _iteritems_sorted
else:
_iteritems = iteritems
_values = ["%s: %s" % _ for _ in _iteritems(self)]
_values = ["%s: %s" % _ for _ in sort_fn(self.items())]
values = "{%s}" % ", ".join(_values)
_origin = ["%s: %s" % (key, node.path)
for key, node in _iteritems(self.origin)]
for key, node in sort_fn(self.origin.items())]
origin = "{%s}" % ", ".join(_origin)
else:
values = "{}"
......@@ -212,7 +209,7 @@ class TreeNode(object):
def __eq__(self, other):
""" Compares node to other node or string to name of this node """
if isinstance(other, string_types): # Compare names
if isinstance(other, str): # Compare names
if self.name == other:
return True
else:
......@@ -333,7 +330,7 @@ class TreeNode(object):
if self._environment is None:
self._environment = (self.parent.environment.copy()
if self.parent else TreeEnvironment())
for key, value in iteritems(self.value):
for key, value in self.value.items():
if isinstance(value, list):
if (key in self._environment and
isinstance(self._environment[key], list)):
......@@ -447,13 +444,13 @@ def tree_view(root, verbose=None, use_utf8=None):
right = charset['Right']
out = [node.name]
if verbose is not None and verbose >= 2 and node.is_leaf:
values = itertools.chain(iteritems(node.environment),
values = itertools.chain(iter(node.environment.items()),
[("filter-only", _)
for _ in node.environment.filter_only],
[("filter-out", _)
for _ in node.environment.filter_out])
elif verbose in (1, 3):
values = itertools.chain(iteritems(node.value),
values = itertools.chain(iter(node.value.items()),
[("filter-only", _)
for _ in node.filters[0]],
[("filter-out", _)
......@@ -508,9 +505,9 @@ def tree_view(root, verbose=None, use_utf8=None):
right = charset['Right']
out = []
if verbose is not None and verbose >= 2 and root.is_leaf:
values = iteritems(root.environment)
values = root.environment.items()
elif verbose in (1, 3):
values = iteritems(root.value)
values = root.value.items()
else:
values = None
if values:
......
......@@ -21,8 +21,6 @@ Base classes for implementing the varianter interface
import hashlib
from six import iteritems, itervalues
from . import tree
from . import dispatcher
from . import output
......@@ -81,7 +79,7 @@ def variant_to_str(variant, verbosity, out_args=None, debug=False):
if verbosity:
env = set()
for node in variant["variant"]:
for key, value in iteritems(node.environment):
for key, value in node.environment.items():
origin = node.environment.origin[key].path
env.add(("%s:%s" % (origin, key), astring.to_text(value)))
if not env:
......@@ -103,7 +101,7 @@ def dump_ivariants(ivariants):
return (astring.to_text(node.path),
[(astring.to_text(node.environment.origin[key].path),
astring.to_text(key), value)
for key, value in iteritems(node.environment)])
for key, value in node.environment.items()])
variants = []
for variant in ivariants():
......@@ -149,7 +147,7 @@ class FakeVariantDispatcher(object):
paths))
env = set()
for node in variant["variant"]:
for key, value in iteritems(node.environment):
for key, value in node.environment.items():
origin = node.environment.origin[key].path
env.add(("%s:%s" % (origin, key), astring.to_text(value)))
if not env:
......@@ -196,7 +194,7 @@ class Varianter(object):
:param args: Parsed cmdline arguments
"""
default_params = self.node_class()
for default_param in itervalues(self.default_params):
for default_param in self.default_params.values():
default_params.merge(default_param)
self._default_params = default_params
self.default_params.clear()
......
......@@ -20,14 +20,11 @@ from avocado.utils import distro
from setuptools import setup, find_packages
if sys.version_info[0] == 3:
fabric = 'Fabric3'
else:
fabric = 'fabric>=1.5.4,<2.0.0'
detected_distro = distro.detect()
if detected_distro.name == 'fedora' and int(detected_distro.version) >= 29:
fabric = 'Fabric3>=1.1.4,<2.0.0'
else:
fabric = 'Fabric3'
setup(name='avocado-framework-plugin-runner-remote',
description='Avocado Runner for Remote Execution',
......
......@@ -3,12 +3,7 @@ import glob
import os
import shutil
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.core.job import Job
from avocado.core import exit_codes, version
......@@ -75,25 +70,25 @@ class RemoteTestRunnerTest(unittest.TestCase):
job.setup()
runner = avocado_runner_remote.RemoteTestRunner(job, job.result)
return_value = (True, (version.MAJOR, version.MINOR))
runner.check_remote_avocado = mock.Mock(return_value=return_value)
runner.check_remote_avocado = unittest.mock.Mock(return_value=return_value)
# These are mocked at their source, and will prevent fabric from
# trying to contact remote hosts
with mock.patch('avocado_runner_remote.Remote'):
with unittest.mock.patch('avocado_runner_remote.Remote'):
runner.remote = avocado_runner_remote.Remote(job_args.remote_hostname)
# This is the result that the run_suite() will get from remote.run
remote_run_result = process.CmdResult()
remote_run_result.stdout = JSON_RESULTS
remote_run_result.exit_status = 0
runner.remote.run = mock.Mock(return_value=remote_run_result)
runner.remote.run = unittest.mock.Mock(return_value=remote_run_result)
# We have to fake the uncompressing and removal of the zip
# archive that was never generated on the "remote" end
# This test could be expand by mocking creating an actual
# zip file instead, but it's really overkill
with mock.patch('avocado_runner_remote.archive.uncompress'):
with mock.patch('avocado_runner_remote.os.remove'):
with unittest.mock.patch('avocado_runner_remote.archive.uncompress'):
with unittest.mock.patch('avocado_runner_remote.os.remove'):
runner.run_suite(None, None, 61)
# The job was created with dry_run so it should have a zeroed id
......
import argparse
import shutil
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.core.job import Job
import avocado_runner_vm
......@@ -20,8 +16,8 @@ class _FakeVM(avocado_runner_vm.VM):
def __init__(self): # pylint: disable=W0231
# don't call avocado_runner_vm.VM.__init__
self.snapshot = True
self.domain = mock.Mock()
self.domain.isActive = mock.Mock(return_value=True)
self.domain = unittest.mock.Mock()
self.domain.isActive = unittest.mock.Mock(return_value=True)
class VMTestRunnerSetup(unittest.TestCase):
......@@ -30,10 +26,10 @@ class VMTestRunnerSetup(unittest.TestCase):
def test_setup(self):
mock_vm = _FakeVM()
mock_vm.start = mock.Mock(return_value=True)
mock_vm.create_snapshot = mock.Mock()
mock_vm.stop = mock.Mock()
mock_vm.restore_snapshot = mock.Mock()
mock_vm.start = unittest.mock.Mock(return_value=True)
mock_vm.create_snapshot = unittest.mock.Mock()
mock_vm.stop = unittest.mock.Mock()
mock_vm.restore_snapshot = unittest.mock.Mock()
job_args = argparse.Namespace(test_result_total=1,
vm_domain='domain',
vm_username='username',
......@@ -54,8 +50,8 @@ class VMTestRunnerSetup(unittest.TestCase):
try:
job = Job(job_args)
job.setup()
with mock.patch('avocado_runner_vm.vm_connect',
return_value=mock_vm):
with unittest.mock.patch('avocado_runner_vm.vm_connect',
return_value=mock_vm):
# VMTestRunner()
runner = avocado_runner_vm.VMTestRunner(job, None)
runner.setup()
......
此差异已折叠。
......@@ -3,17 +3,10 @@
Sphinx==1.7.8
# inspektor (static and style checks)
pylint==1.9.3; python_version <= '2.7'
pylint==2.3.0; python_version >= '3.4'
astroid==2.2.0; python_version >= '3.4'
pylint==2.3.0
astroid==2.2.0
inspektor==0.5.2
# mock (some unittests use it)
mock>=2.0.0; python_version <= '2.7'
# six
six==1.11.0
# funcsigs
funcsigs>=0.4
......
......@@ -2,12 +2,7 @@ import logging
import os
import pkg_resources
import sys
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
#: The base directory for the avocado source tree
......@@ -22,19 +17,15 @@ def recent_mock():
'''
Checks if a recent and capable enough mock library is available
On Python 2.7, it requires at least mock version 2.0. On Python 3,
mock from the standard library is used, but Python 3.6 or later is
required.
On Python 3, mock from the standard library is used, but Python
3.6 or later is required.
Also, it assumes that on a future Python major version, functionality
won't regress.
'''
if sys.version_info[0] < 3:
major = int(mock.__version__.split('.')[0])
return major >= 2
elif sys.version_info[0] == 3:
if sys.version_info[0] == 3:
return sys.version_info[1] >= 6
return True
return sys.version_info[0] > 3
def python_module_available(module_name):
......
......@@ -12,11 +12,7 @@ import xml.dom.minidom
import zipfile
import unittest
import psutil
try:
from io import BytesIO
except ImportError:
from BytesIO import BytesIO
from io import BytesIO
try:
from lxml import etree
......@@ -24,9 +20,6 @@ try:
except ImportError:
SCHEMA_CAPABLE = False
from six import iteritems
from six.moves import xrange as range
from avocado.core import exit_codes
from avocado.utils import astring
from avocado.utils import genio
......@@ -185,7 +178,7 @@ class RunnerOperationTest(unittest.TestCase):
'data_dir': os.path.join(base_dir, 'data'),
'logs_dir': os.path.join(base_dir, 'logs')}
config = '[datadir.paths]\n'
for key, value in iteritems(mapping):
for key, value in mapping.items():
if not os.path.isdir(value):
os.mkdir(value)
config += "%s = %s\n" % (key, value)
......
......@@ -12,8 +12,6 @@ import tempfile
import time
import unittest
from six.moves import xrange as range
from avocado.utils import process
from avocado.utils import lv_utils
from avocado.utils import linux_modules
......
import re
import json
import unittest
from six.moves.urllib.error import URLError
from urllib.error import URLError
from avocado.utils import astring
from avocado.utils import download
......
......@@ -8,8 +8,6 @@ import tempfile
import time
import unittest
from six.moves import xrange as range
from avocado.utils.filelock import FileLock
from avocado.utils.stacktrace import prepare_exc_info
from avocado.utils import process
......
......@@ -5,8 +5,6 @@ import shutil
import sys
import random
from six.moves import xrange as range
from avocado.utils import archive
from avocado.utils import crypto
from avocado.utils import data_factory
......
import unittest
from six.moves import xrange as range
from avocado.utils import data_structures
......
import unittest
import os
import shutil
import tempfile
try:
from unittest import mock
except ImportError:
import mock
from six.moves import xrange as range
import unittest.mock
from avocado.core import settings
......@@ -51,7 +44,7 @@ class DataDirTest(unittest.TestCase):
stg = settings.Settings(self.config_file_path)
# Trick the module to think we're on a system wide install
stg.intree = False
with mock.patch('avocado.core.data_dir.settings.settings', stg):
with unittest.mock.patch('avocado.core.data_dir.settings.settings', stg):
from avocado.core import data_dir
self.assertFalse(data_dir.settings.settings.intree)
for key in self.mapping.keys():
......@@ -66,8 +59,8 @@ class DataDirTest(unittest.TestCase):
unique results.
"""
from avocado.core import data_dir
with mock.patch('avocado.core.data_dir.time.strftime',
return_value="date_would_go_here"):
with unittest.mock.patch('avocado.core.data_dir.time.strftime',
return_value="date_would_go_here"):
logdir = os.path.join(self.mapping['base_dir'], "foor", "bar", "baz")
path_prefix = os.path.join(logdir, "job-date_would_go_here-")
uid = "1234567890"*4
......@@ -96,7 +89,7 @@ class DataDirTest(unittest.TestCase):
(self.alt_mapping,
self.alt_config_file_path) = self._get_temporary_dirs_mapping_and_config()
stg = settings.Settings(self.alt_config_file_path)
with mock.patch('avocado.core.data_dir.settings.settings', stg):
with unittest.mock.patch('avocado.core.data_dir.settings.settings', stg):
from avocado.core import data_dir
for key in self.alt_mapping.keys():
data_dir_func = getattr(data_dir, 'get_%s' % key)
......
import re
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import distro
......@@ -67,8 +62,8 @@ class ProbeTest(unittest.TestCase):
CHECK_FILE_DISTRO_NAME = distro_name
my_probe = MyProbe()
with mock.patch('avocado.utils.distro.os.path.exists',
return_value=True) as mocked:
with unittest.mock.patch('avocado.utils.distro.os.path.exists',
return_value=True) as mocked:
probed_distro_name = my_probe.name_for_file()
mocked.assert_called_once_with(distro_file)
self.assertEqual(distro_name, probed_distro_name)
......
......@@ -2,12 +2,7 @@ import argparse
import os
import shutil
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.core import data_dir
from avocado.core import exceptions
......@@ -226,8 +221,8 @@ class JobTest(unittest.TestCase):
def test_job_no_base_logdir(self):
args = argparse.Namespace()
with mock.patch('avocado.core.job.data_dir.get_logs_dir',
return_value=self.tmpdir):
with unittest.mock.patch('avocado.core.job.data_dir.get_logs_dir',
return_value=self.tmpdir):
self.job = job.Job(args)
self.job.setup()
self.assertTrue(os.path.isdir(self.job.logdir))
......
......@@ -2,12 +2,7 @@ import os
import shutil
import stat
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.core import test
from avocado.core import loader
......@@ -476,7 +471,7 @@ class LoaderTest(unittest.TestCase):
def test_list_raising_exception(self):
simple_test = script.TemporaryScript('simpletest.py', PY_SIMPLE_TEST)
simple_test.save()
with mock.patch('avocado.core.loader.safeloader.find_avocado_tests') as _mock:
with unittest.mock.patch('avocado.core.loader.safeloader.find_avocado_tests') as _mock:
_mock.side_effect = BaseException()
tests = self.loader.discover(simple_test.path)
self.assertEqual(tests[0][1]["name"], simple_test.path)
......
import sys
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import path as utils_path
from avocado.core import output
......@@ -23,9 +19,9 @@ class TestStdOutput(unittest.TestCase):
def test_paginator_not_available(self):
"""Check that without paginator command we proceed without changes"""
std = output.StdOutput()
with mock.patch('avocado.utils.path.find_command',
side_effect=utils_path.CmdNotFoundError('just',
['mocking'])):
with unittest.mock.patch('avocado.utils.path.find_command',
side_effect=utils_path.CmdNotFoundError('just',
['mocking'])):
std.enable_paginator()
self.assertEqual(self.stdout, sys.stdout)
self.assertEqual(self.stderr, sys.stderr)
......
import os
import shutil
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.core import test, exceptions
from avocado.utils import astring, script
......@@ -123,7 +119,7 @@ class TestClassTestUnit(unittest.TestCase):
tst._record_reference('output', 'output.expected')
def test_all_dirs_exists_no_hang(self):
with mock.patch('os.path.exists', return_value=True):
with unittest.mock.patch('os.path.exists', return_value=True):
self.assertRaises(exceptions.TestSetupFail, self.DummyTest, "test",
test.TestID(1, "name"), base_logdir=self.tmpdir)
......
......@@ -2,13 +2,8 @@ import os
import shutil
import tempfile
import threading
import unittest # pylint: disable=C0411
try:
from unittest import mock
except ImportError:
import mock
from six.moves import http_client
import unittest.mock
import http.client
from avocado.utils import cloudinit
from avocado.utils import iso9660
......@@ -28,7 +23,7 @@ def has_iso_create_write():
class CloudInit(unittest.TestCase):
def test_iso_no_create_write(self):
with mock.patch('avocado.utils.iso9660.iso9660', return_value=None):
with unittest.mock.patch('avocado.utils.iso9660.iso9660', return_value=None):
self.assertRaises(RuntimeError, cloudinit.iso, os.devnull, "INSTANCE_ID")
......@@ -60,7 +55,7 @@ class PhoneHome(unittest.TestCase):
ADDRESS = '127.0.0.1'
def post_ignore_response(self, url):
conn = http_client.HTTPConnection(self.ADDRESS, self.port)
conn = http.client.HTTPConnection(self.ADDRESS, self.port)
conn.request('POST', url)
try:
conn.getresponse()
......
import io
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from .. import recent_mock
from avocado.utils import cpu
......@@ -14,9 +9,9 @@ class Cpu(unittest.TestCase):
@staticmethod
def _get_file_mock(content):
file_mock = mock.Mock()
file_mock.__enter__ = mock.Mock(return_value=io.BytesIO(content))
file_mock.__exit__ = mock.Mock()
file_mock = unittest.mock.Mock()
file_mock.__enter__ = unittest.mock.Mock(return_value=io.BytesIO(content))
file_mock.__exit__ = unittest.mock.Mock()
return file_mock
@unittest.skipUnless(recent_mock(),
......@@ -81,12 +76,13 @@ cpu MHz dynamic : 5504
cpu MHz static : 5504
"""
with mock.patch('avocado.utils.cpu.platform.machine', return_value='s390x'):
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(s390x)):
with unittest.mock.patch('avocado.utils.cpu.platform.machine',
return_value='s390x'):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(s390x)):
self.assertEqual(len(cpu.cpu_online_list()), 2)
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(s390x_2)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(s390x_2)):
self.assertEqual(len(cpu.cpu_online_list()), 4)
@unittest.skipUnless(recent_mock(),
......@@ -309,9 +305,10 @@ address sizes : 39 bits physical, 48 bits virtual
power management:
"""
with mock.patch('avocado.utils.cpu.platform.machine', return_value='x86_64'):
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(x86_64)):
with unittest.mock.patch('avocado.utils.cpu.platform.machine',
return_value='x86_64'):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(x86_64)):
self.assertEqual(len(cpu.cpu_online_list()), 8)
@unittest.skipUnless(recent_mock(),
......@@ -347,8 +344,8 @@ cache_alignment : 64
address sizes : 32 bits physical, 32 bits virtual
power management:
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "i386")
@unittest.skipUnless(recent_mock(),
......@@ -381,8 +378,8 @@ cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "x86_64")
@unittest.skipUnless(recent_mock(),
......@@ -399,8 +396,8 @@ model : 8247-21L
machine : PowerNV 8247-21L
firmware : OPAL v3
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "power8")
@unittest.skipUnless(recent_mock(),
......@@ -417,8 +414,8 @@ model : 8247-21L
machine : PowerNV 8247-21L
firmware : OPAL v3
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "power8")
@unittest.skipUnless(recent_mock(),
......@@ -435,8 +432,8 @@ model : 8375-42A
machine : PowerNV 8375-42A
firmware : OPAL
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "power9")
@unittest.skipUnless(recent_mock(),
......@@ -465,8 +462,8 @@ cpu number : 1
cpu MHz dynamic : 5504
cpu MHz static : 5504
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "s390")
@unittest.skipUnless(recent_mock(),
......@@ -485,8 +482,8 @@ Hardware : herring
Revision : 0034
Serial : 3534268a5e0700ec
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "arm")
@unittest.skipUnless(recent_mock(),
......@@ -501,8 +498,8 @@ CPU variant : 0x1
CPU part : 0x0a1
CPU revision : 1
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "aarch64")
@unittest.skipUnless(recent_mock(),
......@@ -513,35 +510,44 @@ isa : rv64imafdc
mmu : sv39
uarch : sifive,rocket0
"""
with mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=self._get_file_mock(cpu_output)):
self.assertEqual(cpu.get_cpu_arch(), "riscv")
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_get_cpuidle_state_off(self):
retval = {0: {0: 0}}
with mock.patch('avocado.utils.cpu.cpu_online_list', return_value=[0]):
with mock.patch('glob.glob', return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with mock.patch('avocado.utils.cpu.open', return_value=io.BytesIO(b'0')):
with unittest.mock.patch('avocado.utils.cpu.cpu_online_list',
return_value=[0]):
with unittest.mock.patch('glob.glob',
return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=io.BytesIO(b'0')):
self.assertEqual(cpu.get_cpuidle_state(), retval)
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_get_cpuidle_state_on(self):
retval = {0: {0: 1}}
with mock.patch('avocado.utils.cpu.cpu_online_list', return_value=[0]):
with mock.patch('glob.glob', return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with mock.patch('avocado.utils.cpu.open', return_value=io.BytesIO(b'1')):
with unittest.mock.patch('avocado.utils.cpu.cpu_online_list',
return_value=[0]):
with unittest.mock.patch('glob.glob',
return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=io.BytesIO(b'1')):
self.assertEqual(cpu.get_cpuidle_state(), retval)
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_set_cpuidle_state_default(self):
output = io.BytesIO()
with mock.patch('avocado.utils.cpu.cpu_online_list', return_value=[0]):
with mock.patch('glob.glob', return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with mock.patch('avocado.utils.cpu.open', return_value=output):
with unittest.mock.patch('avocado.utils.cpu.cpu_online_list',
return_value=[0]):
with unittest.mock.patch('glob.glob',
return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=output):
cpu.set_cpuidle_state()
self.assertEqual(output.getvalue(), b'1')
......@@ -549,9 +555,12 @@ uarch : sifive,rocket0
"mock library version cannot (easily) patch open()")
def test_set_cpuidle_state_withstateno(self):
output = io.BytesIO()
with mock.patch('avocado.utils.cpu.cpu_online_list', return_value=[0]):
with mock.patch('glob.glob', return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state2']):
with mock.patch('avocado.utils.cpu.open', return_value=output):
with unittest.mock.patch('avocado.utils.cpu.cpu_online_list',
return_value=[0]):
with unittest.mock.patch('glob.glob',
return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state2']):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=output):
cpu.set_cpuidle_state(disable=False, state_number='2')
self.assertEqual(output.getvalue(), b'0')
......@@ -559,9 +568,12 @@ uarch : sifive,rocket0
"mock library version cannot (easily) patch open()")
def test_set_cpuidle_state_withsetstate(self):
output = io.BytesIO()
with mock.patch('avocado.utils.cpu.cpu_online_list', return_value=[0, 2]):
with mock.patch('glob.glob', return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with mock.patch('avocado.utils.cpu.open', return_value=output):
with unittest.mock.patch('avocado.utils.cpu.cpu_online_list',
return_value=[0, 2]):
with unittest.mock.patch('glob.glob',
return_value=['/sys/devices/system/cpu/cpu0/cpuidle/state1']):
with unittest.mock.patch('avocado.utils.cpu.open',
return_value=output):
cpu.set_cpuidle_state(setstate={0: {0: 1}, 2: {0: 0}})
self.assertEqual(output.getvalue(), b'10')
......
import sys
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from .. import recent_mock
from avocado.utils import disk
......@@ -54,39 +49,39 @@ class Disk(unittest.TestCase):
mock_result = process.CmdResult(
command='lsblk --json',
stdout=b'{"blockdevices": []}')
with mock.patch('avocado.utils.disk.process.run',
return_value=mock_result):
with unittest.mock.patch('avocado.utils.disk.process.run',
return_value=mock_result):
self.assertEqual(disk.get_disks(), [])
def test_disks(self):
mock_result = process.CmdResult(
command='lsblk --json',
stdout=LSBLK_OUTPUT)
with mock.patch('avocado.utils.disk.process.run',
return_value=mock_result):
with unittest.mock.patch('avocado.utils.disk.process.run',
return_value=mock_result):
self.assertEqual(disk.get_disks(), ['/dev/vda'])
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_get_filesystems(self):
expected_fs = ['dax', 'bpf', 'pipefs', 'hugetlbfs', 'devpts', 'ext3']
open_mocked = mock.mock_open(read_data=PROC_FILESYSTEMS)
with mock.patch(self.builtin_open, open_mocked):
open_mocked = unittest.mock.mock_open(read_data=PROC_FILESYSTEMS)
with unittest.mock.patch(self.builtin_open, open_mocked):
self.assertEqual(sorted(expected_fs),
sorted(disk.get_available_filesystems()))
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_get_filesystem_type_default_root(self):
open_mocked = mock.mock_open(read_data=PROC_MOUNTS)
with mock.patch(self.builtin_open, open_mocked):
open_mocked = unittest.mock.mock_open(read_data=PROC_MOUNTS)
with unittest.mock.patch(self.builtin_open, open_mocked):
self.assertEqual('ext4', disk.get_filesystem_type())
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_get_filesystem_type(self):
open_mocked = mock.mock_open(read_data=PROC_MOUNTS)
with mock.patch(self.builtin_open, open_mocked):
open_mocked = unittest.mock.mock_open(read_data=PROC_MOUNTS)
with unittest.mock.patch(self.builtin_open, open_mocked):
self.assertEqual('ext2', disk.get_filesystem_type(mount_point='/home'))
......
......@@ -4,13 +4,7 @@ Verifies the avocado.utils.iso9660 functionality
import os
import shutil
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import iso9660, process
......@@ -27,16 +21,16 @@ class Capabilities(unittest.TestCase):
os.path.pardir, ".data",
"sample.iso"))
@mock.patch('avocado.utils.iso9660.has_pycdlib', return_value=True)
@unittest.mock.patch('avocado.utils.iso9660.has_pycdlib', return_value=True)
def test_capabilities_pycdlib(self, has_pycdlib_mocked):
instance = iso9660.iso9660(self.iso_path, ['read', 'create', 'write'])
self.assertIsInstance(instance, iso9660.ISO9660PyCDLib)
self.assertTrue(has_pycdlib_mocked.called)
@mock.patch('avocado.utils.iso9660.has_pycdlib', return_value=False)
@mock.patch('avocado.utils.iso9660.has_isoinfo', return_value=False)
@mock.patch('avocado.utils.iso9660.has_isoread', return_value=False)
@mock.patch('avocado.utils.iso9660.can_mount', return_value=False)
@unittest.mock.patch('avocado.utils.iso9660.has_pycdlib', return_value=False)
@unittest.mock.patch('avocado.utils.iso9660.has_isoinfo', return_value=False)
@unittest.mock.patch('avocado.utils.iso9660.has_isoread', return_value=False)
@unittest.mock.patch('avocado.utils.iso9660.can_mount', return_value=False)
def test_capabilities_nobackend(self, has_pycdlib_mocked, has_isoinfo_mocked,
has_isoread_mocked, can_mount_mocked):
self.assertIsNone(iso9660.iso9660(self.iso_path, ['read']))
......
import io
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from .. import recent_mock
from avocado.utils import linux_modules
......@@ -166,9 +161,9 @@ video 45056 2 thinkpad_acpi,i915, Live 0x0000000000000000
@staticmethod
def _get_file_mock(content):
file_mock = mock.Mock()
file_mock.__enter__ = mock.Mock(return_value=io.BytesIO(content))
file_mock.__exit__ = mock.Mock()
file_mock = unittest.mock.Mock()
file_mock.__enter__ = unittest.mock.Mock(return_value=io.BytesIO(content))
file_mock.__exit__ = unittest.mock.Mock()
return file_mock
def test_parse_lsmod(self):
......@@ -203,8 +198,8 @@ video 45056 2 thinkpad_acpi,i915, Live 0x0000000000000000
@unittest.skipUnless(recent_mock(),
"mock library version cannot (easily) patch open()")
def test_is_module_loaded(self):
with mock.patch('avocado.utils.linux_modules.open',
return_value=self._get_file_mock(self.PROC_MODULES_OUT)):
with unittest.mock.patch('avocado.utils.linux_modules.open',
return_value=self._get_file_mock(self.PROC_MODULES_OUT)):
self.assertTrue(linux_modules.module_is_loaded("rfcomm"))
self.assertFalse(linux_modules.module_is_loaded("unknown_module"))
......
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import memory
......@@ -14,9 +9,9 @@ class UtilsMemoryTest(unittest.TestCase):
file_values = [u"0\n", u"1-3", u"0-1,12-14\n"]
expected_values = [[0], [1, 2, 3], [0, 1, 12, 13, 14]]
for value, exp in zip(file_values, expected_values):
with mock.patch('os.path.exists', return_value=True):
with mock.patch('avocado.utils.genio.read_file',
return_value=value):
with unittest.mock.patch('os.path.exists', return_value=True):
with unittest.mock.patch('avocado.utils.genio.read_file',
return_value=value):
self.assertEqual(memory.numa_nodes_with_memory(), exp)
......@@ -27,7 +22,8 @@ BUDDY_INFO_RESPONSE = '\n'.join([
])
@mock.patch('avocado.utils.memory._get_buddy_info_content', return_value=BUDDY_INFO_RESPONSE)
@unittest.mock.patch('avocado.utils.memory._get_buddy_info_content',
return_value=BUDDY_INFO_RESPONSE)
class UtilsMemoryTestGetBuddyInfo(unittest.TestCase):
def test_get_buddy_info_simple_chunk_size(self, buddy_info_content_mocked):
......
import socket
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
try:
import netifaces
......@@ -19,7 +14,7 @@ class PortTrackerTest(unittest.TestCase):
def test_register_port(self):
tracker = network.PortTracker()
network.is_port_free = mock.MagicMock(return_value=True)
network.is_port_free = unittest.mock.MagicMock(return_value=True)
self.assertNotIn(22, tracker.retained_ports)
tracker.register_port(22)
network.is_port_free.assert_called_once_with(22, tracker.address)
......@@ -27,8 +22,8 @@ class PortTrackerTest(unittest.TestCase):
def test_release_port_does_not_poke_system(self):
tracker = network.PortTracker()
tracker.release_port = mock.MagicMock()
network.is_port_free = mock.MagicMock()
tracker.release_port = unittest.mock.MagicMock()
network.is_port_free = unittest.mock.MagicMock()
tracker.release_port(22)
tracker.release_port.assert_called_once_with(22)
network.is_port_free.assert_not_called()
......
......@@ -8,11 +8,7 @@ import os
import shutil
import sys
import tempfile
import unittest # pylint: disable=C0411
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import partition, process
from avocado.utils import path as utils_path
......@@ -133,10 +129,10 @@ class TestPartitionMkfsMount(Base):
proc_mounts = proc_mounts_file.read()
self.assertIn(self.mountpoint, proc_mounts)
proc = self.run_process_to_use_mnt()
with mock.patch('avocado.utils.partition.process.run',
side_effect=process.CmdError):
with mock.patch('avocado.utils.partition.process.system_output',
side_effect=OSError) as mocked_system_output:
with unittest.mock.patch('avocado.utils.partition.process.run',
side_effect=process.CmdError):
with unittest.mock.patch('avocado.utils.partition.process.system_output',
side_effect=OSError) as mocked_system_output:
self.assertRaises(partition.PartitionError, self.disk.unmount)
mocked_system_output.assert_called_with('lsof ' + self.mountpoint,
sudo=True)
......@@ -184,8 +180,8 @@ class TestMtabLock(unittest.TestCase):
def test_lock(self):
""" Check double-lock raises exception after 60s (in 0.1s) """
with partition.MtabLock():
with mock.patch('avocado.utils.filelock.time.time',
mock.MagicMock(side_effect=[1, 2, 62])):
with unittest.mock.patch('avocado.utils.filelock.time.time',
unittest.mock.MagicMock(side_effect=[1, 2, 62])):
self.assertRaises(partition.PartitionError,
partition.MtabLock().__enter__)
......
import os
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import path
......@@ -12,8 +7,8 @@ from avocado.utils import path
class Path(unittest.TestCase):
def test_check_readable_exists(self):
with mock.patch('avocado.utils.path.os.path.exists',
return_value=False) as mocked_exists:
with unittest.mock.patch('avocado.utils.path.os.path.exists',
return_value=False) as mocked_exists:
with self.assertRaises(OSError) as cm:
path.check_readable(os.devnull)
self.assertEqual('File "%s" does not exist' % os.devnull,
......@@ -21,8 +16,8 @@ class Path(unittest.TestCase):
mocked_exists.assert_called_with(os.devnull)
def test_check_readable_access(self):
with mock.patch('avocado.utils.path.os.access',
return_value=False) as mocked_access:
with unittest.mock.patch('avocado.utils.path.os.access',
return_value=False) as mocked_access:
with self.assertRaises(OSError) as cm:
path.check_readable(os.devnull)
self.assertEqual('File "%s" can not be read' % os.devnull,
......
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import pci
......@@ -15,15 +10,15 @@ class UtilsPciTest(unittest.TestCase):
file_values = ['S0001', 'S0001[', 'Slot2', 'SLOT1', 'Backplane USB', 'U78CB.001.WZS07CU-P1-C9-T1', 'PLX Slot1', 'Onboard USB', 'U78D5.001.CSS130E-P1-P2-P2-C1-T1']
expected_values = ['S0001', 'S0001', 'Slot2', 'SLOT1', 'Backplane USB', 'U78CB.001.WZS07CU-P1-C9', 'PLX Slot1', 'Onboard USB', 'U78D5.001.CSS130E-P1-P2-P2-C1']
for value, exp in zip(file_values, expected_values):
with mock.patch('os.path.isfile', return_value=True):
with mock.patch('avocado.utils.genio.read_file',
return_value=value):
with unittest.mock.patch('os.path.isfile', return_value=True):
with unittest.mock.patch('avocado.utils.genio.read_file',
return_value=value):
self.assertEqual(pci.get_slot_from_sysfs(pcid), exp)
def test_get_slot_from_sysfs_negative(self):
with mock.patch('os.path.isfile', return_value=True):
with mock.patch('avocado.utils.genio.read_file',
return_value='.....bad-value.....'):
with unittest.mock.patch('os.path.isfile', return_value=True):
with unittest.mock.patch('avocado.utils.genio.read_file',
return_value='.....bad-value.....'):
self.assertRaises(ValueError, pci.get_slot_from_sysfs,
'0002:01:00.1')
......
......@@ -16,12 +16,7 @@
# The full GNU General Public License is included in this distribution in
# the file called "COPYING".
import unittest
try:
from unittest import mock
except ImportError:
import mock
import unittest.mock
from avocado.utils import service
......@@ -35,14 +30,14 @@ class TestMultipleInstances(unittest.TestCase):
def test_different_runners(self):
# Call 'set_target' on first runner
runner1 = mock.Mock()
runner1 = unittest.mock.Mock()
runner1.return_value.stdout = 'systemd'
service1 = service.service_manager(run=runner1)
service1.set_target('foo_target')
self.assertEqual(runner1.call_args[0][0], # pylint: disable=E1136
'systemctl isolate foo_target')
# Call 'start' on second runner
runner2 = mock.Mock()
runner2 = unittest.mock.Mock()
runner2.return_value.stdout = 'init'
service2 = service.service_manager(run=runner2)
service2.start('foo_service')
......@@ -127,15 +122,17 @@ class TestSysVInit(unittest.TestCase):
class TestSpecificServiceManager(unittest.TestCase):
def setUp(self):
self.run_mock = mock.Mock()
self.run_mock = unittest.mock.Mock()
self.init_name = "init"
get_name_of_init_mock = mock.Mock(return_value="init")
get_name_of_init_mock = unittest.mock.Mock(return_value="init")
@mock.patch.object(service, "get_name_of_init", get_name_of_init_mock)
@unittest.mock.patch.object(service, "get_name_of_init",
get_name_of_init_mock)
def patch_service_command_generator():
return service._auto_create_specific_service_command_generator()
@mock.patch.object(service, "get_name_of_init", get_name_of_init_mock)
@unittest.mock.patch.object(service, "get_name_of_init",
get_name_of_init_mock)
def patch_service_result_parser():
return service._auto_create_specific_service_result_parser()
service_command_generator = patch_service_command_generator()
......@@ -180,7 +177,7 @@ class TestServiceManager(unittest.TestCase):
class TestSystemdServiceManager(TestServiceManager):
def setUp(self):
self.run_mock = mock.Mock()
self.run_mock = unittest.mock.Mock()
self.init_name = "systemd"
self.service_manager = \
(super(TestSystemdServiceManager, self)
......@@ -194,11 +191,12 @@ class TestSystemdServiceManager(TestServiceManager):
self.assertEqual(self.run_mock.call_args[0][0], cmd) # pylint: disable=E1136
def test_list(self):
list_result_mock = mock.Mock(exit_status=0,
stdout_text="sshd.service enabled\n"
"vsftpd.service disabled\n"
"systemd-sysctl.service static\n")
run_mock = mock.Mock(return_value=list_result_mock)
list_result_mock = unittest.mock.Mock(
exit_status=0,
stdout_text="sshd.service enabled\n"
"vsftpd.service disabled\n"
"systemd-sysctl.service static\n")
run_mock = unittest.mock.Mock(return_value=list_result_mock)
service_manager = \
(super(TestSystemdServiceManager, self)
.get_service_manager_from_init_and_run(self.init_name, run_mock))
......@@ -212,13 +210,13 @@ class TestSystemdServiceManager(TestServiceManager):
def test_set_default_runlevel(self):
runlevel = service.convert_sysv_runlevel(3)
mktemp_mock = mock.Mock(return_value="temp_filename")
symlink_mock = mock.Mock()
rename_mock = mock.Mock()
mktemp_mock = unittest.mock.Mock(return_value="temp_filename")
symlink_mock = unittest.mock.Mock()
rename_mock = unittest.mock.Mock()
@mock.patch.object(service, "mktemp", mktemp_mock)
@mock.patch("os.symlink", symlink_mock)
@mock.patch("os.rename", rename_mock)
@unittest.mock.patch.object(service, "mktemp", mktemp_mock)
@unittest.mock.patch("os.symlink", symlink_mock)
@unittest.mock.patch("os.rename", rename_mock)
def _():
self.service_manager.change_default_runlevel(runlevel)
self.assertTrue(mktemp_mock.called)
......@@ -243,7 +241,7 @@ class TestSystemdServiceManager(TestServiceManager):
class TestSysVInitServiceManager(TestServiceManager):
def setUp(self):
self.run_mock = mock.Mock()
self.run_mock = unittest.mock.Mock()
self.init_name = "init"
self.service_manager = \
super(TestSysVInitServiceManager,
......@@ -257,7 +255,7 @@ class TestSysVInitServiceManager(TestServiceManager):
self.assertEqual(self.run_mock.call_args[0][0], cmd) # pylint: disable=E1136
def test_list(self):
list_result_mock = mock.Mock(
list_result_mock = unittest.mock.Mock(
exit_status=0,
stdout_text="sshd 0:off 1:off "
"2:off 3:off 4:off 5:off 6:off\n"
......@@ -267,7 +265,7 @@ class TestSysVInitServiceManager(TestServiceManager):
" amanda: off\n"
" chargen-dgram: on\n")
run_mock = mock.Mock(return_value=list_result_mock)
run_mock = unittest.mock.Mock(return_value=list_result_mock)
service_manager = \
super(TestSysVInitServiceManager,
self).get_service_manager_from_init_and_run(self.init_name,
......
import unittest
try:
from unittest import mock
except ImportError:
import mock
from six.moves.urllib.error import HTTPError
import unittest.mock
from urllib.error import HTTPError
from avocado.utils import vmimage
......@@ -63,31 +58,31 @@ class ImageProviderBase(unittest.TestCase):
html = '<html><head><title>Test</title></head><body>%s</body></html>'
return html % ''.join(['<a href="%s/" />' % v for v in versions])
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_version(self, urlopen_mock):
html_fixture = self.get_html_with_versions([10, 11, 12])
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
base_image = vmimage.ImageProviderBase(version='[0-9]+', build=None, arch=None)
self.assertEqual(base_image.get_version(), 12)
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_version_with_float_versions(self, urlopen_mock):
html_fixture = self.get_html_with_versions([10.1, 10.3, 10.2])
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
base_image = vmimage.ImageProviderBase(version=r'[0-9]+\.[0-9]+', build=None, arch=None)
self.assertEqual(base_image.get_version(), 10.3)
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_version_with_string_versions(self, urlopen_mock):
html_fixture = self.get_html_with_versions(['abc', 'abcd', 'abcde'])
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
base_image = vmimage.ImageProviderBase(version=r'[\w]+', build=None, arch=None)
self.assertEqual(base_image.get_version(), 'abcde')
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_version_from_bad_url_open(self, urlopen_mock):
urlopen_mock.side_effect = HTTPError(None, None, None, None, None)
base_image = vmimage.ImageProviderBase(version='[0-9]+', build=None, arch=None)
......@@ -97,11 +92,11 @@ class ImageProviderBase(unittest.TestCase):
self.assertIn('Cannot open', exc.exception.args[0])
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_version_versions_not_found(self, urlopen_mock):
html_fixture = self.get_html_with_versions([])
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
base_image = vmimage.ImageProviderBase(version='[0-9]+', build=None, arch=None)
with self.assertRaises(vmimage.ImageProviderError) as exc:
......@@ -143,29 +138,29 @@ class OpenSUSEImageProvider(unittest.TestCase):
self.assertEqual(suse_provider.get_best_version(self.suse_available_versions),
suse_latest_version)
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_image_url(self, urlopen_mock):
image = 'openSUSE-Leap-15.0-OpenStack.x86_64-0.0.4-Buildlp150.12.30.qcow2'
html_fixture = self.get_html_with_image_link(image)
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
expected_image_url = self.base_images_url + image
suse_provider = vmimage.OpenSUSEImageProvider(arch='x86_64')
suse_provider.get_version = mock.Mock(return_value='15.0')
suse_provider.get_version = unittest.mock.Mock(return_value='15.0')
self.assertEqual(suse_provider.get_image_url(), expected_image_url)
@mock.patch('avocado.utils.vmimage.urlopen')
@unittest.mock.patch('avocado.utils.vmimage.urlopen')
def test_get_image_url_defining_build(self, urlopen_mock):
image = 'openSUSE-Leap-15.0-OpenStack.x86_64-1.1.1-Buildlp111.11.11.qcow2'
html_fixture = self.get_html_with_image_link(image)
urlread_mocked = mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = mock.Mock(read=urlread_mocked)
urlread_mocked = unittest.mock.Mock(return_value=html_fixture)
urlopen_mock.return_value = unittest.mock.Mock(read=urlread_mocked)
expected_image_url = self.base_images_url + image
suse_provider = vmimage.OpenSUSEImageProvider(build='1.1.1-Buildlp111.11.11',
arch='x86_64')
suse_provider.get_version = mock.Mock(return_value='15.0')
suse_provider.get_version = unittest.mock.Mock(return_value='15.0')
self.assertEqual(suse_provider.get_image_url(), expected_image_url)
......
......@@ -4,11 +4,7 @@ import shutil
import tempfile
import unittest
from xml.dom import minidom
try:
from io import BytesIO
except ImportError:
from BytesIO import BytesIO
from io import BytesIO
try:
from lxml import etree
......
......@@ -30,10 +30,7 @@ def get_long_description():
return req_contents
INSTALL_REQUIREMENTS = ['requests', 'stevedore>=0.14', 'six>=1.10.0', 'setuptools']
if sys.version_info[0] == 2:
INSTALL_REQUIREMENTS.append('enum34')
INSTALL_REQUIREMENTS = ['requests', 'stevedore>=0.14', 'setuptools']
if __name__ == '__main__':
# Force "make develop" inside the "readthedocs.org" environment
......@@ -55,8 +52,6 @@ if __name__ == '__main__':
"Operating System :: POSIX",
"Topic :: Software Development :: Quality Assurance",
"Topic :: Software Development :: Testing",
"Programming Language :: Python :: 2",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
......@@ -112,5 +107,5 @@ if __name__ == '__main__':
},
zip_safe=False,
test_suite='selftests',
python_requires='>=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*',
python_requires='>=3.4',
install_requires=INSTALL_REQUIREMENTS)
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册