From 333658ce09c4e25971facd4dd4be328782995c98 Mon Sep 17 00:00:00 2001 From: Christopher Patton Date: Sat, 30 May 2020 11:58:50 -0700 Subject: [PATCH] Add unit tests for existing functionality --- library/aur.py | 12 +- library/aur_test.py | 542 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 552 insertions(+), 2 deletions(-) create mode 100755 library/aur_test.py diff --git a/library/aur.py b/library/aur.py index e542232..28fe4d5 100644 --- a/library/aur.py +++ b/library/aur.py @@ -161,17 +161,25 @@ def build_command_prefix(use, extra_args, skip_pgp_check=False, ignore_arch=Fals return command +def find_package_url(package): + return 'https://aur.archlinux.org/rpc/?v=5&type=info&arg={}'.format(urllib.parse.quote(package)) + + +def download_package_url(path): + return 'https://aur.archlinux.org/{}'.format(path) + + def install_with_makepkg(module, package, extra_args, skip_pgp_check, ignore_arch): """ Install the specified package with makepkg """ module.get_bin_path('fakeroot', required=True) - f = open_url('https://aur.archlinux.org/rpc/?v=5&type=info&arg={}'.format(urllib.parse.quote(package))) + f = open_url(find_package_url(package)) result = json.loads(f.read().decode('utf8')) if result['resultcount'] != 1: return (1, '', 'package {} not found'.format(package)) result = result['results'][0] - f = open_url('https://aur.archlinux.org/{}'.format(result['URLPath'])) + f = open_url(download_package_url(result['URLPath'])) with tempfile.TemporaryDirectory() as tmpdir: tar = tarfile.open(mode='r|*', fileobj=f) tar.extractall(tmpdir) diff --git a/library/aur_test.py b/library/aur_test.py new file mode 100755 index 0000000..a06aaeb --- /dev/null +++ b/library/aur_test.py @@ -0,0 +1,542 @@ +#!/usr/bin/env python3 + +import json +import tarfile +import tempfile +import unittest +from unittest.mock import patch, MagicMock, Mock + +from ansible.module_utils import basic +from ansible.module_utils._text import to_bytes + +import aur + + +def Any(cls): + ''' + Return an instance of a class that extends the given type and compares as + equal to anything thrown at it. + This is useful when making assertions about how a mock method was invoked + when you only care to constrain a subset of function parameters. + ''' + class Any(cls): + def __eq__(self, other): + return True + + return Any() + + +class AnsibleExitJson(Exception): + ''' + Raised by the mocked function AnsibleModule.exit_json. + Used to terminate control-flow tested functions. + ''' + pass + + +class AnsibleFailJson(Exception): + ''' + Raised by the mocked function AnsibleModule.fail_json. + Used to terminate control-flow tested functions. + ''' + pass + + +def exit_json(*args, **kwargs): + ''' + Package kwargs into an exception and raise to terminate control-flow. + ''' + raise AnsibleExitJson(kwargs) + + +def fail_json(*args, **kwargs): + ''' + Package kwargs into an exception and raise to terminate control-flow. + ''' + raise AnsibleFailJson(kwargs) + + +class MockedFunctionArgumentError(Exception): + ''' + Raised by MatchFnArgs.__call__ when provided arguments do not match. + ''' + def __init__(self, expected, actual): + self.expected = expected + self.actual = actual + + def __str__(self): + return '\nexpected: {} {}\n actual: {} {}'.format( + str(self.expected[0]), + str(self.expected[1]), + str(self.actual[0]), + str(self.actual[1]), + ) + + +class MatchFnArgs: + ''' + A callable object that will match arguments provided at instantiation to + those provided at invocation. + ''' + def __init__(self, *args, **kwargs): + self.args = args + self.kwargs = kwargs + + def __call__(self, *args, **kwargs): + if self.args != args or self.kwargs != kwargs: + raise MockedFunctionArgumentError( + expected=(self.args, self.kwargs), + actual=(args, kwargs), + ) + + +class MatchCall: + ''' + A callable object that matches provided arguments, raises a potential + exception, and returns a specific value. + ''' + def __init__(self, match_args, return_value=None, exception=None): + self.match_args = match_args + self.return_value = return_value + self.exception = exception + + def __call__(self, *args, **kwargs): + self.match_args(*args, **kwargs) + if self.exception: + raise self.exception + return self.return_value + + +class IterativeMatchCall: + ''' + A callable object that matches a series of calls, invoking the next provided + side-effect each call. + ''' + def __init__(self, side_effects): + self.side_effects = side_effects + self.side_effect = iter(self.side_effects) + + def __call__(self, *args, **kwargs): + return next(self.side_effect)(*args, **kwargs) + + +class AurModuleInvalidParamsTest(unittest.TestCase): + def setUp(self): + # Patch the two terminal module methods here so we can avoid repetition + # in every test case. + self.exit_json = patch.object(aur.AnsibleModule, 'exit_json', exit_json) + self.fail_json = patch.object(aur.AnsibleModule, 'fail_json', fail_json) + self.exit_json.start() + self.fail_json.start() + + def tearDown(self): + self.exit_json.stop() + self.fail_json.stop() + + def set_module_args(self, args): + args = json.dumps({'ANSIBLE_MODULE_ARGS': args}) + basic._ANSIBLE_ARGS = to_bytes(args) + + def test_empty(self): + with self.assertRaises(AnsibleFailJson): + self.set_module_args({}) + aur.make_module() + + @patch.object(aur.AnsibleModule, 'get_bin_path') + def test_name_and_upgrade(self, get_bin_path): + with self.assertRaises(AnsibleFailJson): + self.set_module_args({ + 'name': ['ansible'], + 'upgrade': True, + }) + aur.make_module() + get_bin_path.assert_not_called() + + @patch.object(aur.AnsibleModule, 'get_bin_path') + def test_bad_helper(self, get_bin_path): + with self.assertRaises(AnsibleFailJson): + self.set_module_args({ + 'name': ['ansible'], + 'use': 'doesnotexist', + }) + aur.make_module() + get_bin_path.assert_not_called() + + @patch.object(aur.AnsibleModule, 'get_bin_path') + def test_extra_args_without_use(self, get_bin_path): + with self.assertRaises(AnsibleFailJson) as cm: + self.set_module_args({ + 'name': ['ansible'], + 'extra_args': 'custom-arg', + }) + aur.make_module() + get_bin_path.assert_not_called() + + @patch.object(aur.AnsibleModule, 'get_bin_path') + def test_extra_args_with_auto_use(self, get_bin_path): + with self.assertRaises(AnsibleFailJson) as cm: + self.set_module_args({ + 'name': ['ansible'], + 'use': 'auto', + 'extra_args': 'custom-arg', + }) + aur.make_module() + get_bin_path.assert_not_called() + + @patch.object(aur.AnsibleModule, 'get_bin_path') + def test_upgrade_with_makepkg_use(self, get_bin_path): + get_bin_path.return_value = False + with self.assertRaises(AnsibleFailJson) as cm: + self.set_module_args({ + 'upgrade': True, + 'use': 'makepkg', + }) + aur.make_module() + get_bin_path.assert_not_called() + + +class AurModuleApplyTest(unittest.TestCase): + def setUp(self): + self.module = MagicMock() + self.module.check_mode = False + self.module._diff = False + self.module.exit_json.side_effect = exit_json + self.module.fail_json.side_effect = fail_json + + self.ansible_module = patch( + 'aur.AnsibleModule', + return_value=self.module, + ) + self.ansible_module.start() + + def tearDown(self): + self.ansible_module.stop() + + +class AurModuleApplyCheckTest(AurModuleApplyTest): + def test_check_installed(self): + self.module.check_mode = True + self.module.params = {'name': ['ansible']} + self.module.run_command.return_value = (0, '', '') + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), 'makepkg') + + self.module.run_command.assert_called_once_with( + ['pacman', '-Q', 'ansible'], + check_rc=False, + ) + self.module.exit_json.assert_called_once_with( + changed=False, + diff=Any(dict), + msg=Any(str), + ) + self.module.fail_json.assert_not_called() + + def test_check_not_installed(self): + self.module.check_mode = True + self.module.params = {'name': ['ansible']} + self.module.run_command.return_value = (1, '', '') + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), 'makepkg') + + self.module.run_command.assert_called_once_with( + ['pacman', '-Q', 'ansible'], + check_rc=False, + ) + self.module.exit_json.assert_called_once_with( + changed=True, + diff=Any(dict), + msg=Any(str), + ) + self.module.fail_json.assert_not_called() + + +class AurModuleApplyUpgradeTest(AurModuleApplyTest): + def test_upgrade_no_change(self): + use = 'yay' + extra_args = None + + self.module.params = { + 'upgrade': True, + 'extra_args': extra_args, + 'aur_only': False, + } + self.module.run_command.return_value = (0, '', '') + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called_once_with( + aur.build_command_prefix(use, extra_args) + ['-u'], + check_rc=True, + ) + self.module.exit_json.assert_called_once_with( + changed=False, + helper=use, + msg=Any(str), + ) + self.module.fail_json.assert_not_called() + + def test_upgrade_change(self): + use = 'yay' + extra_args = None + + self.module.params = { + 'upgrade': True, + 'extra_args': extra_args, + 'aur_only': False, + } + self.module.run_command.return_value = (0, 'something happened', '') + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called_once_with( + aur.build_command_prefix(use, extra_args) + ['-u'], + check_rc=True, + ) + self.module.exit_json.assert_called_once_with( + changed=True, + helper=use, + msg=Any(str), + ) + self.module.fail_json.assert_not_called() + + +class AurModuleApplyInstallTest(AurModuleApplyTest): + def setUp(self): + super().setUp() + self.close_on_teardown = [] + + def tearDown(self): + super().tearDown() + for element in self.close_on_teardown: + element.close() + + def test_install_present_package(self): + use = 'yay' + extra_args = None + + self.module.params = { + 'name': ['ansible'], + 'state': 'present', + 'extra_args': extra_args, + 'skip_pgp_check': False, + 'ignore_arch': False, + 'aur_only': False, + } + self.module.run_command.return_value = (0, '', '') + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called_once_with( + ['pacman', '-Q', 'ansible'], + check_rc=False, + ) + self.module.exit_json.assert_called_once_with( + changed=False, + msg=Any(str), + helper=use, + rc=0, + ) + self.module.fail_json.assert_not_called() + + def test_install_absent_package_success(self): + use = 'yay' + extra_args = None + + self.module.params = { + 'name': ['ansible'], + 'state': 'present', + 'extra_args': extra_args, + 'skip_pgp_check': False, + 'ignore_arch': False, + 'aur_only': False, + } + self.module.run_command.side_effect = IterativeMatchCall([ + MatchCall( + MatchFnArgs(['pacman', '-Q', 'ansible'], check_rc=False), + return_value=(1, '', ''), + ), + MatchCall( + MatchFnArgs( + aur.build_command_prefix(use, extra_args) + ['ansible'], + check_rc=True, + ), + return_value=(0, 'something happened', ''), + ), + ]) + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called() + self.module.exit_json.assert_called_once_with( + changed=True, + msg=Any(str), + helper=use, + rc=0, + ) + self.module.fail_json.assert_not_called() + + def test_install_absent_package_failure(self): + use = 'yay' + extra_args = None + + self.module.params = { + 'name': ['ansible'], + 'state': 'present', + 'extra_args': extra_args, + 'skip_pgp_check': False, + 'ignore_arch': False, + 'aur_only': False, + } + self.module.run_command.side_effect = IterativeMatchCall([ + MatchCall( + MatchFnArgs(['pacman', '-Q', 'ansible'], check_rc=False), + return_value=(1, '', ''), + ), + MatchCall( + MatchFnArgs( + aur.build_command_prefix(use, extra_args) + ['ansible'], + check_rc=True, + ), + return_value=(1, '', ''), + ), + ]) + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called() + self.module.exit_json.assert_called_once_with( + changed=False, + msg=Any(str), + helper=use, + rc=1, + ) + self.module.fail_json.assert_not_called() + + def test_install_absent_package_extra_args_yay(self): + use = 'yay' + extra_args = 'custom-arg' + + self.module.params = { + 'name': ['ansible'], + 'state': 'present', + 'extra_args': extra_args, + 'skip_pgp_check': False, + 'ignore_arch': False, + 'aur_only': False, + } + self.module.run_command.side_effect = IterativeMatchCall([ + MatchCall( + MatchFnArgs(['pacman', '-Q', 'ansible'], check_rc=False), + return_value=(1, '', ''), + ), + MatchCall( + MatchFnArgs( + # Explicitly append extra_args instead of passing it to + # build_command_prefix() so we can be sure that our args are + # begin included. + aur.build_command_prefix(use, []) + [extra_args, 'ansible'], + check_rc=True, + ), + return_value=(0, 'something happened', ''), + ), + ]) + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + self.module.run_command.assert_called() + self.module.exit_json.assert_called_once_with( + changed=True, + msg=Any(str), + helper=use, + rc=0, + ) + self.module.fail_json.assert_not_called() + + @patch('aur.open_url') + def test_install_absent_package_extra_args_makepkg(self, open_url): + use = 'makepkg' + extra_args = 'custom-arg' + + # Populate a tempfile with a JSON-serialized response of querying for a + # package by name. + find_url_response = { + 'resultcount': 1, + 'results': [ + { + 'URLPath': 'url-path', + 'Name': 'ansible', + }, + ], + } + find_url_tempfile = tempfile.TemporaryFile(mode='w+b') + self.close_on_teardown.append(find_url_tempfile) + find_url_tempfile.write(json.dumps(find_url_response).encode('utf-8')) + find_url_tempfile.seek(0) + + # Populate a tempfile with the contents of an empty tarfile. + download_file_tempfile = tempfile.TemporaryFile('w+b') + self.close_on_teardown.append(download_file_tempfile) + tarfile.open(mode='w', fileobj=download_file_tempfile).close() + download_file_tempfile.seek(0) + + self.module.params = { + 'name': ['ansible'], + 'state': 'present', + 'extra_args': extra_args, + 'skip_pgp_check': False, + 'ignore_arch': False, + 'aur_only': False, + } + open_url.side_effect = IterativeMatchCall([ + MatchCall( + MatchFnArgs(aur.find_package_url('ansible')), + return_value=find_url_tempfile, + ), + MatchCall( + MatchFnArgs(aur.download_package_url('url-path')), + return_value=download_file_tempfile, + ), + ]) + self.module.run_command.side_effect = IterativeMatchCall([ + MatchCall( + MatchFnArgs(['pacman', '-Q', 'ansible'], check_rc=False), + return_value=(1, '', ''), + ), + MatchCall( + MatchFnArgs( + # Explicitly append extra_args instead of passing it to + # build_command_prefix() so we can be sure that our args are + # begin included. + aur.build_command_prefix(use, []) + [extra_args], + cwd=Any(str), + check_rc=True, + ), + return_value=(0, 'something happened', ''), + ), + ]) + + with self.assertRaises(AnsibleExitJson): + aur.apply_module(aur.AnsibleModule(), use) + + open_url.assert_called() + self.module.run_command.assert_called() + self.module.exit_json.assert_called_once_with( + changed=True, + msg=Any(str), + helper=use, + rc=0, + ) + self.module.fail_json.assert_not_called() + + +if __name__ == '__main__': + unittest.main()