66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
|
import importlib.util
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
'sync', Path(__file__).resolve().parents[1] / 'scripts/openbao_policy_sync.py')
|
||
|
|
m = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(m)
|
||
|
|
|
||
|
|
NAME = 'workload-kv-read-activity-core-eso'
|
||
|
|
DECLARED = (m.POLICY_DIR / (NAME + '.hcl')).read_text()
|
||
|
|
|
||
|
|
|
||
|
|
def store(initial):
|
||
|
|
live = {'rules': initial}
|
||
|
|
writes = []
|
||
|
|
|
||
|
|
def write(name, path):
|
||
|
|
writes.append(name)
|
||
|
|
live['rules'] = Path(path).read_text()
|
||
|
|
return live, writes, (lambda name: live['rules']), write
|
||
|
|
|
||
|
|
|
||
|
|
def test_applies_when_live_is_the_expected_prior_version():
|
||
|
|
live, writes, read, write = store('path "old" {}')
|
||
|
|
result = m.sync(NAME, m.digest('path "old" {}'), read, write)
|
||
|
|
assert result['status'] == 'applied' and writes == [NAME]
|
||
|
|
assert m.digest(live['rules']) == m.digest(DECLARED)
|
||
|
|
|
||
|
|
|
||
|
|
def test_already_current_writes_nothing():
|
||
|
|
_, writes, read, write = store(DECLARED)
|
||
|
|
assert m.sync(NAME, 'unused', read, write)['status'] == 'already_current'
|
||
|
|
assert writes == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_refuses_undeclared_drift():
|
||
|
|
_, writes, read, write = store('path "someone-else" {}')
|
||
|
|
with pytest.raises(m.Refused, match='live_policy_drifted'):
|
||
|
|
m.sync(NAME, m.digest('path "old" {}'), read, write)
|
||
|
|
assert writes == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_readback_mismatch_is_refused():
|
||
|
|
live = {'rules': 'path "old" {}'}
|
||
|
|
with pytest.raises(m.Refused, match='readback_mismatch'):
|
||
|
|
m.sync(NAME, m.digest('path "old" {}'), lambda n: live['rules'], lambda n, p: None)
|
||
|
|
|
||
|
|
|
||
|
|
def test_new_paths_are_exact_and_read_only():
|
||
|
|
for worker in ('rein-aharness-railiance01', 'rein-aharness-metered-railiance01'):
|
||
|
|
block = 'path "platform/data/workloads/activity-core/ops-run-workers/%s" {\n capabilities = ["read"]\n}' % worker
|
||
|
|
assert block in DECLARED
|
||
|
|
assert 'ops-run-workers/*' not in DECLARED and '"list"' not in DECLARED
|
||
|
|
|
||
|
|
|
||
|
|
def test_refuses_outside_envelope_and_bad_args(tmp_path, monkeypatch):
|
||
|
|
monkeypatch.setenv('HOME', str(tmp_path))
|
||
|
|
receipt = tmp_path / 'r.json'
|
||
|
|
assert m.main(['--policy', NAME, '--expect-live-sha256', 'x', '--receipt', str(receipt)]) == 1
|
||
|
|
assert json.loads(receipt.read_text())['status'] == 'attended_envelope_required'
|
||
|
|
with pytest.raises(SystemExit):
|
||
|
|
m.parse(['--policy', '../etc', '--expect-live-sha256', 'x', '--receipt', 'r'])
|