mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-13 10:36:35 +00:00
Merge remote-tracking branch 'origin/main' into feat/common-sports-helpers
This commit is contained in:
Chuck
Vendored
+17
@@ -841,6 +841,23 @@
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/system/auto-update",
|
||||
"api_v3.get_auto_update_status",
|
||||
[
|
||||
"GET",
|
||||
"HEAD",
|
||||
"OPTIONS"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/system/auto-update/dismiss",
|
||||
"api_v3.dismiss_auto_update_alert",
|
||||
[
|
||||
"OPTIONS",
|
||||
"POST"
|
||||
]
|
||||
],
|
||||
[
|
||||
"/api/v3/system/check-update",
|
||||
"api_v3.check_for_update",
|
||||
|
||||
@@ -0,0 +1,683 @@
|
||||
"""Weekly automatic updates (web_interface/auto_update.py).
|
||||
|
||||
The updater pulls code and restarts services unattended. These pin down that
|
||||
it runs only when asked and due, refuses to touch a checkout it could damage,
|
||||
never leaves new code running without a health check, and reports every
|
||||
failure instead of logging it and moving on.
|
||||
|
||||
Git runs for real against a throwaway origin/device clone pair; systemd, sudo
|
||||
and the health check service are faked.
|
||||
"""
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from web_interface import auto_update as au # noqa: E402
|
||||
|
||||
# Local time is pinned to UTC below, so 03:00 is inside the quiet hours.
|
||||
NIGHT = datetime(2026, 9, 14, 3, 0, tzinfo=timezone.utc).timestamp()
|
||||
NOON = datetime(2026, 9, 14, 12, 0, tzinfo=timezone.utc).timestamp()
|
||||
ON = {'auto_update': {'enabled': True}}
|
||||
OFF = {'auto_update': {'enabled': False}}
|
||||
|
||||
needs_git = pytest.mark.skipif(shutil.which('git') is None, reason='git not installed')
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def local_is_utc(monkeypatch):
|
||||
"""Pin 'local time' so quiet-hours checks don't depend on the host zone."""
|
||||
monkeypatch.setattr(au, '_zone', lambda name: timezone.utc)
|
||||
|
||||
|
||||
def git(cwd, *args):
|
||||
result = subprocess.run(['git', '-c', 'user.name=t', '-c', 'user.email=t@example.com', *args],
|
||||
cwd=str(cwd), capture_output=True, text=True)
|
||||
assert result.returncode == 0, result.stderr
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
class Repo:
|
||||
"""An origin, a checkout that publishes to it, and the device's clone."""
|
||||
|
||||
def __init__(self, tmp):
|
||||
self.seed = tmp / 'seed'
|
||||
self.seed.mkdir()
|
||||
git(self.seed, 'init', '-q', '-b', 'main')
|
||||
(self.seed / 'app.py').write_text('v = 1\n')
|
||||
git(self.seed, 'add', '.')
|
||||
git(self.seed, 'commit', '-qm', 'one')
|
||||
self.origin = tmp / 'origin.git'
|
||||
git(tmp, 'clone', '-q', '--bare', str(self.seed), str(self.origin))
|
||||
git(self.seed, 'remote', 'add', 'origin', str(self.origin))
|
||||
self.device = tmp / 'device'
|
||||
git(tmp, 'clone', '-q', str(self.origin), str(self.device))
|
||||
|
||||
def head(self):
|
||||
return git(self.device, 'rev-parse', 'HEAD')
|
||||
|
||||
def publish(self, text='v = 2\n'):
|
||||
(self.seed / 'app.py').write_text(text)
|
||||
git(self.seed, 'commit', '-qam', 'next')
|
||||
git(self.seed, 'push', '-q', 'origin', 'main')
|
||||
return git(self.seed, 'rev-parse', 'HEAD')
|
||||
|
||||
|
||||
def real_pull(device, **extra):
|
||||
def core_update():
|
||||
before = git(device, 'rev-parse', 'HEAD')
|
||||
r = subprocess.run(['git', 'pull', '-q', '--rebase', '--autostash'],
|
||||
cwd=str(device), capture_output=True, text=True)
|
||||
after = git(device, 'rev-parse', 'HEAD')
|
||||
result = {'status': 'success' if r.returncode == 0 else 'error',
|
||||
'message': 'Code updated successfully.' if r.returncode == 0 else r.stderr,
|
||||
'restart_required': before != after, 'dependency_failures': []}
|
||||
result.update(extra)
|
||||
return result
|
||||
return core_update
|
||||
|
||||
|
||||
class FakeStore:
|
||||
def __init__(self, plugins_dir, versions, bump=(), fail=()):
|
||||
self.plugins_dir = str(plugins_dir)
|
||||
self.bump, self.fail = set(bump), set(fail)
|
||||
for pid, version in versions.items():
|
||||
d = Path(plugins_dir) / pid
|
||||
d.mkdir(parents=True)
|
||||
(d / 'manifest.json').write_text(json.dumps({'id': pid, 'version': version}))
|
||||
self.updated_calls = []
|
||||
|
||||
def list_installed_plugins(self):
|
||||
return [p.name for p in Path(self.plugins_dir).iterdir()]
|
||||
|
||||
def _get_local_git_info(self, path):
|
||||
return None
|
||||
|
||||
def update_plugin(self, pid):
|
||||
self.updated_calls.append(pid)
|
||||
if pid in self.fail:
|
||||
return False
|
||||
if pid in self.bump:
|
||||
path = Path(self.plugins_dir) / pid / 'manifest.json'
|
||||
m = json.loads(path.read_text())
|
||||
m['version'] = '9.9.9'
|
||||
path.write_text(json.dumps(m))
|
||||
return True
|
||||
|
||||
|
||||
class Harness:
|
||||
"""An AutoUpdater on a real clone.
|
||||
|
||||
``helper`` is whether the health check's path unit is active; ``pickup``
|
||||
whether the health check actually starts when asked. Starting it is
|
||||
simulated the way systemd does it: the request file is consumed
|
||||
(ExecStartPre) and the verifier marks the pending update "verifying".
|
||||
"""
|
||||
|
||||
def __init__(self, tmp, repo, *, helper=True, pickup=True, core_update=None, store=None,
|
||||
disk_free=10 ** 12, display_active=True, enabled=True, clock=NIGHT):
|
||||
self.sudo, self.restarts, self.now = [], [], clock
|
||||
self.state_at_handoff = None
|
||||
|
||||
def run(args, **kwargs):
|
||||
if args[0] == 'sudo':
|
||||
self.sudo.append(list(args))
|
||||
return subprocess.CompletedProcess(args, 0, stdout='', stderr='')
|
||||
return subprocess.run(args, **kwargs)
|
||||
|
||||
def sleep(seconds):
|
||||
updater = self.updater
|
||||
if self.state_at_handoff is None:
|
||||
self.state_at_handoff = au.load_state(updater.state_file)
|
||||
if pickup and updater.request_file.exists():
|
||||
updater.request_file.unlink()
|
||||
pending = au._read_json(updater.pending_file)
|
||||
pending['status'] = 'verifying'
|
||||
au._write_json(updater.pending_file, pending)
|
||||
|
||||
self.config = {'auto_update': {'enabled': enabled}, 'timezone': 'UTC'}
|
||||
cm = MagicMock()
|
||||
cm.load_config.return_value = self.config
|
||||
self.updater = au.AutoUpdater(
|
||||
config_manager=cm, core_update=core_update or real_pull(repo.device),
|
||||
store_manager=store, project_root=repo.device, clock=lambda: self.now,
|
||||
restart=self.restarts.append, run=run,
|
||||
service_active=lambda unit: display_active, helper_ready=lambda: helper,
|
||||
disk_free=lambda path: disk_free, sleep=sleep)
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
return au.load_state(self.updater.state_file)
|
||||
|
||||
@property
|
||||
def pending(self):
|
||||
return au._read_json(self.updater.pending_file)
|
||||
|
||||
def write_pending(self, **fields):
|
||||
au._write_json(self.updater.pending_file, fields)
|
||||
|
||||
|
||||
def stub_updater(tmp_path, outcome='up_to_date', message='ok', store=None, enabled=True, clock=NIGHT):
|
||||
"""For scheduling and plugin tests: the core update is stubbed out."""
|
||||
cm = MagicMock()
|
||||
cm.load_config.return_value = {'auto_update': {'enabled': enabled}, 'timezone': 'UTC'}
|
||||
restarts = []
|
||||
updater = au.AutoUpdater(config_manager=cm, core_update=MagicMock(), store_manager=store,
|
||||
project_root=tmp_path, clock=lambda: clock,
|
||||
restart=restarts.append)
|
||||
updater.update_core = MagicMock(return_value={'outcome': outcome, 'message': message})
|
||||
return updater, restarts
|
||||
|
||||
|
||||
# -- scheduling ---------------------------------------------------------------
|
||||
|
||||
class TestIsDue:
|
||||
def test_not_before_next_due(self):
|
||||
assert not au.is_due(100, 200, 3)
|
||||
|
||||
def test_in_quiet_hours_once_due(self):
|
||||
assert au.is_due(200, 200, 3)
|
||||
|
||||
def test_waits_for_quiet_hours(self):
|
||||
assert not au.is_due(200, 100, 12)
|
||||
|
||||
def test_gives_up_waiting_after_grace(self):
|
||||
assert au.is_due(100 + au.QUIET_HOURS_GRACE_SECONDS, 100, 12)
|
||||
|
||||
|
||||
class TestTick:
|
||||
def test_disabled_never_runs(self, tmp_path):
|
||||
updater, _ = stub_updater(tmp_path, enabled=False)
|
||||
assert updater.tick() is False
|
||||
updater.update_core.assert_not_called()
|
||||
|
||||
def test_enabling_at_noon_waits_for_the_night(self, tmp_path):
|
||||
updater, _ = stub_updater(tmp_path, clock=NOON)
|
||||
assert updater.tick() is False
|
||||
updater.update_core.assert_not_called()
|
||||
assert au.load_state(updater.state_file)['next_due'] == NOON
|
||||
|
||||
def test_runs_when_due_and_schedules_a_week_out(self, tmp_path):
|
||||
updater, _ = stub_updater(tmp_path)
|
||||
assert updater.tick() is True
|
||||
state = au.load_state(updater.state_file)
|
||||
assert state['next_due'] == NIGHT + au.UPDATE_INTERVAL_SECONDS
|
||||
assert updater.tick() is False, "ran twice in one night"
|
||||
|
||||
def test_transient_error_retries_next_day(self, tmp_path):
|
||||
updater, _ = stub_updater(tmp_path, outcome='error', message='no network')
|
||||
updater.tick()
|
||||
state = au.load_state(updater.state_file)
|
||||
assert state['next_due'] == NIGHT + au.RETRY_AFTER_FAILURE_SECONDS
|
||||
assert state['alert']['message'] == 'no network'
|
||||
|
||||
def test_blocked_keeps_weekly_cadence_but_alerts(self, tmp_path):
|
||||
updater, _ = stub_updater(tmp_path, outcome='blocked', message='local edits')
|
||||
updater.tick()
|
||||
state = au.load_state(updater.state_file)
|
||||
assert state['next_due'] == NIGHT + au.UPDATE_INTERVAL_SECONDS
|
||||
assert 'local edits' in state['alert']['message']
|
||||
|
||||
|
||||
# -- checks before touching the checkout ---------------------------------------
|
||||
|
||||
@needs_git
|
||||
class TestPreflightRefuses:
|
||||
def test_nothing_new_is_up_to_date(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
h = Harness(tmp_path, repo)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'up_to_date'
|
||||
assert result['status'] == 'success'
|
||||
assert h.sudo == [] and 'alert' not in h.state
|
||||
|
||||
def test_without_the_health_check_code_is_not_touched(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
repo.publish()
|
||||
h = Harness(tmp_path, repo, helper=False)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert repo.head() == old
|
||||
assert 'health check is set up' in result['core_message']
|
||||
assert h.state['alert']
|
||||
|
||||
def test_local_edits_are_never_stashed(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
repo.publish()
|
||||
(repo.device / 'app.py').write_text('mine\n')
|
||||
h = Harness(tmp_path, repo)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'app.py' in result['core_message']
|
||||
assert repo.head() == old
|
||||
assert (repo.device / 'app.py').read_text() == 'mine\n'
|
||||
assert git(repo.device, 'stash', 'list') == ''
|
||||
|
||||
def test_local_commits_block(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish()
|
||||
(repo.device / 'extra.txt').write_text('x\n')
|
||||
git(repo.device, 'add', 'extra.txt')
|
||||
git(repo.device, 'commit', '-qm', 'local')
|
||||
mine = repo.head()
|
||||
h = Harness(tmp_path, repo)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'local commit' in result['core_message']
|
||||
assert repo.head() == mine
|
||||
|
||||
def test_a_rebase_really_in_progress_blocks(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish('v = 2\n')
|
||||
(repo.device / 'app.py').write_text('v = mine\n')
|
||||
git(repo.device, 'commit', '-qam', 'local')
|
||||
git(repo.device, 'fetch', '-q')
|
||||
stopped = subprocess.run(['git', '-c', 'user.name=t', '-c', 'user.email=t@example.com',
|
||||
'rebase', 'origin/main'], cwd=str(repo.device), capture_output=True, text=True)
|
||||
assert stopped.returncode != 0, "the rebase should stop on the conflict"
|
||||
result = Harness(tmp_path, repo).updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'rebase is in progress' in result['core_message']
|
||||
assert (repo.device / '.git' / 'rebase-merge').exists(), "a live rebase must be left alone"
|
||||
|
||||
def test_an_abandoned_rebase_is_cleared_and_the_update_runs(self, tmp_path):
|
||||
"""Found on a real device: a pull stopped mid-rebase a week earlier,
|
||||
HEAD was later checked out to a branch, and the leftover rebase
|
||||
directory would have blocked every automatic update forever."""
|
||||
repo = Repo(tmp_path)
|
||||
leftover = repo.device / '.git' / 'rebase-merge'
|
||||
leftover.mkdir()
|
||||
(leftover / 'head-name').write_text('refs/heads/main\n')
|
||||
new = repo.publish()
|
||||
result = Harness(tmp_path, repo).updater.run()
|
||||
assert not leftover.exists()
|
||||
assert result['core_outcome'] == 'verifying'
|
||||
assert repo.head() == new
|
||||
|
||||
def test_a_merge_in_progress_blocks(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish()
|
||||
(repo.device / '.git' / 'MERGE_HEAD').write_text(repo.head() + '\n')
|
||||
result = Harness(tmp_path, repo).updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'merge is in progress' in result['core_message']
|
||||
|
||||
def test_low_disk_blocks(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish()
|
||||
result = Harness(tmp_path, repo, disk_free=10 * 1024 * 1024).updater.run()
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'disk space' in result['core_message']
|
||||
|
||||
def test_unreachable_origin_retries_tomorrow(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
git(repo.device, 'remote', 'set-url', 'origin', str(tmp_path / 'nowhere'))
|
||||
h = Harness(tmp_path, repo)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'error'
|
||||
assert h.state['next_due'] == NIGHT + au.RETRY_AFTER_FAILURE_SECONDS
|
||||
|
||||
def test_version_already_rolled_back_is_not_retried(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
bad = repo.publish()
|
||||
h = Harness(tmp_path, repo)
|
||||
au.save_state({'rolled_back_head': bad}, h.updater.state_file)
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'up_to_date'
|
||||
assert 'rolled back before' in result['core_message']
|
||||
assert repo.head() == old and h.sudo == []
|
||||
|
||||
|
||||
# -- the update and its hand-off to the health check ---------------------------
|
||||
|
||||
@needs_git
|
||||
class TestUpdateIsVerified:
|
||||
def test_pull_hands_off_to_the_health_check(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
new = repo.publish()
|
||||
store = FakeStore(tmp_path / 'plugins', {'weather': '1.0.0'}, bump={'weather'})
|
||||
h = Harness(tmp_path, repo, store=store)
|
||||
result = h.updater.run()
|
||||
|
||||
assert result['core_outcome'] == 'verifying' and result['status'] == 'pending'
|
||||
assert repo.head() == new
|
||||
assert h.pending == {**h.pending, 'status': 'verifying', 'old_head': old, 'new_head': new,
|
||||
'display_was_active': True, 'dependency_failures': []}
|
||||
assert not h.updater.request_file.exists(), "the request is consumed when the check starts"
|
||||
assert h.sudo == [], "triggering the health check needs no privilege"
|
||||
assert h.restarts == [], "restarts belong to the health check, not the web process"
|
||||
assert store.updated_calls == [], "plugins must wait until the new code is verified"
|
||||
assert h.state['plugins_pending'] is True
|
||||
assert h.updater.verifier_copy.read_text() == au.VERIFIER_SOURCE.read_text()
|
||||
|
||||
def test_state_is_saved_before_the_health_check_restarts_this_process(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish()
|
||||
h = Harness(tmp_path, repo)
|
||||
h.updater.run()
|
||||
assert h.state_at_handoff['last_result']['core_outcome'] == 'verifying'
|
||||
assert h.state_at_handoff['plugins_pending'] is True
|
||||
|
||||
def test_dependency_failures_reach_the_health_check(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
repo.publish()
|
||||
h = Harness(tmp_path, repo,
|
||||
core_update=real_pull(repo.device, dependency_failures=['requirements.txt']))
|
||||
h.updater.run()
|
||||
assert h.pending['dependency_failures'] == ['requirements.txt']
|
||||
|
||||
def test_a_health_check_that_never_starts_means_the_update_is_undone(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
repo.publish()
|
||||
h = Harness(tmp_path, repo, pickup=False)
|
||||
result = h.updater.run()
|
||||
assert repo.head() == old
|
||||
assert result['core_outcome'] == 'blocked'
|
||||
assert 'did not start' in result['core_message']
|
||||
assert h.pending is None and not h.updater.request_file.exists()
|
||||
assert h.state['alert']
|
||||
|
||||
def test_failed_pull_retries_tomorrow(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
repo.publish()
|
||||
h = Harness(tmp_path, repo,
|
||||
core_update=lambda: {'status': 'error', 'message': 'Update failed: network'})
|
||||
result = h.updater.run()
|
||||
assert result['core_outcome'] == 'error'
|
||||
assert repo.head() == old
|
||||
assert h.state['next_due'] == NIGHT + au.RETRY_AFTER_FAILURE_SECONDS
|
||||
|
||||
def test_failed_pull_that_moved_head_is_rolled_back(self, tmp_path):
|
||||
repo = Repo(tmp_path)
|
||||
old = repo.head()
|
||||
repo.publish()
|
||||
pull = real_pull(repo.device)
|
||||
|
||||
def half_failed():
|
||||
pull()
|
||||
return {'status': 'error', 'message': 'Update failed: interrupted'}
|
||||
result = Harness(tmp_path, repo, core_update=half_failed).updater.run()
|
||||
assert repo.head() == old
|
||||
assert 'rolled back' in result['core_message']
|
||||
|
||||
|
||||
# -- reporting the health check's outcome --------------------------------------
|
||||
|
||||
@needs_git
|
||||
class TestHealthCheckOutcome:
|
||||
def _after_update(self, tmp_path, **pending):
|
||||
repo = Repo(tmp_path)
|
||||
store = FakeStore(tmp_path / 'plugins', {'weather': '1.0.0'}, bump={'weather'})
|
||||
h = Harness(tmp_path, repo, store=store)
|
||||
au.save_state({'plugins_pending': True, 'next_due': NIGHT + au.UPDATE_INTERVAL_SECONDS,
|
||||
'last_result': {'core_outcome': 'verifying', 'status': 'pending',
|
||||
'plugins_deferred': True}}, h.updater.state_file)
|
||||
h.write_pending(old_head='a' * 40, new_head='b' * 40, created_at=NIGHT, **pending)
|
||||
return h, store
|
||||
|
||||
def test_waits_while_the_check_runs(self, tmp_path):
|
||||
h, store = self._after_update(tmp_path, status='verifying')
|
||||
h.now = NIGHT + 60
|
||||
assert h.updater.tick() is False
|
||||
assert store.updated_calls == [] and h.pending
|
||||
|
||||
def test_success_then_runs_the_deferred_plugins(self, tmp_path):
|
||||
h, store = self._after_update(tmp_path, status='success')
|
||||
assert h.updater.tick() is True
|
||||
result = h.state['last_result']
|
||||
assert result['core_outcome'] == 'updated' and result['status'] == 'success'
|
||||
assert result['plugins_updated'] == ['weather']
|
||||
assert h.restarts == ['ledmatrix']
|
||||
assert 'alert' not in h.state and h.pending is None
|
||||
|
||||
def test_rollback_alerts_and_remembers_the_bad_commit(self, tmp_path):
|
||||
h, store = self._after_update(tmp_path, status='rolled_back',
|
||||
reason='the web interface did not respond')
|
||||
h.updater.tick()
|
||||
assert 'rolled back' in h.state['alert']['message']
|
||||
assert 'the web interface did not respond' in h.state['alert']['message']
|
||||
assert h.state['rolled_back_head'] == 'b' * 40
|
||||
assert store.updated_calls == ['weather'], "the old code is healthy; plugins may update"
|
||||
|
||||
def test_failed_rollback_alerts_and_skips_plugins(self, tmp_path):
|
||||
h, store = self._after_update(tmp_path, status='rollback_failed',
|
||||
reason='display down', detail='reset failed')
|
||||
h.updater.tick()
|
||||
assert f'git reset --hard {"a" * 40}' in h.state['alert']['message']
|
||||
assert store.updated_calls == []
|
||||
assert h.state['plugins_pending'] is False
|
||||
|
||||
def test_a_check_that_never_reports_back_alerts(self, tmp_path):
|
||||
h, _ = self._after_update(tmp_path, status='verifying')
|
||||
h.now = NIGHT + au.VERIFY_LOST_SECONDS + 1
|
||||
h.updater.tick()
|
||||
assert h.state['last_result']['core_outcome'] == 'lost'
|
||||
assert 'never reported back' in h.state['alert']['message']
|
||||
|
||||
def test_dismissed_alert_stays_hidden_until_a_new_one(self, tmp_path):
|
||||
h, _ = self._after_update(tmp_path, status='rolled_back', reason='x')
|
||||
h.updater.tick()
|
||||
alert_id = h.state['alert']['id']
|
||||
|
||||
def status():
|
||||
return au.describe_status(h.config, state_file=h.updater.state_file,
|
||||
pending_file=h.updater.pending_file,
|
||||
setup_file=tmp_path / 'no-setup.json', helper=lambda: True)
|
||||
assert status()['alert_id'] == alert_id
|
||||
au.dismiss_alert(alert_id, h.updater.state_file)
|
||||
assert status()['alert'] is None
|
||||
state = h.state
|
||||
state['alert'] = {'id': 'newer', 'message': 'again'}
|
||||
au.save_state(state, h.updater.state_file)
|
||||
assert status()['alert'] == 'again'
|
||||
|
||||
|
||||
def test_describe_status_reports_setup(tmp_path):
|
||||
setup = tmp_path / 'setup.json'
|
||||
setup.write_text(json.dumps({'status': 'failed', 'message': 'not root'}))
|
||||
kwargs = dict(state={}, state_file=tmp_path / 's.json', pending_file=tmp_path / 'p.json',
|
||||
setup_file=setup)
|
||||
on = au.describe_status(ON, helper=lambda: False, **kwargs)
|
||||
assert on['verifier_installed'] is False
|
||||
assert (on['setup_status'], on['setup_message']) == ('failed', 'not root')
|
||||
|
||||
def never(): raise AssertionError("systemctl asked while updates are off")
|
||||
assert au.describe_status(OFF, helper=never, **kwargs)['verifier_installed'] is None
|
||||
|
||||
|
||||
# -- setting it up from the web UI ----------------------------------------------
|
||||
|
||||
class TestSetupFromTheWebUI:
|
||||
def _host(self, monkeypatch, ready=False, active=True, restart_ok=True):
|
||||
restarts = []
|
||||
monkeypatch.setattr(au, 'helper_ready', lambda: ready)
|
||||
monkeypatch.setattr(au, '_service_active', lambda unit: active)
|
||||
monkeypatch.setattr(au, 'restart_service', lambda unit: restarts.append(unit) or restart_ok)
|
||||
return restarts
|
||||
|
||||
def test_switching_on_restarts_the_display_to_finish_setup(self, monkeypatch):
|
||||
restarts = self._host(monkeypatch)
|
||||
note = au.start_setup_if_needed(False, ON)
|
||||
assert restarts == ['ledmatrix']
|
||||
assert 'display is restarting' in note
|
||||
|
||||
@pytest.mark.parametrize('was_enabled, config, ready', [
|
||||
(True, ON, False), # already on: an unrelated save must not restart the display
|
||||
(False, OFF, False), # still off
|
||||
(False, ON, True), # already set up
|
||||
])
|
||||
def test_nothing_to_do(self, monkeypatch, was_enabled, config, ready):
|
||||
restarts = self._host(monkeypatch, ready=ready)
|
||||
assert au.start_setup_if_needed(was_enabled, config) is None
|
||||
assert restarts == []
|
||||
|
||||
def test_a_stopped_display_is_not_started(self, monkeypatch):
|
||||
restarts = self._host(monkeypatch, active=False)
|
||||
assert 'next time the display service starts' in au.start_setup_if_needed(False, ON)
|
||||
assert restarts == []
|
||||
|
||||
def test_a_failed_restart_says_so(self, monkeypatch):
|
||||
self._host(monkeypatch, restart_ok=False)
|
||||
assert 'Could not restart' in au.start_setup_if_needed(False, ON)
|
||||
|
||||
|
||||
# -- plugins --------------------------------------------------------------------
|
||||
|
||||
class TestPlugins:
|
||||
def test_nothing_changed_restarts_nothing(self, tmp_path):
|
||||
store = FakeStore(tmp_path / 'plugins', {'clock': '1.0.0'})
|
||||
updater, restarts = stub_updater(tmp_path, store=store)
|
||||
updater.run()
|
||||
assert store.updated_calls == ['clock']
|
||||
assert restarts == []
|
||||
|
||||
def test_plugin_update_restarts_display_only(self, tmp_path):
|
||||
store = FakeStore(tmp_path / 'plugins', {'clock': '1.0.0', 'weather': '1.0.0'},
|
||||
bump={'weather'})
|
||||
updater, restarts = stub_updater(tmp_path, store=store)
|
||||
result = updater.run()
|
||||
assert result['plugins_updated'] == ['weather']
|
||||
assert restarts == ['ledmatrix']
|
||||
|
||||
def test_plugin_failure_alerts_but_keeps_weekly_cadence(self, tmp_path):
|
||||
store = FakeStore(tmp_path / 'plugins', {'zip-only': '1.0.0'}, fail={'zip-only'})
|
||||
updater, _ = stub_updater(tmp_path, store=store)
|
||||
result = updater.run()
|
||||
assert result['plugins_failed'] == ['zip-only'] and result['status'] == 'error'
|
||||
state = au.load_state(updater.state_file)
|
||||
assert 'zip-only' in state['alert']['message']
|
||||
assert state['next_due'] == NIGHT + au.UPDATE_INTERVAL_SECONDS
|
||||
|
||||
def test_local_only_plugins_are_left_alone(self, tmp_path):
|
||||
store = FakeStore(tmp_path / 'plugins', {'mine': '1.0.0'})
|
||||
path = tmp_path / 'plugins' / 'mine' / 'manifest.json'
|
||||
path.write_text(json.dumps({'id': 'mine', 'version': '1.0.0', 'local_only': True}))
|
||||
updater, _ = stub_updater(tmp_path, store=store)
|
||||
updater.run()
|
||||
assert store.updated_calls == []
|
||||
|
||||
|
||||
def test_restart_service_skips_a_stopped_unit(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def run(args, **kwargs):
|
||||
calls.append(args)
|
||||
return MagicMock(stdout='inactive\n', returncode=3, stderr='')
|
||||
monkeypatch.setattr(au.subprocess, 'run', run)
|
||||
assert au.restart_service('ledmatrix') is False
|
||||
assert calls == [['systemctl', 'is-active', 'ledmatrix']], (
|
||||
"a display the user stopped must not be started by an update")
|
||||
|
||||
|
||||
def test_core_update_names_the_requirement_files_that_failed():
|
||||
"""The health check refuses code whose dependencies did not install, so
|
||||
the failure has to arrive as data, not only as a sentence in the message."""
|
||||
from web_interface.blueprints import api_v3 as pkg
|
||||
from web_interface.blueprints.api_v3 import system
|
||||
|
||||
heads = iter(['aaa111\n', 'bbb222\n'])
|
||||
|
||||
def run(args, **kwargs):
|
||||
if args[:2] == ['git', 'rev-parse'] and args[-1] == 'HEAD':
|
||||
out = next(heads, 'bbb222\n')
|
||||
elif args[:2] == ['git', 'diff']:
|
||||
out = 'requirements.txt\n'
|
||||
elif '@{u}' in args:
|
||||
out = 'origin/main\n'
|
||||
else:
|
||||
out = ''
|
||||
return subprocess.CompletedProcess(args, 0, stdout=out, stderr='')
|
||||
|
||||
pkg.api_v3.plugin_store_manager = None
|
||||
failed_install = subprocess.CompletedProcess([], 1, stdout='', stderr='boom')
|
||||
with patch.object(pkg.subprocess, 'run', run), \
|
||||
patch.object(system, '_pip_install_requirements', return_value=failed_install):
|
||||
result = system.perform_core_update()
|
||||
assert result['status'] == 'success'
|
||||
assert result['dependency_failures'] == ['requirements.txt']
|
||||
|
||||
|
||||
# -- web routes -------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def api_client(monkeypatch):
|
||||
from flask import Flask
|
||||
from web_interface.blueprints.api_v3 import api_v3
|
||||
app = Flask(__name__)
|
||||
app.register_blueprint(api_v3, url_prefix='/api/v3')
|
||||
cm = MagicMock()
|
||||
cm.load_config.return_value = {'timezone': 'UTC', 'auto_update': {'enabled': True}}
|
||||
cm.get_raw_file_content.return_value = {}
|
||||
cm.save_config_atomic.return_value = MagicMock(status=MagicMock(value='success'), message=None)
|
||||
api_v3.config_manager = cm
|
||||
api_v3.plugin_manager = MagicMock(plugins={})
|
||||
# Never restart a real display from a test run.
|
||||
setup_calls = []
|
||||
monkeypatch.setattr(au, 'start_setup_if_needed',
|
||||
lambda was_enabled, config: setup_calls.append(was_enabled) or None)
|
||||
return app.test_client(), cm, setup_calls
|
||||
|
||||
|
||||
class TestStatusRoutes:
|
||||
def test_status_carries_the_alert(self, api_client, monkeypatch):
|
||||
client, _, _ = api_client
|
||||
monkeypatch.setattr(au, 'describe_status', lambda config: {'alert': 'rolled back', 'alert_id': '7'})
|
||||
data = client.get('/api/v3/system/auto-update').get_json()['data']
|
||||
assert data == {'alert': 'rolled back', 'alert_id': '7'}
|
||||
|
||||
@pytest.mark.parametrize('body', [{}, [1], 'x', 5])
|
||||
def test_dismiss_needs_an_id_in_a_json_object(self, api_client, body):
|
||||
client, _, _ = api_client
|
||||
assert client.post('/api/v3/system/auto-update/dismiss', json=body).status_code == 400
|
||||
|
||||
def test_dismiss(self, api_client, monkeypatch):
|
||||
client, _, _ = api_client
|
||||
calls = []
|
||||
monkeypatch.setattr(au, 'dismiss_alert', calls.append)
|
||||
assert client.post('/api/v3/system/auto-update/dismiss', json={'alert_id': '7'}).status_code == 200
|
||||
assert calls == ['7']
|
||||
|
||||
|
||||
class TestSettingsSave:
|
||||
def _saved(self, cm):
|
||||
return cm.save_config_atomic.call_args[0][0]
|
||||
|
||||
def test_checked_toggle_saves_enabled(self, api_client):
|
||||
client, cm, setup_calls = api_client
|
||||
cm.load_config.return_value = {'timezone': 'UTC'}
|
||||
resp = client.post('/api/v3/config/main', json={'timezone': 'UTC', 'auto_update_enabled': 'true'})
|
||||
assert resp.status_code == 200
|
||||
saved = self._saved(cm)
|
||||
assert saved['auto_update'] == {'enabled': True}
|
||||
assert 'auto_update_enabled' not in saved, "the form field must not leak into config.json"
|
||||
assert setup_calls == [False], "setup is told the toggle was previously off"
|
||||
|
||||
def test_unchecked_toggle_on_general_save_disables(self, api_client):
|
||||
client, cm, setup_calls = api_client
|
||||
resp = client.post('/api/v3/config/main', json={'timezone': 'UTC'})
|
||||
assert resp.status_code == 200
|
||||
assert self._saved(cm)['auto_update'] == {'enabled': False}
|
||||
assert setup_calls == [True]
|
||||
|
||||
def test_setup_note_reaches_the_user(self, api_client, monkeypatch):
|
||||
client, cm, _ = api_client
|
||||
cm.load_config.return_value = {'timezone': 'UTC'}
|
||||
monkeypatch.setattr(au, 'start_setup_if_needed',
|
||||
lambda was_enabled, config: 'Finishing automatic update setup: the display is restarting.')
|
||||
body = client.post('/api/v3/config/main',
|
||||
json={'timezone': 'UTC', 'auto_update_enabled': 'true'}).get_json()
|
||||
assert 'the display is restarting' in body['message']
|
||||
@@ -0,0 +1,214 @@
|
||||
"""Installing the automatic-update health check from the display service.
|
||||
|
||||
src/auto_update_setup.py is how a user who never opens a terminal gets updates
|
||||
with a safety net: it runs as root inside the display service and writes
|
||||
systemd units. So it has to install exactly the right thing when asked, do
|
||||
nothing when not asked, refuse templates that would run as anyone but the web
|
||||
user, and say why whenever it could not finish.
|
||||
"""
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from src import auto_update_setup as aus # noqa: E402
|
||||
|
||||
ON = {'auto_update': {'enabled': True}}
|
||||
|
||||
|
||||
class FakeSystemctl:
|
||||
def __init__(self, fail=(), activates=True):
|
||||
self.calls = []
|
||||
self.fail = set(fail)
|
||||
self.activates = activates
|
||||
self.active = False
|
||||
|
||||
def __call__(self, args, **kwargs):
|
||||
assert args[0] == 'systemctl', args
|
||||
self.calls.append(args[1:])
|
||||
verb = args[1]
|
||||
if verb == 'is-active':
|
||||
return subprocess.CompletedProcess(args, 0 if self.active else 3,
|
||||
stdout='active\n' if self.active else 'inactive\n', stderr='')
|
||||
if verb in self.fail:
|
||||
return subprocess.CompletedProcess(args, 1, stdout='', stderr=f'{verb} refused')
|
||||
if verb in ('restart', 'enable') and self.activates and (verb == 'restart' or '--now' in args):
|
||||
self.active = True
|
||||
return subprocess.CompletedProcess(args, 0, stdout='', stderr='')
|
||||
|
||||
|
||||
def project(tmp_path, user='hdpi', workdir=None, name='LEDMatrix'):
|
||||
root = tmp_path / name
|
||||
(root / 'systemd').mkdir(parents=True)
|
||||
for name in aus.UNITS:
|
||||
shutil.copy(ROOT / 'systemd' / name, root / 'systemd' / name)
|
||||
etc = tmp_path / 'etc'
|
||||
etc.mkdir()
|
||||
(etc / aus.WEB_UNIT).write_text(
|
||||
f'[Service]\nUser={user}\nWorkingDirectory={workdir or root}\n', encoding='utf-8')
|
||||
return root, etc
|
||||
|
||||
|
||||
def setup(root, etc, systemctl, root_user=True):
|
||||
return aus.UpdateHelperSetup(project_root=root, systemd_dir=etc, run=systemctl,
|
||||
is_root=lambda: root_user,
|
||||
lookup_ids=lambda user: None if user == 'ghost' else (1000, 1000),
|
||||
clock=lambda: 1234.0)
|
||||
|
||||
|
||||
def result(root):
|
||||
return json.loads((root / aus.RESULT_REL).read_text())
|
||||
|
||||
|
||||
def test_does_nothing_while_updates_are_off(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
systemctl = FakeSystemctl()
|
||||
assert setup(root, etc, systemctl).ensure({'auto_update': {'enabled': False}}) is None
|
||||
assert systemctl.calls == []
|
||||
assert not (etc / aus.SERVICE_UNIT).exists()
|
||||
assert not (root / aus.RESULT_REL).exists()
|
||||
|
||||
|
||||
def test_installs_both_units_for_the_web_user(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
systemctl = FakeSystemctl()
|
||||
out = setup(root, etc, systemctl).ensure(ON)
|
||||
|
||||
assert out['status'] == 'installed' and result(root)['status'] == 'installed'
|
||||
service = (etc / aus.SERVICE_UNIT).read_text()
|
||||
assert 'User=hdpi' in service
|
||||
assert f'ExecStart=/usr/bin/python3 "{root}/data/auto_update_verifier.py" "{root}"' in service
|
||||
assert f'PathExists={root}/data/auto_update_verify.request' in (etc / aus.PATH_UNIT).read_text()
|
||||
assert '__' not in service
|
||||
assert ['daemon-reload'] in systemctl.calls
|
||||
assert ['enable', aus.PATH_UNIT] in systemctl.calls
|
||||
assert systemctl.active
|
||||
|
||||
|
||||
def test_second_start_changes_nothing(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
systemctl = FakeSystemctl()
|
||||
setup(root, etc, systemctl).ensure(ON)
|
||||
first = result(root)
|
||||
systemctl.calls.clear()
|
||||
setup(root, etc, systemctl).ensure(ON)
|
||||
assert systemctl.calls == [['is-active', aus.PATH_UNIT], ['is-active', aus.PATH_UNIT]]
|
||||
assert result(root) == first, "an unchanged setup must not rewrite its result every boot"
|
||||
|
||||
|
||||
def test_a_changed_template_is_reinstalled(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
systemctl = FakeSystemctl()
|
||||
setup(root, etc, systemctl).ensure(ON)
|
||||
template = root / 'systemd' / aus.SERVICE_UNIT
|
||||
template.write_text(template.read_text() + '\n# newer\n')
|
||||
systemctl.calls.clear()
|
||||
assert setup(root, etc, systemctl).ensure(ON)['message'] == 'Installed the update health check.'
|
||||
assert (etc / aus.SERVICE_UNIT).read_text().endswith('# newer\n')
|
||||
assert ['daemon-reload'] in systemctl.calls
|
||||
|
||||
|
||||
def test_a_template_that_would_not_run_as_the_web_user_is_refused(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
template = root / 'systemd' / aus.SERVICE_UNIT
|
||||
template.write_text(template.read_text().replace('User=__USER__', 'User=root'))
|
||||
systemctl = FakeSystemctl()
|
||||
out = setup(root, etc, systemctl).ensure(ON)
|
||||
assert out['status'] == 'failed' and 'refusing' in out['message']
|
||||
assert not (etc / aus.SERVICE_UNIT).exists()
|
||||
assert systemctl.calls == []
|
||||
|
||||
|
||||
def test_a_path_unit_that_starts_something_else_is_refused(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
template = root / 'systemd' / aus.PATH_UNIT
|
||||
template.write_text(template.read_text().replace('Unit=ledmatrix-update-verify.service', 'Unit=other.service'))
|
||||
out = setup(root, etc, FakeSystemctl()).ensure(ON)
|
||||
assert out['status'] == 'failed' and not (etc / aus.PATH_UNIT).exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('case, expected', [
|
||||
('not_root', 'not running as root'),
|
||||
('no_web_unit', 'not installed'),
|
||||
('other_folder', 'runs from'),
|
||||
('bad_user', 'not a usable account'),
|
||||
])
|
||||
def test_reports_why_it_could_not_install(tmp_path, case, expected):
|
||||
root, etc = project(tmp_path, user='ghost' if case == 'bad_user' else 'hdpi',
|
||||
workdir=tmp_path / 'elsewhere' if case == 'other_folder' else None)
|
||||
if case == 'no_web_unit':
|
||||
(etc / aus.WEB_UNIT).unlink()
|
||||
out = setup(root, etc, FakeSystemctl(), root_user=case != 'not_root').ensure(ON)
|
||||
assert out['status'] == 'failed' and expected in out['message']
|
||||
assert result(root)['message'] == out['message']
|
||||
assert not (etc / aus.SERVICE_UNIT).exists()
|
||||
|
||||
|
||||
def test_a_path_unit_that_will_not_start_is_a_failure(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
out = setup(root, etc, FakeSystemctl(activates=False)).ensure(ON)
|
||||
assert out['status'] == 'failed' and 'did not start' in out['message']
|
||||
|
||||
|
||||
def test_systemctl_errors_are_reported(tmp_path):
|
||||
root, etc = project(tmp_path)
|
||||
out = setup(root, etc, FakeSystemctl(fail={'daemon-reload'})).ensure(ON)
|
||||
assert out['status'] == 'failed' and 'daemon-reload refused' in out['message']
|
||||
|
||||
|
||||
def test_not_a_systemd_host_is_left_alone(tmp_path):
|
||||
root, _ = project(tmp_path)
|
||||
systemctl = FakeSystemctl()
|
||||
assert setup(root, tmp_path / 'no-etc', systemctl).ensure(ON) is None
|
||||
assert systemctl.calls == []
|
||||
|
||||
|
||||
def test_a_folder_with_spaces_is_quoted_in_the_commands(tmp_path):
|
||||
root, etc = project(tmp_path, name='LED Matrix')
|
||||
assert setup(root, etc, FakeSystemctl()).ensure(ON)['status'] == 'installed'
|
||||
service = (etc / aus.SERVICE_UNIT).read_text()
|
||||
assert f'ExecStartPre=/bin/rm -f "{root}/data/auto_update_verify.request"' in service
|
||||
assert f'ExecStart=/usr/bin/python3 "{root}/data/auto_update_verifier.py" "{root}"' in service
|
||||
|
||||
|
||||
def test_a_folder_name_systemd_would_reinterpret_is_refused(tmp_path):
|
||||
root, etc = project(tmp_path, name='LED%Matrix')
|
||||
systemctl = FakeSystemctl()
|
||||
out = setup(root, etc, systemctl).ensure(ON)
|
||||
assert out['status'] == 'failed' and 'systemd cannot use' in out['message']
|
||||
assert not (etc / aus.SERVICE_UNIT).exists() and systemctl.calls == []
|
||||
|
||||
|
||||
def test_installer_sets_the_toggle_without_a_half_written_config(tmp_path):
|
||||
"""first_time_install.sh may run while the display service watches
|
||||
config.json; the file must be replaced whole, never truncated in place."""
|
||||
import re
|
||||
text = (ROOT / 'first_time_install.sh').read_text(encoding='utf-8').replace('\r\n', '\n')
|
||||
script = next(s for s in re.findall(r"<<'PY'\n(.*?)\nPY\n", text, re.S) if 'auto_update' in s)
|
||||
assert 'os.replace(' in script
|
||||
config = tmp_path / 'config.json'
|
||||
config.write_text(json.dumps({'timezone': 'UTC', 'auto_update': {'enabled': False}}))
|
||||
run = subprocess.run([sys.executable, '-', str(config), '1'], input=script, text=True,
|
||||
capture_output=True)
|
||||
assert run.returncode == 0, run.stderr
|
||||
assert json.loads(config.read_text()) == {'timezone': 'UTC', 'auto_update': {'enabled': True}}
|
||||
assert [p.name for p in tmp_path.iterdir()] == ['config.json'], "a temp file was left behind"
|
||||
|
||||
|
||||
def test_the_result_is_kept_when_it_cannot_be_given_to_the_web_user(tmp_path, monkeypatch):
|
||||
"""Caught by CI, which runs as a non-root Linux user: chown needs root, and
|
||||
a failed chown used to discard the result, so the General tab never
|
||||
learned whether setup worked."""
|
||||
def refuse(*args):
|
||||
raise PermissionError('Operation not permitted')
|
||||
monkeypatch.setattr(aus.os, 'chown', refuse, raising=False)
|
||||
root, etc = project(tmp_path)
|
||||
out = setup(root, etc, FakeSystemctl()).ensure(ON)
|
||||
assert out['status'] == 'installed'
|
||||
assert result(root) == out
|
||||
@@ -0,0 +1,248 @@
|
||||
"""The automatic update's health check and rollback (scripts/utils/auto_update_verify.py).
|
||||
|
||||
This is what stands between an unattended update and a device that no longer
|
||||
works, so each way an update can fail is exercised against a real git repo:
|
||||
the new commit is checked out, the services are "restarted", and whether they
|
||||
come up healthy depends on which commit they were restarted onto. Only
|
||||
systemd, sudo and the HTTP check are faked.
|
||||
"""
|
||||
import importlib.util
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
'auto_update_verify', ROOT / 'scripts' / 'utils' / 'auto_update_verify.py')
|
||||
av = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(av)
|
||||
|
||||
pytestmark = pytest.mark.skipif(shutil.which('git') is None, reason='git not installed')
|
||||
|
||||
|
||||
def git(cwd, *args):
|
||||
result = subprocess.run(['git', '-c', 'user.name=t', '-c', 'user.email=t@example.com', *args],
|
||||
cwd=str(cwd), capture_output=True, text=True)
|
||||
assert result.returncode == 0, result.stderr
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def done(args, stdout='', rc=0):
|
||||
return subprocess.CompletedProcess(args, rc, stdout=stdout, stderr='' if rc == 0 else 'failed')
|
||||
|
||||
|
||||
def updated_repo(tmp_path, new_requirements=False):
|
||||
"""A checkout on a new commit, with the commit it came from."""
|
||||
repo = tmp_path / 'LEDMatrix'
|
||||
repo.mkdir()
|
||||
git(repo, 'init', '-q', '-b', 'main')
|
||||
(repo / 'app.py').write_text('v = 1\n')
|
||||
(repo / 'requirements.txt').write_text('requests\n')
|
||||
git(repo, 'add', '.')
|
||||
git(repo, 'commit', '-qm', 'old')
|
||||
old = git(repo, 'rev-parse', 'HEAD')
|
||||
(repo / 'app.py').write_text('v = 2\n')
|
||||
if new_requirements:
|
||||
(repo / 'requirements.txt').write_text('requests\nnewthing\n')
|
||||
git(repo, 'commit', '-qam', 'new')
|
||||
return repo, old, git(repo, 'rev-parse', 'HEAD')
|
||||
|
||||
|
||||
class FakeHost:
|
||||
"""systemd, sudo and the web interface.
|
||||
|
||||
Services run whatever commit was checked out when they were last
|
||||
restarted; ``failure`` says how they misbehave on the new commit
|
||||
("display_down", "web_down", "crash_loop") or on any commit ("always").
|
||||
"""
|
||||
|
||||
def __init__(self, repo, bad_head, failure=None, pip_ok=True, restart_failures=0, count_readable=True):
|
||||
self.repo, self.bad_head, self.failure, self.pip_ok = repo, bad_head, failure, pip_ok
|
||||
self.restart_failures = restart_failures # how many restart commands fail, first to last
|
||||
self.count_readable = count_readable
|
||||
self.running_head = None # not restarted yet: still the old, healthy code
|
||||
self.restarts = [] # (unit, commit it was restarted onto)
|
||||
self.pip_installs = []
|
||||
self.now = 0.0
|
||||
self.nrestarts = 0
|
||||
|
||||
def broken(self, kind):
|
||||
if self.running_head is None:
|
||||
return False
|
||||
return self.failure == 'always' or (self.running_head == self.bad_head and self.failure == kind)
|
||||
|
||||
def run(self, args, **kwargs):
|
||||
if args[0] == 'git':
|
||||
return subprocess.run(args, **kwargs)
|
||||
if args[:2] == ['systemctl', 'is-active']:
|
||||
return done(args, 'failed\n' if self.broken('display_down') else 'active\n')
|
||||
if args[:2] == ['systemctl', 'show']:
|
||||
if not self.count_readable:
|
||||
return done(args, rc=1)
|
||||
if self.broken('crash_loop'):
|
||||
self.nrestarts += 1
|
||||
return done(args, f'{self.nrestarts}\n')
|
||||
if args[0] == 'sudo' and any(a.endswith('safe_pip_install.sh') for a in args):
|
||||
self.pip_installs.append(args[-1])
|
||||
return done(args, rc=0 if self.pip_ok else 1)
|
||||
if args[:4] == ['sudo', '-n', 'systemctl', 'restart']:
|
||||
if self.restart_failures:
|
||||
self.restart_failures -= 1
|
||||
return done(args, rc=1) # the old process keeps running
|
||||
self.running_head = git(self.repo, 'rev-parse', 'HEAD')
|
||||
self.restarts.append((args[4], self.running_head))
|
||||
return done(args)
|
||||
raise AssertionError(f'unexpected command: {args}')
|
||||
|
||||
def web_responds(self):
|
||||
return not self.broken('web_down')
|
||||
|
||||
def sleep(self, seconds):
|
||||
self.now += seconds
|
||||
|
||||
def verifier(self):
|
||||
return av.Verifier(self.repo, run=self.run, sleep=self.sleep, clock=lambda: self.now,
|
||||
web_responds=self.web_responds, log=lambda msg: None)
|
||||
|
||||
|
||||
def check(tmp_path, failure=None, new_requirements=False, pip_ok=True, restart_failures=0,
|
||||
count_readable=True, **pending):
|
||||
repo, old, new = updated_repo(tmp_path, new_requirements)
|
||||
fields = {'status': 'pending', 'old_head': old, 'new_head': new,
|
||||
'display_was_active': True, 'dependency_failures': []}
|
||||
fields.update(pending)
|
||||
av.write_pending(av.pending_path(repo), fields)
|
||||
host = FakeHost(repo, new, failure, pip_ok, restart_failures, count_readable)
|
||||
code = host.verifier().verify()
|
||||
result = av.read_pending(av.pending_path(repo))
|
||||
return code, result, host, git(repo, 'rev-parse', 'HEAD'), old, new
|
||||
|
||||
|
||||
def test_a_healthy_update_is_kept(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path)
|
||||
assert code == 0 and result['status'] == 'success'
|
||||
assert head == new
|
||||
assert host.restarts == [('ledmatrix.service', new), ('ledmatrix-web.service', new)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure, reason', [
|
||||
('display_down', 'the display service did not stay running'),
|
||||
('web_down', 'the web interface did not respond'),
|
||||
('crash_loop', 'the display service kept restarting'),
|
||||
])
|
||||
def test_an_unhealthy_update_is_rolled_back(tmp_path, failure, reason):
|
||||
code, result, host, head, old, new = check(tmp_path, failure)
|
||||
assert code == 0
|
||||
assert result['status'] == 'rolled_back' and result['reason'] == reason
|
||||
assert head == old
|
||||
assert host.restarts[-2:] == [('ledmatrix.service', old), ('ledmatrix-web.service', old)]
|
||||
|
||||
|
||||
def test_a_stopped_display_is_not_started(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, display_was_active=False)
|
||||
assert result['status'] == 'success'
|
||||
assert [unit for unit, _ in host.restarts] == ['ledmatrix-web.service']
|
||||
|
||||
|
||||
def test_failed_dependencies_roll_back_before_anything_restarts_onto_them(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, dependency_failures=['requirements.txt'])
|
||||
assert result['status'] == 'rolled_back'
|
||||
assert 'requirements.txt' in result['reason']
|
||||
assert head == old
|
||||
assert all(commit == old for _, commit in host.restarts), host.restarts
|
||||
|
||||
|
||||
def test_rollback_reinstalls_the_previous_dependencies(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, 'display_down', new_requirements=True)
|
||||
assert result['status'] == 'rolled_back' and result['detail'] is None
|
||||
assert host.pip_installs == [str(tmp_path / 'LEDMatrix' / 'requirements.txt')]
|
||||
|
||||
|
||||
def test_a_failed_dependency_reinstall_is_reported(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, 'display_down', new_requirements=True,
|
||||
pip_ok=False)
|
||||
assert result['status'] == 'rolled_back' and head == old
|
||||
assert 'Install Base Requirements' in result['detail']
|
||||
|
||||
|
||||
def test_still_broken_after_rolling_back_is_rollback_failed(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, 'always')
|
||||
assert code == 1 and result['status'] == 'rollback_failed'
|
||||
assert 'still unhealthy after rolling back' in result['detail']
|
||||
|
||||
|
||||
def test_a_failed_restart_is_not_mistaken_for_a_healthy_update(tmp_path):
|
||||
"""When the restart command itself fails the old process keeps answering,
|
||||
so checking it would pass an update that never started."""
|
||||
code, result, host, head, old, new = check(tmp_path, restart_failures=1)
|
||||
assert result['status'] == 'rolled_back'
|
||||
assert result['reason'] == 'restarting the services failed'
|
||||
assert head == old
|
||||
|
||||
|
||||
def test_restarts_that_keep_failing_after_rollback_are_rollback_failed(tmp_path):
|
||||
code, result, host, head, old, new = check(tmp_path, restart_failures=100)
|
||||
assert code == 1 and result['status'] == 'rollback_failed'
|
||||
assert 'restarting the services failed' in result['detail']
|
||||
|
||||
|
||||
def test_an_unreadable_restart_count_is_never_called_stable(tmp_path):
|
||||
"""With no restart count a crash loop looks healthy between attempts."""
|
||||
code, result, host, head, old, new = check(tmp_path, count_readable=False)
|
||||
assert result['status'] != 'success'
|
||||
assert 'restart count could not be read' in result['reason']
|
||||
|
||||
|
||||
def test_nothing_pending_does_nothing(tmp_path):
|
||||
repo, _, _ = updated_repo(tmp_path)
|
||||
host = FakeHost(repo, None)
|
||||
assert host.verifier().verify() == 0
|
||||
assert host.restarts == []
|
||||
|
||||
|
||||
def test_a_crash_is_recorded_not_left_as_running(tmp_path, monkeypatch):
|
||||
repo, old, new = updated_repo(tmp_path)
|
||||
av.write_pending(av.pending_path(repo), {'status': 'pending', 'old_head': old, 'new_head': new})
|
||||
|
||||
def boom(self):
|
||||
raise RuntimeError('boom')
|
||||
monkeypatch.setattr(av.Verifier, 'verify', boom)
|
||||
assert av.main(['auto_update_verify.py', str(repo)]) == 1
|
||||
result = av.read_pending(av.pending_path(repo))
|
||||
assert result['status'] == 'rollback_failed' and result['detail'] == 'boom'
|
||||
|
||||
|
||||
def test_units_installers_and_updater_agree():
|
||||
"""The request file, the copied verifier, the unit names and the places
|
||||
that install them must all line up, or the health check silently never
|
||||
starts on real devices -- and code updates stay paused."""
|
||||
from web_interface import auto_update as au
|
||||
from src import auto_update_setup as aus
|
||||
from src.startup_validator import StartupValidator
|
||||
|
||||
service = (ROOT / 'systemd' / aus.SERVICE_UNIT).read_text(encoding='utf-8')
|
||||
path = (ROOT / 'systemd' / aus.PATH_UNIT).read_text(encoding='utf-8')
|
||||
request = f'__PROJECT_ROOT_DIR__/{au.REQUEST_REL.as_posix()}'
|
||||
assert f'PathExists={request}' in path
|
||||
assert f'Unit={aus.SERVICE_UNIT}' in path
|
||||
assert f'ExecStartPre=/bin/rm -f "{request}"' in service, "the request must be consumed or the path unit re-fires"
|
||||
assert f'"__PROJECT_ROOT_DIR__/{au.VERIFIER_COPY_REL.as_posix()}" "__PROJECT_ROOT_DIR__"' in service
|
||||
assert au.PENDING_REL.name == av.PENDING_NAME
|
||||
assert au.PATH_UNIT == aus.PATH_UNIT and au.SETUP_RESULT_REL == aus.RESULT_REL
|
||||
|
||||
for installer in ('scripts/install/install_web_service.sh', 'scripts/install/install_service.sh'):
|
||||
text = (ROOT / installer).read_text(encoding='utf-8')
|
||||
assert 'ledmatrix-update-verify.service ledmatrix-update-verify.path' in text, installer
|
||||
assert 'enable --now ledmatrix-update-verify.path' in text, installer
|
||||
for unit in aus.UNITS:
|
||||
assert (f'systemd/{unit}', f'/etc/systemd/system/{unit}') in StartupValidator._UNITS
|
||||
|
||||
# Triggering takes no privilege any more; no sudoers rule should linger.
|
||||
for sudoers in ('scripts/install/configure_web_sudo.sh', 'first_time_install.sh'):
|
||||
assert not re.search(r'NOPASSWD:.*update-verify', (ROOT / sudoers).read_text(encoding='utf-8')), sudoers
|
||||
@@ -387,9 +387,81 @@ class TestFrameStatsPercentiles:
|
||||
|
||||
def test_log_frame_rate_emits_the_line_and_clears_the_window(self, helper):
|
||||
helper._window = [0.010] * 20
|
||||
helper.last_frame_time = time.time() # arm the clock; see below
|
||||
helper.last_fps_log_time = 0.0 # force the 5s boundary
|
||||
with patch.object(helper.logger, "info") as info:
|
||||
helper.log_frame_rate()
|
||||
assert info.called
|
||||
assert "Scroll frame stats" in info.call_args[0][0]
|
||||
assert helper._window == []
|
||||
|
||||
|
||||
class TestIdleGapIsNotAFrame:
|
||||
"""The first frame of a scroll has no predecessor, so timing one measures
|
||||
the idle gap since the last scroll rather than a frame.
|
||||
|
||||
Left in, that gap lands in the max field of otherwise healthy windows and
|
||||
counts as one stall per scroll start -- about 0.2% at 500 frames to a
|
||||
window, which is the same order as the real stall rates it sits next to.
|
||||
The stall rate is the number used to judge whether a scroll change worked,
|
||||
so it has to be clean.
|
||||
"""
|
||||
|
||||
def test_clock_starts_unarmed(self, helper):
|
||||
assert helper.last_frame_time is None
|
||||
|
||||
def test_first_frame_seeds_the_clock_without_sampling(self, helper):
|
||||
helper.log_frame_rate()
|
||||
assert helper.last_frame_time is not None
|
||||
assert helper._window == []
|
||||
assert helper.frame_times == []
|
||||
|
||||
def test_second_frame_is_sampled(self, helper):
|
||||
helper.log_frame_rate()
|
||||
helper.log_frame_rate()
|
||||
assert len(helper._window) == 1
|
||||
|
||||
def test_reset_scroll_disarms_the_clock(self, helper):
|
||||
helper.log_frame_rate()
|
||||
assert helper.last_frame_time is not None
|
||||
helper.reset_scroll()
|
||||
assert helper.last_frame_time is None
|
||||
|
||||
def test_gap_longer_than_the_log_interval_is_dropped(self, helper):
|
||||
"""Covers callers that scroll without ever calling reset_scroll()."""
|
||||
helper.last_frame_time = time.time() - 137.0 # a real observed gap
|
||||
helper.log_frame_rate()
|
||||
assert helper._window == []
|
||||
assert helper.frame_times == []
|
||||
|
||||
def test_a_gap_does_not_reach_the_stats_line(self, helper):
|
||||
helper.last_frame_time = time.time() - 137.0
|
||||
helper.last_fps_log_time = 0.0 # the 5s boundary is also due
|
||||
with patch.object(helper.logger, "info") as info:
|
||||
helper.log_frame_rate()
|
||||
assert not info.called, "the idle gap was reported as a frame"
|
||||
|
||||
def test_window_of_only_gaps_logs_nothing(self, helper):
|
||||
"""A window whose every sample was dropped has nothing to report --
|
||||
and reporting the gap itself is the bug this guards."""
|
||||
helper.last_fps_log_time = 0.0
|
||||
with patch.object(helper.logger, "info") as info:
|
||||
helper.log_frame_rate() # seeds
|
||||
helper.last_frame_time = time.time() - 137.0
|
||||
helper.log_frame_rate() # dropped
|
||||
assert not info.called
|
||||
|
||||
def test_seeding_restarts_the_window_timer(self, helper):
|
||||
"""A new scroll should not open by reporting a one-frame window."""
|
||||
helper.last_fps_log_time = 0.0 # boundary long overdue
|
||||
helper.log_frame_rate() # seeds
|
||||
with patch.object(helper.logger, "info") as info:
|
||||
helper.log_frame_rate() # first real frame
|
||||
assert not info.called
|
||||
assert len(helper._window) == 1
|
||||
|
||||
def test_a_normal_frame_still_counts(self, helper):
|
||||
helper.last_frame_time = time.time() - 0.010
|
||||
helper.log_frame_rate()
|
||||
assert len(helper._window) == 1
|
||||
assert helper._window[0] == pytest.approx(0.010, abs=0.005)
|
||||
|
||||
Reference in New Issue
Block a user