109 lines
5.3 KiB
Python
109 lines
5.3 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
spec = importlib.util.spec_from_file_location('remote', HERE / 'remote.py')
|
|
remote = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(remote)
|
|
|
|
|
|
class LifecycleTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.home = Path(tempfile.mkdtemp())
|
|
self.addCleanup(lambda: shutil.rmtree(self.home))
|
|
self.p = remote.paths(self.home)
|
|
for p in self.p.values():
|
|
if p != self.p['current']:
|
|
remote.mkdir(p)
|
|
self.staging = self.home / '.local/share/go-sip-tools' / ('a' * 40)
|
|
remote.mkdir(self.staging)
|
|
self.hashes = {}
|
|
for name in ('sip-go-agent', 'preprod-probe'):
|
|
(self.staging / name).write_bytes(b'synthetic executable')
|
|
self.hashes[name] = hashlib.sha256((self.staging / name).read_bytes()).hexdigest()
|
|
self.payload = {'commit': 'a' * 40, 'staging': str(self.staging), 'hashes': self.hashes, 'source': {'dispatcher_id': 'synthetic', 'probe': {}}}
|
|
self.state = self.p['state'] / 'retained-recording'
|
|
self.state.write_bytes(b'original recording')
|
|
self.calls = []
|
|
for name, value in (('host_check', {}), ('ari', []), ('unit_properties', {'ActiveState': 'active'}), ('status', {'ok': True, 'business_ready': False}), ('validate_environments', None)):
|
|
p = patch.object(remote, name, return_value=value)
|
|
p.start(); self.addCleanup(p.stop)
|
|
p = patch.object(remote, 'stop_for_change', side_effect=lambda *a, **k: self.calls.append('stop'))
|
|
p.start(); self.addCleanup(p.stop)
|
|
p = patch.object(remote, 'bootstrap', side_effect=self.bootstrap)
|
|
p.start(); self.addCleanup(p.stop)
|
|
p = patch.object(remote, 'run', return_value='')
|
|
p.start(); self.addCleanup(p.stop)
|
|
p = patch.object(remote.time, 'sleep')
|
|
p.start(); self.addCleanup(p.stop)
|
|
p = patch.object(remote.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1, json.dumps({'success': False, 'phase': 'sip', 'http_status': 404, 'mq_connected': True, 'mq_connection_closed': True})))
|
|
p.start(); self.addCleanup(p.stop)
|
|
|
|
def bootstrap(self, *args):
|
|
remote.mkdir(self.p['state'])
|
|
return {'agent': {}, 'dispatcher': {}}
|
|
|
|
def test_saas_down_installs_and_repeat_preserves_original_state(self):
|
|
with patch.object(remote, 'backup_dir') as backup:
|
|
first = remote.install(self.payload, self.p)
|
|
second = remote.install(self.payload, self.p)
|
|
backup.assert_not_called()
|
|
self.assertEqual(self.state.read_bytes(), b'original recording')
|
|
self.assertEqual(first['dependency_state'], 'saas_pending')
|
|
self.assertFalse(second['reset_performed'])
|
|
self.assertTrue(self.p['current'].is_symlink())
|
|
for role in ('agent', 'dispatcher'):
|
|
unit = (self.p['units'] / ('go-sip-' + role + '.service')).read_text()
|
|
self.assertIn('Restart=on-failure', unit)
|
|
self.assertNotIn('--mode mock', unit)
|
|
|
|
def test_explicit_reset_stops_and_archives_before_deleting(self):
|
|
original = self.state.read_bytes()
|
|
def backup(*args):
|
|
self.assertIn('stop', self.calls)
|
|
self.assertEqual(self.state.read_bytes(), original)
|
|
self.calls.append('backup')
|
|
with patch.object(remote, 'backup_dir', side_effect=backup):
|
|
result = remote.install(dict(self.payload, reset_state=True, confirm_reset=True), self.p)
|
|
self.assertTrue(result['reset_performed'])
|
|
self.assertEqual(self.calls[:2], ['stop', 'backup'])
|
|
self.assertFalse(self.state.exists())
|
|
|
|
def test_unconfirmed_reset_does_not_remove_original(self):
|
|
with patch.object(remote, 'backup_dir') as backup:
|
|
with self.assertRaises(ValueError):
|
|
remote.install(dict(self.payload, reset_state=True), self.p)
|
|
backup.assert_not_called()
|
|
self.assertEqual(self.state.read_bytes(), b'original recording')
|
|
|
|
def test_independent_mq_failure_is_not_saas_waiting(self):
|
|
failed = subprocess.CompletedProcess([], 1, json.dumps({'success': False, 'phase': 'mq_connect', 'mq_connected': False}))
|
|
with patch.object(remote.subprocess, 'run', return_value=failed):
|
|
with self.assertRaisesRegex(ValueError, 'independent dependency failure'):
|
|
remote.install(self.payload, self.p)
|
|
self.assertEqual(self.state.read_bytes(), b'original recording')
|
|
|
|
def test_active_calls_prevent_stopping_agent(self):
|
|
# Exercise the real stop routine separately from the install mocks.
|
|
namespace = {}
|
|
exec(compile(HERE.joinpath('remote.py').read_text(), str(HERE / 'remote.py'), 'exec'), namespace)
|
|
calls = []
|
|
namespace['unit_properties'] = lambda u: {'FragmentPath': '/owned/unit'}
|
|
namespace['run'] = lambda args: calls.append(args)
|
|
namespace['ari'] = lambda *a: [{'id': 'active'}]
|
|
with self.assertRaises(ValueError):
|
|
namespace['stop_for_change'](self.p)
|
|
self.assertTrue(any('go-sip-dispatcher.service' in a for a in calls))
|
|
self.assertFalse(any('go-sip-agent.service' in a for a in calls))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|