Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use mock_import method in Nose

Best Python code snippet using nose

config_test.py

Source:config_test.py

1# -*- coding: utf-8 -*-2import copy3import datetime4import mock5import pytest6import elastalert.alerts7import elastalert.ruletypes8from elastalert.config import get_file_paths9from elastalert.config import load_configuration10from elastalert.config import load_modules11from elastalert.config import load_options12from elastalert.config import load_rules13from elastalert.util import EAException14test_config = {'rules_folder': 'test_folder',15               'run_every': {'minutes': 10},16               'buffer_time': {'minutes': 10},17               'es_host': 'elasticsearch.test',18               'es_port': 12345,19               'writeback_index': 'test_index'}20test_rule = {'es_host': 'test_host',21             'es_port': 12345,22             'name': 'testrule',23             'type': 'spike',24             'spike_height': 2,25             'spike_type': 'up',26             'timeframe': {'minutes': 10},27             'index': 'test_index',28             'query_key': 'testkey',29             'compare_key': 'comparekey',30             'filter': [{'term': {'key': 'value'}}],31             'alert': 'email',32             'use_count_query': True,33             'doc_type': 'blsh',34             'email': 'test@test.test',35             'aggregation': {'hours': 2},36             'include': ['comparekey', '@timestamp']}37test_args = mock.Mock()38test_args.config = 'test_config'39test_args.rule = None40def test_import_rules():41    test_rule_copy = copy.deepcopy(test_rule)42    test_rule_copy['type'] = 'testing.test.RuleType'43    with mock.patch('elastalert.config.yaml_loader') as mock_open:44        mock_open.return_value = test_rule_copy45        # Test that type is imported46        with mock.patch('__builtin__.__import__') as mock_import:47            mock_import.return_value = elastalert.ruletypes48            load_configuration('test_config', test_config)49        assert mock_import.call_args_list[0][0][0] == 'testing.test'50        assert mock_import.call_args_list[0][0][3] == ['RuleType']51        # Test that alerts are imported52        test_rule_copy = copy.deepcopy(test_rule)53        mock_open.return_value = test_rule_copy54        test_rule_copy['alert'] = 'testing2.test2.Alerter'55        with mock.patch('__builtin__.__import__') as mock_import:56            mock_import.return_value = elastalert.alerts57            load_configuration('test_config', test_config)58        assert mock_import.call_args_list[0][0][0] == 'testing2.test2'59        assert mock_import.call_args_list[0][0][3] == ['Alerter']60def test_load_inline_alert_rule():61    test_rule_copy = copy.deepcopy(test_rule)62    test_rule_copy['alert'] = [63        {64            'email': {65                'email': 'foo@bar.baz'66            }67        },68        {69            'email': {70                'email': 'baz@foo.bar'71            }72        }73    ]74    test_config_copy = copy.deepcopy(test_config)75    with mock.patch('elastalert.config.yaml_loader') as mock_open:76        mock_open.side_effect = [test_config_copy, test_rule_copy]77        load_modules(test_rule_copy)78        assert isinstance(test_rule_copy['alert'][0], elastalert.alerts.EmailAlerter)79        assert isinstance(test_rule_copy['alert'][1], elastalert.alerts.EmailAlerter)80        assert 'foo@bar.baz' in test_rule_copy['alert'][0].rule['email']81        assert 'baz@foo.bar' in test_rule_copy['alert'][1].rule['email']82def test_load_rules():83    test_rule_copy = copy.deepcopy(test_rule)84    test_config_copy = copy.deepcopy(test_config)85    with mock.patch('elastalert.config.yaml_loader') as mock_open:86        mock_open.side_effect = [test_config_copy, test_rule_copy]87        with mock.patch('os.listdir') as mock_ls:88            mock_ls.return_value = ['testrule.yaml']89            rules = load_rules(test_args)90            assert isinstance(rules['rules'][0]['type'], elastalert.ruletypes.RuleType)91            assert isinstance(rules['rules'][0]['alert'][0], elastalert.alerts.Alerter)92            assert isinstance(rules['rules'][0]['timeframe'], datetime.timedelta)93            assert isinstance(rules['run_every'], datetime.timedelta)94            for included_key in ['comparekey', 'testkey', '@timestamp']:95                assert included_key in rules['rules'][0]['include']96            # Assert include doesn't contain duplicates97            assert rules['rules'][0]['include'].count('@timestamp') == 198            assert rules['rules'][0]['include'].count('comparekey') == 199def test_load_default_host_port():100    test_rule_copy = copy.deepcopy(test_rule)101    test_rule_copy.pop('es_host')102    test_rule_copy.pop('es_port')103    test_config_copy = copy.deepcopy(test_config)104    with mock.patch('elastalert.config.yaml_loader') as mock_open:105        mock_open.side_effect = [test_config_copy, test_rule_copy]106        with mock.patch('os.listdir') as mock_ls:107            mock_ls.return_value = ['testrule.yaml']108            rules = load_rules(test_args)109            # Assert include doesn't contain duplicates110            assert rules['es_port'] == 12345111            assert rules['es_host'] == 'elasticsearch.test'112def test_compound_query_key():113    test_rule_copy = copy.deepcopy(test_rule)114    test_rule_copy.pop('use_count_query')115    test_rule_copy['query_key'] = ['field1', 'field2']116    load_options(test_rule_copy, test_config)117    assert 'field1' in test_rule_copy['include']118    assert 'field2' in test_rule_copy['include']119    assert test_rule_copy['query_key'] == 'field1,field2'120    assert test_rule_copy['compound_query_key'] == ['field1', 'field2']121def test_raises_on_missing_config():122    optional_keys = ('aggregation', 'use_count_query', 'query_key', 'compare_key', 'filter', 'include', 'es_host', 'es_port')123    test_rule_copy = copy.deepcopy(test_rule)124    for key in test_rule_copy.keys():125        test_rule_copy = copy.deepcopy(test_rule)126        test_config_copy = copy.deepcopy(test_config)127        test_rule_copy.pop(key)128        # Non required keys129        if key in optional_keys:130            continue131        with mock.patch('elastalert.config.yaml_loader') as mock_open:132            mock_open.side_effect = [test_config_copy, test_rule_copy]133            with mock.patch('os.listdir') as mock_ls:134                mock_ls.return_value = ['testrule.yaml']135                with pytest.raises(EAException):136                    load_rules(test_args)137def test_raises_on_bad_generate_kibana_filters():138    test_rule['generate_kibana_link'] = True139    bad_filters = [[{'not': {'terms': {'blah': 'blah'}}}],140                   [{'terms': {'blah': 'blah'}}],141                   [{'query': {'not_querystring': 'this:that'}}],142                   [{'query': {'wildcard': 'this*that'}}],143                   [{'blah': 'blah'}]]144    good_filters = [[{'term': {'field': 'value'}}],145                    [{'not': {'term': {'this': 'that'}}}],146                    [{'not': {'query': {'query_string': {'query': 'this:that'}}}}],147                    [{'query': {'query_string': {'query': 'this:that'}}}],148                    [{'range': {'blah': {'from': 'a', 'to': 'b'}}}],149                    [{'not': {'range': {'blah': {'from': 'a', 'to': 'b'}}}}]]150    # Test that all the good filters work, but fail with a bad filter added151    for good in good_filters:152        test_rule_copy = copy.deepcopy(test_rule)153        test_rule_copy['filter'] = good154        with mock.patch('elastalert.config.yaml_loader') as mock_open:155            mock_open.return_value = test_rule_copy156            load_configuration('blah', test_config)157            for bad in bad_filters:158                test_rule_copy['filter'] = good + bad159                with pytest.raises(EAException):160                    load_configuration('blah', test_config)161def test_get_file_paths_recursive():162    conf = {'scan_subdirectories': True, 'rules_folder': 'root'}163    walk_paths = (('root', ('folder_a', 'folder_b'), ('rule.yaml',)),164                  ('root/folder_a', (), ('a.yaml', 'ab.yaml')),165                  ('root/folder_b', (), ('b.yaml',)))166    with mock.patch('os.walk') as mock_walk:167        mock_walk.return_value = walk_paths168        paths = get_file_paths(conf)169    assert 'root/rule.yaml' in paths170    assert 'root/folder_a/a.yaml' in paths171    assert 'root/folder_a/ab.yaml' in paths172    assert 'root/folder_b/b.yaml' in paths173    assert len(paths) == 4174def test_get_file_paths():175    # Check for no subdirectory176    conf = {'scan_subdirectories': False, 'rules_folder': 'root'}177    files = ['badfile', 'a.yaml', 'b.yaml']178    with mock.patch('os.listdir') as mock_list:179        with mock.patch('os.path.isfile') as mock_path:180            mock_path.return_value = True181            mock_list.return_value = files182            paths = get_file_paths(conf)183    assert 'root/a.yaml' in paths184    assert 'root/b.yaml' in paths...

test_imports.py

Source:test_imports.py

...17            if mod in sys.modules:18                del sys.modules[mod]19    @patch.object(builtins, '__import__')20    def test_module_utils_basic_import_syslog(self, mock_import):21        def _mock_import(name, *args, **kwargs):22            if name == 'syslog':23                raise ImportError24            return realimport(name, *args, **kwargs)25        self.clear_modules(['syslog', 'ansible.module_utils.basic'])26        mod = builtins.__import__('ansible.module_utils.basic')27        self.assertTrue(mod.module_utils.basic.HAS_SYSLOG)28        self.clear_modules(['syslog', 'ansible.module_utils.basic'])29        mock_import.side_effect = _mock_import30        mod = builtins.__import__('ansible.module_utils.basic')31        self.assertFalse(mod.module_utils.basic.HAS_SYSLOG)32    @patch.object(builtins, '__import__')33    def test_module_utils_basic_import_selinux(self, mock_import):34        def _mock_import(name, *args, **kwargs):35            if name == 'ansible.module_utils.compat.selinux':36                raise ImportError37            return realimport(name, *args, **kwargs)38        try:39            self.clear_modules(['ansible.module_utils.compat.selinux', 'ansible.module_utils.basic'])40            mod = builtins.__import__('ansible.module_utils.basic')41            self.assertTrue(mod.module_utils.basic.HAVE_SELINUX)42        except ImportError:43            # no selinux on test system, so skip44            pass45        self.clear_modules(['ansible.module_utils.compat.selinux', 'ansible.module_utils.basic'])46        mock_import.side_effect = _mock_import47        mod = builtins.__import__('ansible.module_utils.basic')48        self.assertFalse(mod.module_utils.basic.HAVE_SELINUX)49    @patch.object(builtins, '__import__')50    def test_module_utils_basic_import_json(self, mock_import):51        def _mock_import(name, *args, **kwargs):52            if name == 'ansible.module_utils.common._json_compat':53                raise ImportError54            return realimport(name, *args, **kwargs)55        self.clear_modules(['json', 'ansible.module_utils.basic'])56        builtins.__import__('ansible.module_utils.basic')57        self.clear_modules(['json', 'ansible.module_utils.basic'])58        mock_import.side_effect = _mock_import59        with self.assertRaises(SystemExit):60            builtins.__import__('ansible.module_utils.basic')61    # FIXME: doesn't work yet62    # @patch.object(builtins, 'bytes')63    # def test_module_utils_basic_bytes(self, mock_bytes):64    #     mock_bytes.side_effect = NameError()65    #     from ansible.module_utils import basic66    @patch.object(builtins, '__import__')67    @unittest.skipIf(sys.version_info[0] >= 3, "literal_eval is available in every version of Python3")68    def test_module_utils_basic_import_literal_eval(self, mock_import):69        def _mock_import(name, *args, **kwargs):70            try:71                fromlist = kwargs.get('fromlist', args[2])72            except IndexError:73                fromlist = []74            if name == 'ast' and 'literal_eval' in fromlist:75                raise ImportError76            return realimport(name, *args, **kwargs)77        mock_import.side_effect = _mock_import78        self.clear_modules(['ast', 'ansible.module_utils.basic'])79        mod = builtins.__import__('ansible.module_utils.basic')80        self.assertEqual(mod.module_utils.basic.literal_eval("'1'"), "1")81        self.assertEqual(mod.module_utils.basic.literal_eval("1"), 1)82        self.assertEqual(mod.module_utils.basic.literal_eval("-1"), -1)83        self.assertEqual(mod.module_utils.basic.literal_eval("(1,2,3)"), (1, 2, 3))84        self.assertEqual(mod.module_utils.basic.literal_eval("[1]"), [1])85        self.assertEqual(mod.module_utils.basic.literal_eval("True"), True)86        self.assertEqual(mod.module_utils.basic.literal_eval("False"), False)87        self.assertEqual(mod.module_utils.basic.literal_eval("None"), None)88        # self.assertEqual(mod.module_utils.basic.literal_eval('{"a": 1}'), dict(a=1))89        self.assertRaises(ValueError, mod.module_utils.basic.literal_eval, "asdfasdfasdf")90    @patch.object(builtins, '__import__')91    def test_module_utils_basic_import_systemd_journal(self, mock_import):92        def _mock_import(name, *args, **kwargs):93            try:94                fromlist = kwargs.get('fromlist', args[2])95            except IndexError:96                fromlist = []97            if name == 'systemd' and 'journal' in fromlist:98                raise ImportError99            return realimport(name, *args, **kwargs)100        self.clear_modules(['systemd', 'ansible.module_utils.basic'])101        mod = builtins.__import__('ansible.module_utils.basic')102        self.assertTrue(mod.module_utils.basic.has_journal)103        self.clear_modules(['systemd', 'ansible.module_utils.basic'])104        mock_import.side_effect = _mock_import105        mod = builtins.__import__('ansible.module_utils.basic')106        self.assertFalse(mod.module_utils.basic.has_journal)

test_netshowlib.py

Source:test_netshowlib.py

1# http://pylint-messages.wikidot.com/all-codes2# pylint: disable=R09133# disable unused argument4# pylint: disable=W06135# disable docstring checking6# pylint: disable=C01117# disable checking no-self-use8# pylint: disable=R02019# pylint: disable=W021210# disable invalid name11# pylint: disable=C010312# pylint: disable=F040113# pylint: disable=E061114# pylint: disable=W061115from asserts import assert_equals, mod_args_generator16import netshowlib.netshowlib as nn17import mock18from mock import MagicMock19from netshowlib._version import get_version20def test_import_module():21    """ test import module """22    ospath = nn.import_module('os.path')23    assert_equals(ospath.exists('/etc/hosts'), True)24@mock.patch('netshowlib.netshowlib.pkg_resources.require')25@mock.patch('netshowlib.netshowlib.glob.glob')26def test_get_version(mock_glob, mock_require):27    require_mock = MagicMock()28    require_mock.version = 'netshowlibver'29    require_mock.location = '/var/me'30    mock_require.return_value = [require_mock]31    mock_glob.return_value = [32        '/var/me/linux']33    get_version()34    mock_require.assert_called_with('netshow-linux-lib')35    mock_glob.assert_called_with(36        '/var/me/../../../share/netshow-lib/providers/*')37@mock.patch('netshowlib.netshowlib.os.path.dirname')38@mock.patch('netshowlib.netshowlib.glob.glob')39@mock.patch('netshowlib.netshowlib.import_module')40@mock.patch('netshowlib.netshowlib.pkg_resources.require')41def test_provider_check(mock_pkg_requires, mock_import,42                        mock_glob,43                        mock_os_dirname):44    """ test os discovery """45    # return a directory with 3 OS types, each will return different priorities46    # choose the one with the highest priority47    mock_glob.return_value = ['path/providers/linux',48                              'path/providers/debian',49                              'path/providers/ubuntu']50    mock_linux = MagicMock()51    mock_linux.name_and_priority.return_value = {'Linux': 0}52    mock_debian = MagicMock()53    mock_debian.name_and_priority.return_value = {'Debian': 1}54    mock_debian = MagicMock()55    mock_debian.name_and_priority.return_value = {'Ubuntu': 2}56    mock_os_dirname.return_value = 'netshowlib'57    values = {58        'netshowlib.linux.provider_discovery': mock_linux,59        'netshowlib.debian.provider_discovery': mock_debian,60        'netshowlib.ubuntu.provider_discovery': mock_debian61    }62    mock_import.side_effect = mod_args_generator(values)63    mock_me = MagicMock()64    mock_me.location = '/me/and/my/loc'65    mock_pkg_requires.return_value = [mock_me]66    assert_equals(nn.provider_check(), 'ubuntu')67    mock_glob.assert_called_with('/me/and/my/loc/../../../share/netshow-lib/providers/*')68@mock.patch('netshowlib.netshowlib.provider_check')69@mock.patch('netshowlib.netshowlib.import_module')70def test_iface_discovery(mock_import, mock_provider_check):71    """ test iface discovery """72    # provider_check is none73    mock_provider_check.return_value = 'debian'74    mock_debian_iface = MagicMock()75    mock_debian_iface.iface.return_value = 'its a debian bridge'76    values = {'netshowlib.debian.iface': mock_debian_iface}77    mock_import.side_effect = mod_args_generator(values)78    assert_equals(nn.iface('eth1'), 'its a debian bridge')79    # if provider_check is not none80    mock_debian_iface = MagicMock()81    mock_debian_iface.iface.return_value = 'its a debian bridge'82    values['netshowlib.debian.iface'] = mock_debian_iface83    assert_equals(nn.iface('eth1', providername='debian'), 'its a debian bridge')84    # if cache is set provider_check is none85    mock_debian_iface.reset_mock()86    mock_debian_iface = MagicMock()87    mock_debian_iface.iface.return_value = 'its a debian bridge'88    mock_debian_cache = MagicMock()89    values = {'netshowlib.debian.iface': mock_debian_iface,90              'netshowlib.debian.cache': mock_debian_cache}91    mock_import.side_effect = mod_args_generator(values)92    all_cache = nn.feature_cache()93    assert_equals(nn.iface('eth1', cache=all_cache),94                  'its a debian bridge')95    # confirm syntax for iface_type accepts cache96    mock_debian_iface.iface.assert_called_with('eth1', cache=all_cache)97@mock.patch('netshowlib.netshowlib.provider_check')98@mock.patch('netshowlib.netshowlib.import_module')99def test_system_discovery(mock_import, mock_provider_check):100    mock_provider_check.return_value = 'debian'101    mock_debian_system = MagicMock()102    mock_debian_system.SystemSummary.return_value = 'debian system summary'103    mock_import.return_value = mock_debian_system104    assert_equals(nn.system_summary(), 'debian system summary')105    mock_import.assert_called_with('netshowlib.debian.system_summary')106@mock.patch('netshowlib.netshowlib.provider_check')107@mock.patch('netshowlib.netshowlib.import_module')108def test_portlist(mock_import, mock_provider_check):109    mock_provider_check.return_value = 'debian'110    mock_debian_iface = MagicMock()111    mock_debian_iface.portname_list.return_value = ['eth22', 'eth33']112    values = {113        'netshowlib.debian.iface': mock_debian_iface114    }115    mock_import.side_effect = mod_args_generator(values)...

test_conf.py

Source:test_conf.py

1# Copyright 2016 HPCC-ICSE Ltd.2#3# Licensed under the Apache License, Version 2.0 (the "License"); you may4# not use this file except in compliance with the License. You may obtain5# a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12# License for the specific language governing permissions and limitations13# under the License.14import collections15import mock16from oslo_config import cfg17import six18import calplus.conf19from calplus.conf import opts20from calplus.tests import base21CONF = calplus.conf.CONF22class ConfTestCase(base.TestCase):23    def test_list_opts(self):24        for group, opt_list in opts.list_opts():25            if isinstance(group, six.string_types):26                self.assertEqual(group, 'DEFAULT')27            else:28                self.assertIsInstance(group, cfg.OptGroup)29            for opt in opt_list:30                self.assertIsInstance(opt, cfg.Opt)31    def test_list_module_name_invalid_mods(self):32        with mock.patch('pkgutil.iter_modules') as mock_mods:33            mock_mods.return_value = [(None, 'foo', True),34                                      (None, 'opts', False)]35            self.assertEqual([], opts._list_module_names())36    def test_list_module_name_valid_mods(self):37        with mock.patch('pkgutil.iter_modules') as mock_mods:38            mock_mods.return_value = [(None, 'foo', False)]39            self.assertEqual(['foo'], opts._list_module_names())40    def test_import_mods_no_func(self):41        modules = ['foo', 'bar']42        with mock.patch('importlib.import_module') as mock_import:43            mock_import.return_value = mock.sentinel.mods44            self.assertRaises(AttributeError, opts._import_modules, modules)45            mock_import.assert_called_once_with('calplus.conf.foo')46    def test_import_mods_valid_func(self):47        modules = ['foo', 'bar']48        with mock.patch('importlib.import_module') as mock_import:49            mock_mod = mock.MagicMock()50            mock_import.return_value = mock_mod51            self.assertEqual([mock_mod, mock_mod],52                             opts._import_modules(modules))53            mock_import.assert_has_calls([mock.call('calplus.conf.foo'),54                                          mock.call('calplus.conf.bar')])55    def test_append_config(self):56        opt = collections.defaultdict(list)57        mock_module = mock.MagicMock()58        mock_conf = mock.MagicMock()59        mock_module.list_opts.return_value = mock_conf60        mock_conf.items.return_value = [('foo', 'bar')]61        opts._append_config_options([mock_module], opt)62        self.assertEqual({'foo': ['b', 'a', 'r']}, opt)63    def test_load_config_file_to_realize_all_driver(self):64        CONF(['--config-file',65              'calplus/tests/fake_config_file.conf'])66        # TODO: Maybe we need remove example group,67        # such as: openstack and amazon68        # ensure all driver groups have been registered69        sections = CONF.list_all_sections()70        for section in sections:71            CONF.register_group(cfg.OptGroup(section))72        # ensure all of enable drivers configured exact opts73        enable_drivers = CONF.providers.enable_drivers74        for driver in enable_drivers.keys():75            if enable_drivers.get(driver) == 'openstack':76                CONF.register_opts(77                    calplus.conf.providers.openstack_opts, driver)78            elif enable_drivers.get(driver) == 'amazon':79                CONF.register_opts(80                    calplus.conf.providers.amazon_opts, driver)81            else:82                continue83        self.assertEqual(CONF.openstack1['driver_name'], 'HUST')84        self.assertEqual(CONF.openstack2['driver_name'], 'SOICT')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.