tegwick
c976c9c277
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f0-a6f6-7b93-9124-55ef584930e2
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
import json
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from railiance_clock import health
|
|
|
|
|
|
@pytest.fixture
|
|
def policy():
|
|
return {'status': 'admitted', 'policy_id': 'test', 'kernel_helper': '/test/helper',
|
|
'host_drift_ppm': 1000, 'upstreams': ['ntp.test'],
|
|
'measurement_margin_ns': 1000, 'max_error_ns': 100_000_000}
|
|
|
|
|
|
@pytest.fixture
|
|
def commands(monkeypatch):
|
|
values = {
|
|
'/test/helper': json.dumps({'state': 0, 'status': 0, 'maxerror_us': 1000,
|
|
'esterror_us': 100, 'tolerance_scaled_ppm': 32768000}),
|
|
'show': 'NTP=yes\nNTPSynchronized=yes\n',
|
|
'show-timesync': 'ServerName=ntp.test\nNTPMessage={ Leap=0, Ignored=no, }\n',
|
|
}
|
|
def command(argv):
|
|
value = values[argv[0] if len(argv) == 1 else argv[1]]
|
|
if isinstance(value, Exception):
|
|
raise value
|
|
return value
|
|
monkeypatch.setattr(health, 'command', command)
|
|
return values
|
|
|
|
|
|
def test_collection_and_replacement_on_failure(tmp_path, policy, commands):
|
|
path = tmp_path / 'health.json'
|
|
good = health.collect(policy)
|
|
assert good['quality'] == 'bounded'
|
|
health.write_health(path, good)
|
|
commands['show'] = 'NTP=yes\nNTPSynchronized=no\n'
|
|
bad = health.collect(policy)
|
|
health.write_health(path, bad)
|
|
assert json.loads(path.read_text()) == {'quality': 'unusable', 'reason': 'host_quality_unavailable'}
|
|
assert path.stat().st_mode & 0o777 == 0o644
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
@pytest.mark.parametrize('bad', [None, [], True, 'bad', {'status': 'admitted'},
|
|
{'host_drift_ppm': -1}, {'host_drift_ppm': True}, {'upstreams': 'ntp.test'},
|
|
{'measurement_margin_ns': -1}, {'max_error_ns': False}, {'kernel_helper': []}])
|
|
def test_bad_policy_refuses(policy, commands, bad):
|
|
if isinstance(bad, dict):
|
|
if bad == {'status': 'admitted'}:
|
|
policy = bad
|
|
else:
|
|
policy.update(bad)
|
|
else:
|
|
policy = bad
|
|
assert health.collect(policy)['quality'] == 'unusable'
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['null', '[]', 'true', '{',
|
|
'{"state":false,"status":0}', '{"state":0,"status":"0"}',
|
|
OSError('private details'), subprocess.TimeoutExpired('private command', 3)])
|
|
def test_helper_failures_are_sanitized(policy, commands, bad):
|
|
commands['/test/helper'] = bad
|
|
assert health.collect(policy) == {'quality': 'unusable', 'reason': 'host_quality_unavailable'}
|