tegwick
83849b75d4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
82 lines
4 KiB
Python
82 lines
4 KiB
Python
from dataclasses import replace
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from informed_decision.runtime import Runtime, AuditPump, token_file
|
|
from informed_decision.store import Store
|
|
from test_policy_client import policy,request
|
|
from test_durable_store import storage,present
|
|
from test_durable_component import receiver
|
|
|
|
|
|
def test_schema_one_upgrade_preserves_existing_evidence(storage):
|
|
store,memo=storage;p=present(store,memo)
|
|
before=store.evidence()
|
|
with sqlite3.connect(store.path) as db:
|
|
db.execute('DROP TABLE policy_observations');db.execute('PRAGMA user_version=1')
|
|
reopened=Store(store.path)
|
|
assert reopened.retrieve_presentation(p.id)[0]==memo and reopened.evidence()==before
|
|
assert reopened.policy_observations()==[]
|
|
observation=policy().check(request());reopened.observe_policy(observation)
|
|
assert Store(store.path).policy_observations()[0]['decision_attributable']==0
|
|
with sqlite3.connect(store.path) as db:
|
|
for statement in ['DELETE FROM policy_observations',"UPDATE policy_observations SET outcome='allow'"]:
|
|
with pytest.raises(sqlite3.IntegrityError):db.execute(statement)
|
|
|
|
|
|
def test_caller_token_is_reread_on_rotation_and_bounded(tmp_path):
|
|
path=tmp_path/'token';path.write_text('synthetic-first\n');read=token_file(path)
|
|
assert read()=='synthetic-first'
|
|
path.write_text('synthetic-rotated');assert read()=='synthetic-rotated'
|
|
path.write_bytes(b'x'*32769)
|
|
with pytest.raises(ValueError):read()
|
|
with pytest.raises(ValueError):token_file('relative')
|
|
|
|
|
|
def config(tmp_path):
|
|
private=tmp_path/'private';private.mkdir(mode=0o700)
|
|
path=private/'runtime.json'
|
|
data={'schema':'informed-decision.review-runtime.v1','evidence_db':str(private/'review.sqlite'),
|
|
'approval_origin':'https://approval.test','policy':{'origin':'https://policy.test','package':'fixture',
|
|
'version':'v1','package_digest':'sha256:'+'a'*64,'caller_token_file':str(private/'unprovisioned-caller')},
|
|
'audit':{'origin':'https://audit.test','sender_token_file':str(private/'unprovisioned-sender')}}
|
|
path.write_text(json.dumps(data));path.chmod(0o600)
|
|
return path,data
|
|
|
|
|
|
def test_config_load_does_not_provision_or_read_credentials(tmp_path):
|
|
path,data=config(tmp_path);runtime=Runtime.from_file(path)
|
|
assert not runtime.pump.ready()
|
|
assert not (path.parent/'unprovisioned-caller').exists()
|
|
runtime.pump.tick();assert not runtime.pump.ready()
|
|
assert all(r['state']=='blocked' for r in runtime.controller.store.outbox())
|
|
|
|
|
|
@pytest.mark.parametrize('fault',['permissions','symlink','unknown-key','relative-store','missing-pin','unsafe-origin'])
|
|
def test_unsafe_or_incomplete_config_refused(tmp_path,fault):
|
|
path,data=config(tmp_path)
|
|
if fault=='permissions':path.chmod(0o644)
|
|
elif fault=='symlink':
|
|
linked=path.parent/'link.json';linked.symlink_to(path);path=linked
|
|
elif fault=='unknown-key':data['allow_everyone']=True
|
|
elif fault=='relative-store':data['evidence_db']='relative.db'
|
|
elif fault=='missing-pin':data['policy']['package_digest']=''
|
|
elif fault=='unsafe-origin':data['approval_origin']='http://public.example.com'
|
|
if fault not in ('permissions','symlink'):path.write_text(json.dumps(data))
|
|
with pytest.raises(ValueError):Runtime.from_file(path)
|
|
|
|
|
|
def test_pump_retains_blocked_delivery_and_requires_repair(storage,receiver):
|
|
store,memo=storage;present(store,memo)
|
|
sink,_,_=receiver;original=sink.token_provider;sink.token_provider=lambda:'synthetic-wrong-token'
|
|
pump=AuditPump(store,sink);pump.tick();assert not pump.ready()
|
|
ids=[r['id'] for r in store.outbox() if r['state']=='blocked'];assert ids
|
|
sink.token_provider=original
|
|
pump.tick();assert not pump.ready() # No implicit credential repair/requeue.
|
|
for ident in ids:store.requeue_blocked(ident)
|
|
pump.tick();assert pump.ready()
|
|
assert (store.path.parent/'audit-reconciliation.json').stat().st_mode & 0o777 == 0o600
|
|
pump.clock=lambda:time.time()+91;assert not pump.ready()
|