archive
/
ansible
mirror of https://github.com/ansible/ansible.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

test: update tests (#83686)

Browse Source 
* Remove commented code
* Enable disabled tests
* Formatting

Signed-off-by: Abhijeet Kasurde <akasurde@redhat.com>
pull/83673/head
Abhijeet Kasurde 1 year ago committed by GitHub
parent 3daf01e270
commit c6d5be5cac
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 127 additions and 132 deletions
Show Stats Download Patch File Download Diff File

40
test/units/cli/test_galaxy.py
Unescape Escape View File

@ -114,13 +114,13 @@ class TestGalaxy(unittest.TestCase):
def test_init(self): def test_init(self):
galaxy_cli = GalaxyCLI(args=self.default_args) galaxy_cli = GalaxyCLI(args=self.default_args)
self.assertTrue(isinstance(galaxy_cli, GalaxyCLI)) assert isinstance(galaxy_cli, GalaxyCLI)
def test_display_min(self): def test_display_min(self):
gc = GalaxyCLI(args=self.default_args) gc = GalaxyCLI(args=self.default_args)
role_info = {'name': 'some_role_name'} role_info = {'name': 'some_role_name'}
display_result = gc._display_role_info(role_info) display_result = gc._display_role_info(role_info)
self.assertTrue(display_result.find('some_role_name') > -1) assert display_result.find('some_role_name') > -1
def test_display_galaxy_info(self): def test_display_galaxy_info(self):
gc = GalaxyCLI(args=self.default_args) gc = GalaxyCLI(args=self.default_args)
@ -139,7 +139,7 @@ class TestGalaxy(unittest.TestCase):
# testing # testing
self.assertIsInstance(gc.galaxy, ansible.galaxy.Galaxy) self.assertIsInstance(gc.galaxy, ansible.galaxy.Galaxy)
self.assertEqual(mock_run.call_count, 1) self.assertEqual(mock_run.call_count, 1)
self.assertTrue(isinstance(gc.api, ansible.galaxy.api.GalaxyAPI)) assert isinstance(gc.api, ansible.galaxy.api.GalaxyAPI)
def test_execute_remove(self): def test_execute_remove(self):
# installing role # installing role
@ -200,32 +200,32 @@ class TestGalaxy(unittest.TestCase):
''' testing the options parser when the action 'import' is given ''' ''' testing the options parser when the action 'import' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "import", "foo", "bar"]) gc = GalaxyCLI(args=["ansible-galaxy", "import", "foo", "bar"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['wait'], True) assert context.CLIARGS['wait']
self.assertEqual(context.CLIARGS['reference'], None) assert context.CLIARGS['reference'] is None
self.assertEqual(context.CLIARGS['check_status'], False) assert not context.CLIARGS['check_status']
self.assertEqual(context.CLIARGS['verbosity'], 0) assert context.CLIARGS['verbosity'] == 0
def test_parse_info(self): def test_parse_info(self):
''' testing the options parser when the action 'info' is given ''' ''' testing the options parser when the action 'info' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "info", "foo", "bar"]) gc = GalaxyCLI(args=["ansible-galaxy", "info", "foo", "bar"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['offline'], False) assert not context.CLIARGS['offline']
def test_parse_init(self): def test_parse_init(self):
''' testing the options parser when the action 'init' is given ''' ''' testing the options parser when the action 'init' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "init", "foo"]) gc = GalaxyCLI(args=["ansible-galaxy", "init", "foo"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['offline'], False) assert not context.CLIARGS['offline']
self.assertEqual(context.CLIARGS['force'], False) assert not context.CLIARGS['force']
def test_parse_install(self): def test_parse_install(self):
''' testing the options parser when the action 'install' is given ''' ''' testing the options parser when the action 'install' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "install"]) gc = GalaxyCLI(args=["ansible-galaxy", "install"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['ignore_errors'], False) assert not context.CLIARGS['ignore_errors']
self.assertEqual(context.CLIARGS['no_deps'], False) assert not context.CLIARGS['no_deps']
self.assertEqual(context.CLIARGS['requirements'], None) assert context.CLIARGS['requirements'] is None
self.assertEqual(context.CLIARGS['force'], False) assert not context.CLIARGS['force']
def test_parse_list(self): def test_parse_list(self):
''' testing the options parser when the action 'list' is given ''' ''' testing the options parser when the action 'list' is given '''
@ -243,17 +243,17 @@ class TestGalaxy(unittest.TestCase):
''' testing the options parswer when the action 'search' is given ''' ''' testing the options parswer when the action 'search' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "search"]) gc = GalaxyCLI(args=["ansible-galaxy", "search"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['platforms'], None) assert context.CLIARGS['platforms'] is None
self.assertEqual(context.CLIARGS['galaxy_tags'], None) assert context.CLIARGS['galaxy_tags'] is None
self.assertEqual(context.CLIARGS['author'], None) assert context.CLIARGS['author'] is None
def test_parse_setup(self): def test_parse_setup(self):
''' testing the options parser when the action 'setup' is given ''' ''' testing the options parser when the action 'setup' is given '''
gc = GalaxyCLI(args=["ansible-galaxy", "setup", "source", "github_user", "github_repo", "secret"]) gc = GalaxyCLI(args=["ansible-galaxy", "setup", "source", "github_user", "github_repo", "secret"])
gc.parse() gc.parse()
self.assertEqual(context.CLIARGS['verbosity'], 0) assert context.CLIARGS['verbosity'] == 0
self.assertEqual(context.CLIARGS['remove_id'], None) assert context.CLIARGS['remove_id'] is None
self.assertEqual(context.CLIARGS['setup_list'], False) assert not context.CLIARGS['setup_list']
class ValidRoleTests(object): class ValidRoleTests(object):

28
test/units/module_utils/basic/test_filesystem.py
Unescape Escape View File

@ -62,13 +62,13 @@ class TestOtherFilesystem(ModuleTestCase):
argument_spec=dict(), argument_spec=dict(),
) )
self.assertEqual(am.set_owner_if_different('/path/to/file', None, True), True) assert am.set_owner_if_different('/path/to/file', None, True)
self.assertEqual(am.set_owner_if_different('/path/to/file', None, False), False) assert not am.set_owner_if_different('/path/to/file', None, False)
am.user_and_group = MagicMock(return_value=(500, 500)) am.user_and_group = MagicMock(return_value=(500, 500))
with patch('os.lchown', return_value=None) as m: with patch('os.lchown', return_value=None) as m:
self.assertEqual(am.set_owner_if_different('/path/to/file', 0, False), True) assert am.set_owner_if_different('/path/to/file', 0, False)
m.assert_called_with(b'/path/to/file', 0, -1) m.assert_called_with(b'/path/to/file', 0, -1)
def _mock_getpwnam(*args, **kwargs): def _mock_getpwnam(*args, **kwargs):
@ -78,7 +78,7 @@ class TestOtherFilesystem(ModuleTestCase):
m.reset_mock() m.reset_mock()
with patch('pwd.getpwnam', side_effect=_mock_getpwnam): with patch('pwd.getpwnam', side_effect=_mock_getpwnam):
self.assertEqual(am.set_owner_if_different('/path/to/file', 'root', False), True) assert am.set_owner_if_different('/path/to/file', 'root', False)
m.assert_called_with(b'/path/to/file', 0, -1) m.assert_called_with(b'/path/to/file', 0, -1)
with patch('pwd.getpwnam', side_effect=KeyError): with patch('pwd.getpwnam', side_effect=KeyError):
@ -86,8 +86,8 @@ class TestOtherFilesystem(ModuleTestCase):
m.reset_mock() m.reset_mock()
am.check_mode = True am.check_mode = True
self.assertEqual(am.set_owner_if_different('/path/to/file', 0, False), True) assert am.set_owner_if_different('/path/to/file', 0, False)
self.assertEqual(m.called, False) assert not m.called
am.check_mode = False am.check_mode = False
with patch('os.lchown', side_effect=OSError) as m: with patch('os.lchown', side_effect=OSError) as m:
@ -101,13 +101,13 @@ class TestOtherFilesystem(ModuleTestCase):
argument_spec=dict(), argument_spec=dict(),
) )
self.assertEqual(am.set_group_if_different('/path/to/file', None, True), True) assert am.set_group_if_different('/path/to/file', None, True)
self.assertEqual(am.set_group_if_different('/path/to/file', None, False), False) assert not am.set_group_if_different('/path/to/file', None, False)
am.user_and_group = MagicMock(return_value=(500, 500)) am.user_and_group = MagicMock(return_value=(500, 500))
with patch('os.lchown', return_value=None) as m: with patch('os.lchown', return_value=None) as m:
self.assertEqual(am.set_group_if_different('/path/to/file', 0, False), True) assert am.set_group_if_different('/path/to/file', 0, False)
m.assert_called_with(b'/path/to/file', -1, 0) m.assert_called_with(b'/path/to/file', -1, 0)
def _mock_getgrnam(*args, **kwargs): def _mock_getgrnam(*args, **kwargs):
@ -117,7 +117,7 @@ class TestOtherFilesystem(ModuleTestCase):
m.reset_mock() m.reset_mock()
with patch('grp.getgrnam', side_effect=_mock_getgrnam): with patch('grp.getgrnam', side_effect=_mock_getgrnam):
self.assertEqual(am.set_group_if_different('/path/to/file', 'root', False), True) assert am.set_group_if_different('/path/to/file', 'root', False)
m.assert_called_with(b'/path/to/file', -1, 0) m.assert_called_with(b'/path/to/file', -1, 0)
with patch('grp.getgrnam', side_effect=KeyError): with patch('grp.getgrnam', side_effect=KeyError):
@ -125,8 +125,8 @@ class TestOtherFilesystem(ModuleTestCase):
m.reset_mock() m.reset_mock()
am.check_mode = True am.check_mode = True
self.assertEqual(am.set_group_if_different('/path/to/file', 0, False), True) assert am.set_group_if_different('/path/to/file', 0, False)
self.assertEqual(m.called, False) assert not m.called
am.check_mode = False am.check_mode = False
with patch('os.lchown', side_effect=OSError) as m: with patch('os.lchown', side_effect=OSError) as m:
@ -155,5 +155,5 @@ class TestOtherFilesystem(ModuleTestCase):
'attributes': None, 'attributes': None,
} }
self.assertEqual(am.set_directory_attributes_if_different(file_args, True), True) assert am.set_directory_attributes_if_different(file_args, True)
self.assertEqual(am.set_directory_attributes_if_different(file_args, False), False) assert not am.set_directory_attributes_if_different(file_args, False)

40
test/units/module_utils/facts/test_collector.py
Unescape Escape View File

@ -54,7 +54,7 @@ class TestFindCollectorsForPlatform(unittest.TestCase):
class TestSelectCollectorNames(unittest.TestCase): class TestSelectCollectorNames(unittest.TestCase):
def _assert_equal_detail(self, obj1, obj2, msg=None): def _assert_equal_detail(self, obj1, obj2):
msg = 'objects are not equal\n%s\n\n!=\n\n%s' % (pprint.pformat(obj1), pprint.pformat(obj2)) msg = 'objects are not equal\n%s\n\n!=\n\n%s' % (pprint.pformat(obj1), pprint.pformat(obj2))
return self.assertEqual(obj1, obj2, msg) return self.assertEqual(obj1, obj2, msg)
@ -92,12 +92,8 @@ class TestSelectCollectorNames(unittest.TestCase):
res = collector.select_collector_classes(ordered_collector_names, res = collector.select_collector_classes(ordered_collector_names,
all_fact_subsets) all_fact_subsets)
self.assertTrue(res.index(default_collectors.ServiceMgrFactCollector) > assert res.index(default_collectors.ServiceMgrFactCollector) > res.index(default_collectors.DistributionFactCollector)
res.index(default_collectors.DistributionFactCollector), assert res.index(default_collectors.ServiceMgrFactCollector) > res.index(default_collectors.PlatformFactCollector)
res)
self.assertTrue(res.index(default_collectors.ServiceMgrFactCollector) >
res.index(default_collectors.PlatformFactCollector),
res)
def _all_fact_subsets(self, data=None): def _all_fact_subsets(self, data=None):
all_fact_subsets = defaultdict(list) all_fact_subsets = defaultdict(list)
@ -278,7 +274,6 @@ class TestFindUnresolvedRequires(unittest.TestCase):
'network': [default_collectors.LinuxNetworkCollector], 'network': [default_collectors.LinuxNetworkCollector],
'virtual': [default_collectors.LinuxVirtualCollector]} 'virtual': [default_collectors.LinuxVirtualCollector]}
res = collector.find_unresolved_requires(names, all_fact_subsets) res = collector.find_unresolved_requires(names, all_fact_subsets)
# pprint.pprint(res)
self.assertIsInstance(res, set) self.assertIsInstance(res, set)
self.assertEqual(res, set(['platform', 'distribution'])) self.assertEqual(res, set(['platform', 'distribution']))
@ -291,7 +286,6 @@ class TestFindUnresolvedRequires(unittest.TestCase):
'platform': [default_collectors.PlatformFactCollector], 'platform': [default_collectors.PlatformFactCollector],
'virtual': [default_collectors.LinuxVirtualCollector]} 'virtual': [default_collectors.LinuxVirtualCollector]}
res = collector.find_unresolved_requires(names, all_fact_subsets) res = collector.find_unresolved_requires(names, all_fact_subsets)
# pprint.pprint(res)
self.assertIsInstance(res, set) self.assertIsInstance(res, set)
self.assertEqual(res, set()) self.assertEqual(res, set())
@ -305,7 +299,6 @@ class TestBuildDepData(unittest.TestCase):
'virtual': [default_collectors.LinuxVirtualCollector]} 'virtual': [default_collectors.LinuxVirtualCollector]}
res = collector.build_dep_data(names, all_fact_subsets) res = collector.build_dep_data(names, all_fact_subsets)
# pprint.pprint(dict(res))
self.assertIsInstance(res, defaultdict) self.assertIsInstance(res, defaultdict)
self.assertEqual(dict(res), self.assertEqual(dict(res),
{'network': set(['platform', 'distribution']), {'network': set(['platform', 'distribution']),
@ -332,8 +325,7 @@ class TestSolveDeps(unittest.TestCase):
'virtual': [default_collectors.LinuxVirtualCollector], 'virtual': [default_collectors.LinuxVirtualCollector],
'platform': [default_collectors.PlatformFactCollector], 'platform': [default_collectors.PlatformFactCollector],
'distribution': [default_collectors.DistributionFactCollector]} 'distribution': [default_collectors.DistributionFactCollector]}
res = collector.resolve_requires(unresolved, all_fact_subsets) collector.resolve_requires(unresolved, all_fact_subsets)
res = collector._solve_deps(unresolved, all_fact_subsets) res = collector._solve_deps(unresolved, all_fact_subsets)
self.assertIsInstance(res, set) self.assertIsInstance(res, set)
@ -379,13 +371,12 @@ class TestTsort(unittest.TestCase):
'network_stuff': set(['network'])} 'network_stuff': set(['network'])}
res = collector.tsort(dep_map) res = collector.tsort(dep_map)
# pprint.pprint(res)
self.assertIsInstance(res, list) self.assertIsInstance(res, list)
names = [x[0] for x in res] names = [x[0] for x in res]
self.assertTrue(names.index('network_stuff') > names.index('network')) assert names.index('network_stuff') > names.index('network')
self.assertTrue(names.index('platform') > names.index('what_platform_wants')) assert names.index('platform') > names.index('what_platform_wants')
self.assertTrue(names.index('network') > names.index('platform')) assert names.index('network') > names.index('platform')
def test_cycles(self): def test_cycles(self):
dep_map = {'leaf1': set(), dep_map = {'leaf1': set(),
@ -448,9 +439,9 @@ class TestTsort(unittest.TestCase):
self.assertIsInstance(res, list) self.assertIsInstance(res, list)
names = [x[0] for x in res] names = [x[0] for x in res]
self.assertEqual(set(names), set(dep_map.keys())) self.assertEqual(set(names), set(dep_map.keys()))
self.assertTrue(names.index('leaf1') < names.index('leaf2')) assert names.index('leaf1') < names.index('leaf2')
for leaf in ('leaf2', 'leaf3', 'leaf4', 'leaf5'): for leaf in ('leaf2', 'leaf3', 'leaf4', 'leaf5'):
self.assertTrue(names.index('leaf1') < names.index(leaf)) assert names.index('leaf1') < names.index(leaf)
class TestCollectorClassesFromGatherSubset(unittest.TestCase): class TestCollectorClassesFromGatherSubset(unittest.TestCase):
@ -494,8 +485,7 @@ class TestCollectorClassesFromGatherSubset(unittest.TestCase):
self.assertIn(default_collectors.PlatformFactCollector, res) self.assertIn(default_collectors.PlatformFactCollector, res)
self.assertIn(default_collectors.LinuxHardwareCollector, res) self.assertIn(default_collectors.LinuxHardwareCollector, res)
self.assertTrue(res.index(default_collectors.LinuxHardwareCollector) > assert res.index(default_collectors.LinuxHardwareCollector) > res.index(default_collectors.PlatformFactCollector)
res.index(default_collectors.PlatformFactCollector))
def test_network(self): def test_network(self):
res = self._classes(all_collector_classes=default_collectors.collectors, res = self._classes(all_collector_classes=default_collectors.collectors,
@ -505,14 +495,8 @@ class TestCollectorClassesFromGatherSubset(unittest.TestCase):
self.assertIn(default_collectors.PlatformFactCollector, res) self.assertIn(default_collectors.PlatformFactCollector, res)
self.assertIn(default_collectors.LinuxNetworkCollector, res) self.assertIn(default_collectors.LinuxNetworkCollector, res)
self.assertTrue(res.index(default_collectors.LinuxNetworkCollector) > assert res.index(default_collectors.LinuxNetworkCollector) > res.index(default_collectors.PlatformFactCollector)
res.index(default_collectors.PlatformFactCollector)) assert res.index(default_collectors.LinuxNetworkCollector) > res.index(default_collectors.DistributionFactCollector)
self.assertTrue(res.index(default_collectors.LinuxNetworkCollector) >
res.index(default_collectors.DistributionFactCollector))
# self.assertEqual(set(res, [default_collectors.DistributionFactCollector,
# default_collectors.PlatformFactCollector,
# default_collectors.LinuxNetworkCollector])
def test_env(self): def test_env(self):
res = self._classes(all_collector_classes=default_collectors.collectors, res = self._classes(all_collector_classes=default_collectors.collectors,

42
test/units/module_utils/urls/test_fetch_url.py
Unescape Escape View File

@ -16,6 +16,8 @@ from ansible.module_utils.urls import fetch_url, ConnectionError
import pytest import pytest
from unittest.mock import MagicMock from unittest.mock import MagicMock
BASE_URL = 'https://ansible.com/'
class AnsibleModuleExit(Exception): class AnsibleModuleExit(Exception):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
@ -54,11 +56,11 @@ class FakeAnsibleModule:
def test_fetch_url(open_url_mock, fake_ansible_module): def test_fetch_url(open_url_mock, fake_ansible_module):
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
dummy, kwargs = open_url_mock.call_args dummy, kwargs = open_url_mock.call_args
open_url_mock.assert_called_once_with('http://ansible.com/', client_cert=None, client_key=None, cookies=kwargs['cookies'], data=None, open_url_mock.assert_called_once_with(BASE_URL, client_cert=None, client_key=None, cookies=kwargs['cookies'], data=None,
follow_redirects='urllib2', force=False, force_basic_auth='', headers=None, follow_redirects='urllib2', force=False, force_basic_auth='', headers=None,
http_agent='ansible-httpget', last_mod_time=None, method=None, timeout=10, url_password='', url_username='', http_agent='ansible-httpget', last_mod_time=None, method=None, timeout=10, url_password='', url_username='',
use_proxy=True, validate_certs=True, use_gssapi=False, unix_socket=None, ca_path=None, unredirected_headers=None, use_proxy=True, validate_certs=True, use_gssapi=False, unix_socket=None, ca_path=None, unredirected_headers=None,
@ -77,11 +79,11 @@ def test_fetch_url_params(open_url_mock, fake_ansible_module):
'client_key': 'client.key', 'client_key': 'client.key',
} }
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
dummy, kwargs = open_url_mock.call_args dummy, kwargs = open_url_mock.call_args
open_url_mock.assert_called_once_with('http://ansible.com/', client_cert='client.pem', client_key='client.key', cookies=kwargs['cookies'], data=None, open_url_mock.assert_called_once_with(BASE_URL, client_cert='client.pem', client_key='client.key', cookies=kwargs['cookies'], data=None,
follow_redirects='all', force=False, force_basic_auth=True, headers=None, follow_redirects='all', force=False, force_basic_auth=True, headers=None,
http_agent='ansible-test', last_mod_time=None, method=None, timeout=10, url_password='passwd', url_username='user', http_agent='ansible-test', last_mod_time=None, method=None, timeout=10, url_password='passwd', url_username='user',
use_proxy=True, validate_certs=False, use_gssapi=False, unix_socket=None, ca_path=None, unredirected_headers=None, use_proxy=True, validate_certs=False, use_gssapi=False, unix_socket=None, ca_path=None, unredirected_headers=None,
@ -121,7 +123,7 @@ def test_fetch_url_cookies(mocker, fake_ansible_module):
mocker = mocker.patch('ansible.module_utils.urls.open_url', new=make_cookies) mocker = mocker.patch('ansible.module_utils.urls.open_url', new=make_cookies)
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
assert info['cookies'] == {'Baz': 'qux', 'Foo': 'bar'} assert info['cookies'] == {'Baz': 'qux', 'Foo': 'bar'}
@ -142,57 +144,57 @@ def test_fetch_url_cookies(mocker, fake_ansible_module):
def test_fetch_url_connectionerror(open_url_mock, fake_ansible_module): def test_fetch_url_connectionerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = ConnectionError('TESTS') open_url_mock.side_effect = ConnectionError('TESTS')
with pytest.raises(FailJson) as excinfo: with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/') fetch_url(fake_ansible_module, BASE_URL)
assert excinfo.value.kwargs['msg'] == 'TESTS' assert excinfo.value.kwargs['msg'] == 'TESTS'
assert 'http://ansible.com/' == excinfo.value.kwargs['url'] assert BASE_URL == excinfo.value.kwargs['url']
assert excinfo.value.kwargs['status'] == -1 assert excinfo.value.kwargs['status'] == -1
open_url_mock.side_effect = ValueError('TESTS') open_url_mock.side_effect = ValueError('TESTS')
with pytest.raises(FailJson) as excinfo: with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/') fetch_url(fake_ansible_module, BASE_URL)
assert excinfo.value.kwargs['msg'] == 'TESTS' assert excinfo.value.kwargs['msg'] == 'TESTS'
assert 'http://ansible.com/' == excinfo.value.kwargs['url'] assert BASE_URL == excinfo.value.kwargs['url']
assert excinfo.value.kwargs['status'] == -1 assert excinfo.value.kwargs['status'] == -1
def test_fetch_url_httperror(open_url_mock, fake_ansible_module): def test_fetch_url_httperror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = urllib.error.HTTPError( open_url_mock.side_effect = urllib.error.HTTPError(
'http://ansible.com/', BASE_URL,
500, 500,
'Internal Server Error', 'Internal Server Error',
{'Content-Type': 'application/json'}, {'Content-Type': 'application/json'},
io.StringIO('TESTS') io.StringIO('TESTS')
) )
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
assert info == {'msg': 'HTTP Error 500: Internal Server Error', 'body': 'TESTS', assert info == {'msg': 'HTTP Error 500: Internal Server Error', 'body': 'TESTS',
'status': 500, 'url': 'http://ansible.com/', 'content-type': 'application/json'} 'status': 500, 'url': BASE_URL, 'content-type': 'application/json'}
def test_fetch_url_urlerror(open_url_mock, fake_ansible_module): def test_fetch_url_urlerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = urllib.error.URLError('TESTS') open_url_mock.side_effect = urllib.error.URLError('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
assert info == {'msg': 'Request failed: <urlopen error TESTS>', 'status': -1, 'url': 'http://ansible.com/'} assert info == {'msg': 'Request failed: <urlopen error TESTS>', 'status': -1, 'url': BASE_URL}
def test_fetch_url_socketerror(open_url_mock, fake_ansible_module): def test_fetch_url_socketerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = socket.error('TESTS') open_url_mock.side_effect = socket.error('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
assert info == {'msg': 'Connection failure: TESTS', 'status': -1, 'url': 'http://ansible.com/'} assert info == {'msg': 'Connection failure: TESTS', 'status': -1, 'url': BASE_URL}
def test_fetch_url_exception(open_url_mock, fake_ansible_module): def test_fetch_url_exception(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = Exception('TESTS') open_url_mock.side_effect = Exception('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
exception = info.pop('exception') exception = info.pop('exception')
assert info == {'msg': 'An unknown error occurred: TESTS', 'status': -1, 'url': 'http://ansible.com/'} assert info == {'msg': 'An unknown error occurred: TESTS', 'status': -1, 'url': BASE_URL}
assert "Exception: TESTS" in exception assert "Exception: TESTS" in exception
def test_fetch_url_badstatusline(open_url_mock, fake_ansible_module): def test_fetch_url_badstatusline(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = http.client.BadStatusLine('TESTS') open_url_mock.side_effect = http.client.BadStatusLine('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/') r, info = fetch_url(fake_ansible_module, BASE_URL)
assert info == {'msg': 'Connection failure: connection was closed before a valid response was received: TESTS', 'status': -1, 'url': 'http://ansible.com/'} assert info == {'msg': 'Connection failure: connection was closed before a valid response was received: TESTS', 'status': -1, 'url': BASE_URL}

34
test/units/template/test_templar.py
Unescape Escape View File

@ -22,7 +22,7 @@ from jinja2.runtime import Context
import unittest import unittest
from ansible import constants as C from ansible import constants as C
from ansible.errors import AnsibleError, AnsibleUndefinedVariable from ansible.errors import AnsibleError, AnsibleUndefinedVariable, AnsibleAssertionError
from ansible.plugins.loader import init_plugin_loader from ansible.plugins.loader import init_plugin_loader
from ansible.template import Templar, AnsibleContext, AnsibleEnvironment, AnsibleUndefined from ansible.template import Templar, AnsibleContext, AnsibleEnvironment, AnsibleUndefined
from ansible.utils.unsafe_proxy import AnsibleUnsafe, wrap_var from ansible.utils.unsafe_proxy import AnsibleUnsafe, wrap_var
@ -70,8 +70,7 @@ class TestTemplarTemplate(BaseTemplar, unittest.TestCase):
def test_lookup_jinja_dict_key_in_static_vars(self): def test_lookup_jinja_dict_key_in_static_vars(self):
res = self.templar.template("{'some_static_var': '{{ some_var }}'}", res = self.templar.template("{'some_static_var': '{{ some_var }}'}",
static_vars=['some_static_var']) static_vars=['some_static_var'])
# self.assertEqual(res['{{ a_keyword }}'], "blip") assert res['some_static_var'] == "blip"
print(res)
def test_is_possibly_template_true(self): def test_is_possibly_template_true(self):
tests = [ tests = [
@ -230,8 +229,8 @@ class TestTemplarMisc(BaseTemplar, unittest.TestCase):
self.assertEqual(templar.template("{{foo}}\n", preserve_trailing_newlines=False), "bar") self.assertEqual(templar.template("{{foo}}\n", preserve_trailing_newlines=False), "bar")
self.assertEqual(templar.template("{{bam}}"), "bar") self.assertEqual(templar.template("{{bam}}"), "bar")
self.assertEqual(templar.template("{{num}}"), 1) self.assertEqual(templar.template("{{num}}"), 1)
self.assertEqual(templar.template("{{var_true}}"), True) assert templar.template("{{var_true}}")
self.assertEqual(templar.template("{{var_false}}"), False) assert not templar.template("{{var_false}}")
self.assertEqual(templar.template("{{var_dict}}"), dict(a="b")) self.assertEqual(templar.template("{{var_dict}}"), dict(a="b"))
self.assertEqual(templar.template("{{bad_dict}}"), "{a='b'") self.assertEqual(templar.template("{{bad_dict}}"), "{a='b'")
self.assertEqual(templar.template("{{var_list}}"), [1]) self.assertEqual(templar.template("{{var_list}}"), [1])
@ -251,11 +250,8 @@ class TestTemplarMisc(BaseTemplar, unittest.TestCase):
templar.available_variables = dict(foo="bam") templar.available_variables = dict(foo="bam")
self.assertEqual(templar.template("{{foo}}"), "bam") self.assertEqual(templar.template("{{foo}}"), "bam")
# variables must be a dict() for available_variables setter # variables must be a dict() for available_variables setter
# FIXME Use assertRaises() as a context manager (added in 2.7) once we do not run tests on Python 2.6 anymore. with self.assertRaisesRegex(AnsibleAssertionError, r"the type of 'variables'"):
try:
templar.available_variables = "foo=bam" templar.available_variables = "foo=bam"
except AssertionError:
pass
def test_templar_escape_backslashes(self): def test_templar_escape_backslashes(self):
# Rule of thumb: If escape backslashes is True you should end up with # Rule of thumb: If escape backslashes is True you should end up with
@ -341,29 +337,29 @@ class TestTemplarLookup(BaseTemplar, unittest.TestCase):
res = self.templar._lookup('list', '{{ some_unsafe_var }}', wantlist=True) res = self.templar._lookup('list', '{{ some_unsafe_var }}', wantlist=True)
for lookup_result in res: for lookup_result in res:
self.assertTrue(self.is_unsafe(lookup_result)) self.assertTrue(self.is_unsafe(lookup_result))
# self.assertIsInstance(lookup_result, AnsibleUnsafe) assert isinstance(lookup_result, AnsibleUnsafe)
# Should this be an AnsibleUnsafe # TODO: Should this be an AnsibleUnsafe
# self.assertIsInstance(res, AnsibleUnsafe) # self.assertIsInstance(res, AnsibleUnsafe)
def test_lookup_jinja_dict(self): def test_lookup_jinja_dict(self):
res = self.templar._lookup('list', {'{{ a_keyword }}': '{{ some_var }}'}) res = self.templar._lookup('list', {'{{ a_keyword }}': '{{ some_var }}'})
self.assertEqual(res['{{ a_keyword }}'], "blip") self.assertEqual(res['{{ a_keyword }}'], "blip")
assert isinstance(res['{{ a_keyword }}'], AnsibleUnsafe)
# TODO: Should this be an AnsibleUnsafe # TODO: Should this be an AnsibleUnsafe
# self.assertIsInstance(res['{{ a_keyword }}'], AnsibleUnsafe)
# self.assertIsInstance(res, AnsibleUnsafe) # self.assertIsInstance(res, AnsibleUnsafe)
def test_lookup_jinja_dict_unsafe(self): def test_lookup_jinja_dict_unsafe(self):
res = self.templar._lookup('list', {'{{ some_unsafe_key }}': '{{ some_unsafe_var }}'}) res = self.templar._lookup('list', {'{{ some_unsafe_key }}': '{{ some_unsafe_var }}'})
self.assertTrue(self.is_unsafe(res['{{ some_unsafe_key }}'])) self.assertTrue(self.is_unsafe(res['{{ some_unsafe_key }}']))
# self.assertIsInstance(res['{{ some_unsafe_key }}'], AnsibleUnsafe) assert isinstance(res['{{ some_unsafe_key }}'], AnsibleUnsafe)
# TODO: Should this be an AnsibleUnsafe # TODO: Should this be an AnsibleUnsafe
# self.assertIsInstance(res, AnsibleUnsafe) # self.assertIsInstance(res, AnsibleUnsafe)
def test_lookup_jinja_dict_unsafe_value(self): def test_lookup_jinja_dict_unsafe_value(self):
res = self.templar._lookup('list', {'{{ a_keyword }}': '{{ some_unsafe_var }}'}) res = self.templar._lookup('list', {'{{ a_keyword }}': '{{ some_unsafe_var }}'})
self.assertTrue(self.is_unsafe(res['{{ a_keyword }}'])) self.assertTrue(self.is_unsafe(res['{{ a_keyword }}']))
# self.assertIsInstance(res['{{ a_keyword }}'], AnsibleUnsafe) assert isinstance(res['{{ a_keyword }}'], AnsibleUnsafe)
# TODO: Should this be an AnsibleUnsafe # TODO: Should this be an AnsibleUnsafe
# self.assertIsInstance(res, AnsibleUnsafe) # self.assertIsInstance(res, AnsibleUnsafe)
@ -393,14 +389,14 @@ class TestAnsibleContext(BaseTemplar, unittest.TestCase):
def test_resolve_unsafe(self): def test_resolve_unsafe(self):
context = self._context(variables={'some_unsafe_key': wrap_var('some_unsafe_string')}) context = self._context(variables={'some_unsafe_key': wrap_var('some_unsafe_string')})
res = context.resolve('some_unsafe_key') res = context.resolve('some_unsafe_key')
# self.assertIsInstance(res, AnsibleUnsafe) assert isinstance(res, AnsibleUnsafe)
self.assertTrue(self.is_unsafe(res), self.assertTrue(self.is_unsafe(res),
'return of AnsibleContext.resolve (%s) was expected to be marked unsafe but was not' % res) 'return of AnsibleContext.resolve (%s) was expected to be marked unsafe but was not' % res)
def test_resolve_unsafe_list(self): def test_resolve_unsafe_list(self):
context = self._context(variables={'some_unsafe_key': [wrap_var('some unsafe string 1')]}) context = self._context(variables={'some_unsafe_key': [wrap_var('some unsafe string 1')]})
res = context.resolve('some_unsafe_key') res = context.resolve('some_unsafe_key')
# self.assertIsInstance(res[0], AnsibleUnsafe) assert isinstance(res[0], AnsibleUnsafe)
self.assertTrue(self.is_unsafe(res), self.assertTrue(self.is_unsafe(res),
'return of AnsibleContext.resolve (%s) was expected to be marked unsafe but was not' % res) 'return of AnsibleContext.resolve (%s) was expected to be marked unsafe but was not' % res)
@ -416,15 +412,15 @@ class TestAnsibleContext(BaseTemplar, unittest.TestCase):
context = self._context(variables={'some_key': 'some_string'}) context = self._context(variables={'some_key': 'some_string'})
res = context.resolve('some_key') res = context.resolve('some_key')
self.assertEqual(res, 'some_string') self.assertEqual(res, 'some_string')
# self.assertNotIsInstance(res, AnsibleUnsafe) assert not isinstance(res, AnsibleUnsafe)
self.assertFalse(self.is_unsafe(res), self.assertFalse(self.is_unsafe(res),
'return of AnsibleContext.resolve (%s) was not expected to be marked unsafe but was' % res) 'return of AnsibleContext.resolve (%s) was not expected to be marked unsafe but was' % res)
def test_resolve_none(self): def test_resolve_none(self):
context = self._context(variables={'some_key': None}) context = self._context(variables={'some_key': None})
res = context.resolve('some_key') res = context.resolve('some_key')
self.assertEqual(res, None) assert res is None
# self.assertNotIsInstance(res, AnsibleUnsafe) assert not isinstance(res, AnsibleUnsafe)
self.assertFalse(self.is_unsafe(res), self.assertFalse(self.is_unsafe(res),
'return of AnsibleContext.resolve (%s) was not expected to be marked unsafe but was' % res) 'return of AnsibleContext.resolve (%s) was not expected to be marked unsafe but was' % res)

69
test/units/utils/test_helpers.py
Unescape Escape View File

@ -1,33 +1,52 @@
# (c) 2015, Marius Gedminas <marius@gedmin.as> # (c) 2015, Marius Gedminas <marius@gedmin.as>
# # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
from __future__ import annotations from __future__ import annotations
import unittest import pytest
from ansible.utils.helpers import pct_to_int from datetime import datetime
from ansible.utils.helpers import pct_to_int, object_to_dict, deduplicate_list
class TestHelpers(unittest.TestCase):
def test_pct_to_int(self): pct_to_int_testdata = [
self.assertEqual(pct_to_int(1, 100), 1) pytest.param(
self.assertEqual(pct_to_int(-1, 100), -1) 1, 100, 1, 1, id="positive_percentage"
self.assertEqual(pct_to_int("1%", 10), 1) ),
self.assertEqual(pct_to_int("1%", 10, 0), 0) pytest.param(
self.assertEqual(pct_to_int("1", 100), 1) -1, 100, 1, -1, id="negative_percentage"
self.assertEqual(pct_to_int("10%", 100), 10) ),
pytest.param(
"1%", 10, 1, 1, id="string_percentage"
),
pytest.param(
"1%", 10, 0, 0, id="string_percentage_with_zero_min_value"
),
pytest.param(
"1", 100, 1, 1, id="string_percentage_without_sign"
),
pytest.param(
"10%", 100, 1, 10, id="string_percentage_two_digit"
)
]
@pytest.mark.parametrize("value,num_items,min_value,expected", pct_to_int_testdata)
def test_pct_to_int(value, num_items, min_value, expected):
assert pct_to_int(value, num_items, min_value) == expected
def test_object_to_dict():
test_dict = object_to_dict(datetime(2024, 7, 30))
assert test_dict['day'] == 30
assert test_dict['year'] == 2024
assert test_dict['month'] == 7
test_dict_without_day = object_to_dict(datetime(2024, 7, 30), exclude=['day'])
assert 'day' not in list(test_dict_without_day.keys())
def test_deduplicate_list():
assert deduplicate_list([1, 2, 2, 3]) == [1, 2, 3]
assert deduplicate_list([1, 2, 3]) == [1, 2, 3]

6
test/units/utils/test_vars.py
Unescape Escape View File

@ -29,12 +29,6 @@ from ansible.vars.manager import VarsWithSources
class TestVariableUtils(unittest.TestCase): class TestVariableUtils(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
combine_vars_merge_data = ( combine_vars_merge_data = (
dict( dict(
a=dict(a=1), a=dict(a=1),

Write Preview
Loading…
Cancel
Save