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')...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!