95 lines
3.9 KiB
Python
95 lines
3.9 KiB
Python
|
"""Reject invalid predicate results and broken actor isolation instrumentation."""
|
||
|
|
from dataclasses import replace
|
||
|
|
import json
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from scenarios.alice_bob_carol import build
|
||
|
|
from testdriver import Runner, Verdict
|
||
|
|
from testdriver.classification import classify
|
||
|
|
from testdriver.crystallization import assess_stability
|
||
|
|
from testdriver.oracles import Oracle
|
||
|
|
|
||
|
|
|
||
|
|
class CannotCoerce:
|
||
|
|
def __bool__(self):
|
||
|
|
raise AssertionError('oracle must not coerce predicate results')
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('value', [None, 'unknown', {}, [], 0, 1, CannotCoerce()])
|
||
|
|
def test_non_boolean_predicate_results_are_inconclusive(value):
|
||
|
|
_, _, _, asset, _ = build()
|
||
|
|
for assertion in (*asset.scenario.use_case.claims, *asset.scenario.use_case.invariants):
|
||
|
|
assertion = replace(assertion, predicate=lambda snapshot: value)
|
||
|
|
judgment = Oracle().judge(assertion, {'observed': True}, 'step')
|
||
|
|
assert judgment.verdict is Verdict.INCONCLUSIVE
|
||
|
|
assert 'boolean' in judgment.detail['reason']
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('value,verdict', [(True, Verdict.PASS), (False, Verdict.FAIL)])
|
||
|
|
def test_boolean_predicate_results_keep_their_meaning(value, verdict):
|
||
|
|
_, _, _, asset, _ = build()
|
||
|
|
assertion = replace(asset.scenario.use_case.claims[0], predicate=lambda snapshot: value)
|
||
|
|
assert Oracle().judge(assertion, {'observed': True}, 'step').verdict is verdict
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('value', [None, 'unknown', 1])
|
||
|
|
def test_invalid_predicates_cannot_authorize_acceptance_or_crystallization(value):
|
||
|
|
packs = []
|
||
|
|
for _ in range(3):
|
||
|
|
world, driver, observer, asset, oracle = build()
|
||
|
|
case = asset.scenario.use_case
|
||
|
|
asset.scenario = replace(asset.scenario, use_case=replace(case, claims=tuple(
|
||
|
|
replace(c, predicate=lambda snapshot: value) for c in case.claims)))
|
||
|
|
result = Runner(world, driver, observer, oracle).run(asset)
|
||
|
|
assert result.verdict is Verdict.INCONCLUSIVE
|
||
|
|
packs.append(json.loads(result.evidence.to_json()))
|
||
|
|
assert not classify(packs[0], packs[1]).safe_to_accept
|
||
|
|
assert not assess_stability(packs).stable
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('damage', ['shared', 'missing', 'changed', 'duplicate'])
|
||
|
|
@pytest.mark.parametrize('during_run', [False, True])
|
||
|
|
def test_corrupt_isolation_guards_stop_execution(damage, during_run):
|
||
|
|
packs = []
|
||
|
|
for _ in range(3):
|
||
|
|
world, driver, observer, asset, oracle = build()
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def corrupt():
|
||
|
|
alice, bob = world.cast['alice'], world.cast['bob']
|
||
|
|
if damage == 'shared':
|
||
|
|
alice._memory = bob._memory = {}
|
||
|
|
alice.remember('private', 'private-value')
|
||
|
|
assert bob.recall('private') == 'private-value'
|
||
|
|
elif damage == 'missing':
|
||
|
|
bob._memory.pop('__canary__')
|
||
|
|
elif damage == 'changed':
|
||
|
|
bob.remember('__canary__', 'replacement')
|
||
|
|
else:
|
||
|
|
bob.canary = alice.canary
|
||
|
|
bob.remember('__canary__', alice.canary)
|
||
|
|
|
||
|
|
class Driver:
|
||
|
|
def realize(self, actor, action):
|
||
|
|
calls.append(action.name)
|
||
|
|
result = driver.realize(actor, action)
|
||
|
|
corrupt()
|
||
|
|
return result
|
||
|
|
|
||
|
|
if not during_run:
|
||
|
|
corrupt()
|
||
|
|
result = Runner(world, Driver(), observer, oracle).run(asset)
|
||
|
|
assert result.verdict is Verdict.INCONCLUSIVE
|
||
|
|
assert len(calls) == int(during_run)
|
||
|
|
pack = json.loads(result.evidence.to_json())
|
||
|
|
guards = [o['data']['violations'] for o in pack['observations']
|
||
|
|
if o['kind'] == 'actor_isolation']
|
||
|
|
assert guards[-1]
|
||
|
|
assert 'private-value' not in result.evidence.to_json()
|
||
|
|
if damage == 'shared':
|
||
|
|
assert any('memory store' in v for v in guards[-1])
|
||
|
|
packs.append(pack)
|
||
|
|
assert not classify(packs[0], packs[1]).safe_to_accept
|
||
|
|
assert not assess_stability(packs).stable
|