84 lines
3.7 KiB
Python
84 lines
3.7 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import json
|
|
import pytest
|
|
|
|
|
|
def updater():
|
|
path=Path(__file__).resolve().parents[1]/'deploy/update.py'
|
|
spec=importlib.util.spec_from_file_location('linux_updater',path)
|
|
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class Commands:
|
|
def __init__(self, *, dirty=False, same=False, diverged=False, failed=None, states=None):
|
|
self.calls=[];self.dirty=dirty;self.same=same;self.diverged=diverged
|
|
self.failed=failed;self.states=states or ['active','active']
|
|
|
|
def __call__(self,*args):
|
|
self.calls.append(args)
|
|
if self.failed and self.failed(args):raise RuntimeError('injected failure')
|
|
if args[0]=='git':
|
|
cmd=args[3:]
|
|
if cmd[0]=='status':return ' M edited.py' if self.dirty else ''
|
|
if cmd==('rev-parse','HEAD'):return 'a'*40
|
|
if cmd[0]=='rev-parse':return 'origin/main' if '--abbrev-ref' in cmd else ('a' if self.same else 'b')*40
|
|
if cmd[0]=='merge-base' and self.diverged:raise RuntimeError('not fast forward')
|
|
if args[:2]==('systemctl','show'):
|
|
return self.states[0 if 'worker' in args[-1] else 1]
|
|
return ''
|
|
|
|
|
|
def test_plan_has_no_mutations():
|
|
module=updater()
|
|
result=subprocess.run([sys.executable,module.__file__,'--dry-run'],capture_output=True,text=True,check=True)
|
|
assert json.loads(result.stdout)['status']=='PLAN'
|
|
|
|
|
|
@pytest.mark.parametrize('kwargs',[{'dirty':True},{'diverged':True}])
|
|
def test_refuses_local_changes_or_divergence_before_stopping(kwargs):
|
|
module=updater();commands=Commands(**kwargs)
|
|
with pytest.raises(RuntimeError):module.update(execute=commands)
|
|
assert not any(c[0]=='systemctl' for c in commands.calls)
|
|
|
|
|
|
def test_same_revision_is_installed_after_bootstrap_pull(monkeypatch):
|
|
module=updater();commands=Commands(same=True)
|
|
monkeypatch.setattr(module,'trusted',lambda root:None)
|
|
assert module.update(execute=commands)['status']=='UPDATED'
|
|
assert any('pip' in c for c in commands.calls)
|
|
|
|
|
|
@pytest.mark.parametrize('states,started',[(['active','active'],['api','worker']),(['active','inactive'],['worker']),(['inactive','inactive'],[])])
|
|
def test_update_order_and_preserves_inactive_services(monkeypatch,states,started):
|
|
module=updater();commands=Commands(states=states)
|
|
monkeypatch.setattr(module,'trusted',lambda root:None)
|
|
result=module.update(execute=commands)
|
|
assert result['status']=='UPDATED'
|
|
calls=commands.calls
|
|
stops=[c for c in calls if c[:2]==('systemctl','stop')]
|
|
assert stops==[('systemctl','stop',s) for s in module.SERVICES]
|
|
merged=next(i for i,c in enumerate(calls) if c[0]=='git' and c[3]=='merge')
|
|
installed=next(i for i,c in enumerate(calls) if 'pip' in c)
|
|
assert calls.index(stops[-1])<merged<installed
|
|
assert [c[-1].split('-')[-1].split('.')[0] for c in calls if c[:2]==('systemctl','start')]==started
|
|
assert not any('enable' in c or 'reset' in c or 'clean' in c for c in calls)
|
|
|
|
|
|
@pytest.mark.parametrize('failure',['install','start'])
|
|
def test_failure_keeps_both_services_stopped(monkeypatch,failure):
|
|
module=updater()
|
|
commands=Commands(failed=lambda c: ('pip' in c) if failure=='install' else c[:2]==('systemctl','start'))
|
|
monkeypatch.setattr(module,'trusted',lambda root:None)
|
|
with pytest.raises(RuntimeError):module.update(execute=commands)
|
|
assert commands.calls[-1]==('systemctl','stop',*module.SERVICES)
|
|
|
|
|
|
def test_fetch_failure_does_not_stop_services():
|
|
module=updater();commands=Commands(failed=lambda c:'fetch' in c)
|
|
with pytest.raises(RuntimeError):module.update(execute=commands)
|
|
assert not any(c[0]=='systemctl' for c in commands.calls)
|