test_upgrade_policy.py (4202B)
1 #!/usr/bin/env python3 2 """Check the selective upgrade policy and its restoration on real apt failure.""" 3 from pathlib import Path 4 import shlex 5 import subprocess 6 import tempfile 7 import unittest 8 9 from jinja2 import Environment 10 11 TEMPLATE = Path(__file__).resolve().parents[2] / 'roles/common_packages/templates/upgrade-policy-rc.d.j2' 12 13 14 def verify_upgrade_policy_restoration(container, play): 15 # This callback is only invoked in test's disposable container. 16 container('sh', '-eu', '-c', ''' 17 rm -f /usr/sbin/policy-rc.d 18 printf '#!/bin/sh\\nexit 0\\n' > /usr/sbin/taler-test-original-policy 19 chmod 755 /usr/sbin/taler-test-original-policy 20 ln -s taler-test-original-policy /usr/sbin/policy-rc.d 21 ''') 22 tasks = [{'ansible.builtin.include_role': {'name': 'common_packages', 'tasks_from': 'upgrade'}}] 23 try: 24 play(tasks, check_mode=True) 25 assert container('readlink', '/usr/sbin/policy-rc.d').stdout.strip() == 'taler-test-original-policy' 26 play(tasks) 27 assert container('readlink', '/usr/sbin/policy-rc.d').stdout.strip() == 'taler-test-original-policy' 28 # fcntl (not flock) is the locking mechanism used by apt/dpkg. 29 container('systemd-run', '--unit=taler-upgrade-lock-test', 'python3', '-c', 30 'import fcntl,time; f=open("/var/lib/dpkg/lock-frontend","w"); ' 31 'fcntl.lockf(f,fcntl.LOCK_EX); open("/tmp/taler-upgrade-locked","w").close(); time.sleep(120)') 32 container('sh', '-eu', '-c', 'while [ ! -e /tmp/taler-upgrade-locked ]; do sleep .1; done') 33 play([{'block': tasks, 'module_defaults': {'ansible.builtin.apt': {'lock_timeout': 0}}}], 34 expect_failure=True) 35 assert container('readlink', '/usr/sbin/policy-rc.d').stdout.strip() == 'taler-test-original-policy' 36 assert container('/usr/sbin/policy-rc.d', 'ssh', 'restart', check=False).returncode == 0 37 finally: 38 container('systemctl', 'stop', 'taler-upgrade-lock-test.service', check=False) 39 container('rm', '-f', '/usr/sbin/policy-rc.d', '/usr/sbin/taler-test-original-policy', '/tmp/taler-upgrade-locked') 40 print('PASS: selective policy restores the original symlink after success and apt failure', flush=True) 41 42 43 class UpgradePolicyTest(unittest.TestCase): 44 def setUp(self): 45 self.temp = tempfile.TemporaryDirectory(prefix='taler-policy-test-') 46 self.addCleanup(self.temp.cleanup) 47 self.root = Path(self.temp.name) 48 49 def policy(self, original=None): 50 backup = self.root / "original policy's file" 51 if original: 52 backup.write_text(original) 53 backup.chmod(0o700) 54 env = Environment() 55 env.filters['quote'] = shlex.quote 56 script = self.root / 'policy' 57 script.write_text(env.from_string(TEMPLATE.read_text()).render( 58 common_upgrade_original_policy={'stat': {'exists': original is not None}}, 59 common_upgrade_policy_backup={'path': str(backup)})) 60 return script 61 62 def test_only_managed_applications_are_denied_without_an_existing_policy(self): 63 script = self.policy() 64 for service in ['taler-exchange.target', 'taler-helper-auditor-wire-credit.service', 65 'libeufin-nexus', 'challenger-httpd', 'sms-challenger-httpd', 66 'email-challenger-httpd', 'postal-challenger-httpd']: 67 self.assertEqual(subprocess.run(['sh', str(script), service, 'restart']).returncode, 101) 68 self.assertEqual(subprocess.run(['sh', str(script), 'ssh', 'restart']).returncode, 0) 69 self.assertEqual(subprocess.run(['sh', str(script), '--quiet', 'taler-exchange', 'start']).returncode, 101) 70 71 def test_existing_policy_receives_other_actions_unchanged(self): 72 script = self.policy('#!/bin/sh\nprintf "%s\\n" "$@"\nexit 17\n') 73 result = subprocess.run(['sh', str(script), '--quiet', 'nginx', 'restart'], capture_output=True, text=True) 74 self.assertEqual(result.returncode, 17) 75 self.assertEqual(result.stdout, '--quiet\nnginx\nrestart\n') 76 self.assertEqual(subprocess.run(['sh', str(script), 'taler-exchange', 'start']).returncode, 101) 77 78 79 if __name__ == '__main__': 80 unittest.main(verbosity=2)