railiance-platform/tests/test_task_execution_readback.py

"""Exact disposable native acceptance, partial effects and restart refusal."""
import json
from dataclasses import replace
import pytest
from test_task_execution_native_fixture import owner,bao_dev,compiler,admit
from task_execution_native_fixture import NativeFixtureAdapter
from task_execution_journal import Refused,Journal
from task_execution_cleanup import run_cleanup


def start(native,journal,binding):
    admit(native,journal,binding);native.materialize_grant()
    journal.transition(binding,'grant-issued','applying')


def test_expectation_digest_pin_rejected_before_reserve(owner):
    n,j,b,c,compiler,*_=owner
    with pytest.raises(Refused,match='expected-state digest'):
        NativeFixtureAdapter(compiler,n.plan,j,b,replace(n.admission,expected_after=b'{}'))
    assert c('GET','sys/policies/acl/'+n.policy_name)[0]==404


@pytest.mark.parametrize('target',[0,3])
def test_actual_conflicting_baseline_gates_its_target_write(owner,target):
    n,j,b,c,compiler,*_=owner
    call=compiler.before_call(n.plan,target+1)
    existing={'policy':'path "unrelated" { capabilities = ["read"] }'} if target==0 else json.loads(call.payload)
    assert c('PUT',call.path,existing)[0] in {200,204}
    with j.locked():
        start(n,j,b)
        for i in range(target): n.metadata_call(i)
        with pytest.raises(Refused,match='baseline or readback'):n.metadata_call(target)
        assert j.observe(b)['state']=='unknown'
        if target==3:
            # Policy effect already happened: this is deliberately no aggregate CAS.
            assert c('GET',compiler.before_call(n.plan,1).path)[1]['data']['policy']==json.loads(compiler.before_call(n.plan,1).payload)['policy']
        assert run_cleanup(j,b,n)['confirmed']
        assert j.observe(b)['state']=='unknown'


@pytest.mark.parametrize('target',[2,5])
def test_actual_readback_drift_never_verifies(owner,target):
    n,j,b,c,compiler,*_=owner
    with j.locked():
        start(n,j,b)
        for i in range(target):n.metadata_call(i)
        call=compiler.before_call(n.plan,target-1)
        body=json.loads(call.payload)
        if target==2:body['policy']+='\npath "sibling/*" { capabilities = ["read"] }'
        else:body['policies']+=',unexpected-policy'
        assert c('PUT',call.path,body)[0] in {200,204}
        with pytest.raises(Refused,match='baseline or readback'):n.metadata_call(target)
        with pytest.raises(Refused):n.verify_metadata()
        assert run_cleanup(j,b,n)['confirmed']
        assert j.observe(b)['state']=='unknown'


@pytest.mark.parametrize('status,response',[(403,{'errors':[]}), (404,{}),(404,{'errors':['permission denied']}), (200,{'data':None})])
def test_only_exact_backend_absence_counts(status,response):
    with pytest.raises(Refused):NativeFixtureAdapter._metadata_snapshot('auth/kubernetes/role/x',status,response)


def test_json_types_and_unknown_role_fields_remain_significant(owner):
    n,j,b,c,compiler,*_=owner
    original=n._http
    def changed(method,path,token,body=b''):
        status,response=original(method,path,token,body)
        if method=='GET' and path==compiler.before_call(n.plan,5).path and status==200:
            response['data']['token_num_uses']=False # False == 0 in Python; never canonical JSON.
        return status,response
    n._http=changed
    with j.locked():
        start(n,j,b)
        for i in range(5):n.metadata_call(i)
        with pytest.raises(Refused):n.metadata_call(5)
        assert run_cleanup(j,b,n)['confirmed']
        assert j.observe(b)['state']=='unknown'


def test_reopened_sqlite_cannot_resume_or_recover_memory_handles(owner):
    n,j,b,c,compiler,*_=owner
    with j.locked():
        start(n,j,b)
        n.metadata_call(0);n.metadata_call(1)
    # Model loss of the live admission; handles deliberately are not reconstructed.
    # The still-running disposable server proves native effect survives reopening.
    location=j.directory
    assert c('GET',compiler.before_call(n.plan,1).path)[0]==200
    j.db.close()
    # Owner journal retains the exact DB path; this is a real reopened SQLite.
    reopened=Journal(location)
    try:
        with reopened.locked():
            assert reopened.observe(b)['dispatch_allowed'] is False
            with pytest.raises(Refused):reopened.reserve(b,n.cleanup)
            with pytest.raises(Refused):reopened.next_call_ordinal(b)
    finally:reopened.close()
    # No recovery/cleanup claim: only same-process fixture teardown retains handles.


def test_verified_readback_with_real_cleanup_denial_is_not_complete(owner):
    n,j,b,c,compiler,*_=owner
    with j.locked():
        start(n,j,b)
        for i in range(6):n.metadata_call(i)
        n.verify_metadata()
        # Actual backend removal denies the task's self-revoke capability.
        assert c('DELETE','sys/policies/acl/'+n.policy_name)[0] in {200,204}
        result=run_cleanup(j,b,n)
        assert result['confirmed'] is False
        assert j.observe(b)['state']=='verified'
        with pytest.raises(Refused):j.transition(b,'verified','cleanup-confirmed','0'*64)
        assert c('POST','auth/token/lookup-accessor',{'accessor':n.task.accessor})[0]==200


def test_actual_denied_baseline_cannot_be_caught_then_write_retried(owner):
    n,j,b,c,compiler,*_=owner
    with j.locked():
        start(n,j,b)
        # Task still has update but loses read: real 403 is not absent baseline.
        denied=n.hcl.replace('["read", "create", "update"]','["create", "update"]')
        assert c('PUT','sys/policies/acl/'+n.policy_name,{'policy':denied})[0] in {200,204}
        with pytest.raises(Refused):n.metadata_call(0)
        assert j.observe(b)['state']=='unknown'
        with pytest.raises(Refused):n.metadata_call(1)
        assert c('GET',compiler.before_call(n.plan,1).path)[0]==404
        assert run_cleanup(j,b,n)['confirmed']