fix: report truthful compatibility outcomes and verify audit coverage

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929

tests/test_compatibility_access.py

@@ -0,0 +1,140 @@
+"""Enforced compatibility writes: real SQL transactions, synthetic authority."""
+import json
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+import sqlalchemy as sa
+from fastapi.testclient import TestClient
+
+from hub_core.runtime.app import create_app
+from hub_core.runtime.config import RuntimeSettings
+from hub_core.runtime.tables import (compat_api_keys, compat_api_consumers,
+    runtime_audit_ledger, runtime_outcome_outbox, runtime_interaction_events)
+from test_access_boundary import HEADERS
+from test_outcome_audit import target, rows
+
+GROUPS = frozenset({'registry','credentials','interaction','deferred','system'})
+
+
+@pytest.fixture
+def compatibility(target):
+    _,store,owners,url = target
+    settings = RuntimeSettings(environment='test',access_mode='enforce',backend='postgresql',
+        database_url=url,v2_groups=GROUPS,v2_write_groups=GROUPS)
+    app = create_app(settings=settings,port_store=store,access_controller=owners.controller())
+    with TestClient(app,raise_server_exceptions=False) as client:
+        yield client,store,owners
+
+
+@pytest.mark.parametrize('prefix',['/api/v2',''])
+def test_all_implemented_compatibility_mutations_have_attributed_outcomes(compatibility,prefix):
+    client,store,owners = compatibility
+    ids = []
+    def write(path,body,method='POST',status=201):
+        response = client.request(method,prefix+path,headers=HEADERS,json=body)
+        assert response.status_code == status, response.text
+        ids.append(response.headers['x-correlation-id'])
+        return response.json()
+    hub = write('/hubs',{'name':'Fixture Hub','slug':'fixture-hub'})
+    manifest = write('/hub-capability-manifests',{'hubId':hub['id'],'manifestVersion':'1.0'})
+    write('/hub-capability-manifests/'+manifest['id'],{'description':'private manifest value'},'PATCH',200)
+    write('/hub-capability-manifests/'+manifest['id']+'/activate',{},status=200)
+    consumer = write('/api-consumers',{'name':'Fixture Consumer'})
+    issued = write('/api-consumers/'+consumer['id']+'/api-keys',{'scopes':'fixture-only'})
+    widget = write('/widgets',{'hubId':hub['id'],'name':'Fixture Widget'})
+    interaction = write('/interaction-events',{'widgetId':widget['id'],'eventType':'fixture.interaction',
+        'id':'caller-forged-id','metadata':{'text':'private event body'}})
+    assert interaction['id'] != 'caller-forged-id'
+    event, = rows(store,runtime_interaction_events)
+    assert interaction['id'] == event['id']
+    pending = rows(store,runtime_outcome_outbox)
+    assert len(pending) == len(ids) == 8
+    assert {row['envelope']['correlation_id'] for row in pending} == set(ids)
+    decisions = {record['correlation_id']: record for record in owners.records}
+    operations = set()
+    for row in pending:
+        event = row['envelope']
+        auth = event['data']['authorization']
+        assert auth['subject'] == 'immutable-root'
+        assert auth['decision_id'] == decisions[event['correlation_id']]['decision_id']
+        operations.add(event['data']['operation'])
+    assert operations == {'compat.hub.created','compat.manifest.created','compat.manifest.updated',
+        'compat.manifest.activated','compat.consumer.created','compat.api_key.created',
+        'compat.widget.created','event.interaction.accepted'}
+    serialized = json.dumps(pending)+json.dumps(rows(store,runtime_audit_ledger),default=str)
+    assert issued['fullKey'] not in serialized
+    assert 'private manifest value' not in serialized and 'private event body' not in serialized
+    # Legacy compatibility keys cannot authenticate an enforced route.
+    assert client.get(prefix+'/hubs',headers={'Authorization':'Bearer '+issued['fullKey']}).status_code == 401
+
+
+@pytest.mark.parametrize('prefix',['/api/v2',''])
+@pytest.mark.parametrize('path',['annotations','requirement-candidates','decision-records',
+    'deployment-records','outcome-signals','token'])
+def test_unimplemented_compatibility_writes_never_report_success(compatibility,prefix,path):
+    client,store,_ = compatibility
+    response = client.post(prefix+'/'+path,headers=HEADERS,json={})
+    assert response.status_code == 501
+    responses = client.app.openapi()['paths']['/api/v2/'+path]['post']['responses']
+    assert '501' in responses and '200' not in responses and '201' not in responses
+    assert not rows(store,runtime_outcome_outbox)
+    assert not rows(store,runtime_audit_ledger)
+
+
+@pytest.mark.parametrize('prefix',['/api/v2',''])
+@pytest.mark.parametrize('activate',[False,True])
+def test_missing_manifest_is_not_a_successful_mutation(compatibility,prefix,activate):
+    client,store,_ = compatibility
+    path = prefix+'/hub-capability-manifests/'+str(uuid4())
+    response = client.request('POST' if activate else 'PATCH',path+('/activate' if activate else ''),
+        headers=HEADERS,json={})
+    assert response.status_code == 404
+    assert not rows(store,runtime_outcome_outbox)
+
+
+def test_key_issuance_and_consumer_update_roll_back_on_outbox_failure(compatibility):
+    client,store,_ = compatibility
+    consumer = client.post('/api/v2/api-consumers',headers=HEADERS,json={'name':'Fixture Consumer'}).json()
+    before = rows(store,compat_api_consumers)
+    def fail(connection,cursor,statement,parameters,context,many):
+        if statement.startswith('INSERT INTO runtime_outcome_outbox'):
+            raise RuntimeError('fixture outbox failure')
+    sa.event.listen(store.engine.sync_engine,'before_cursor_execute',fail)
+    try:
+        response = client.post('/api/v2/api-consumers/'+consumer['id']+'/api-keys',headers=HEADERS,json={})
+        assert response.status_code == 500
+        assert 'fullKey' not in response.text
+    finally:
+        sa.event.remove(store.engine.sync_engine,'before_cursor_execute',fail)
+    assert not rows(store,compat_api_keys)
+    assert rows(store,compat_api_consumers) == before
+    assert len(rows(store,runtime_outcome_outbox)) == 1
+    assert len(rows(store,runtime_audit_ledger)) == 1
+
+
+@pytest.mark.parametrize('failure,status',[('anonymous',401),('invalid',401),('ordinary',403),
+    ('revoked',403),('wrong_tenant',403),('policy',503),('audit',503)])
+def test_compatibility_denials_do_not_mutate_or_queue_outcomes(compatibility,failure,status):
+    client,store,owners = compatibility
+    headers = HEADERS
+    if failure == 'anonymous': headers = {}
+    elif failure == 'invalid': headers = {'Authorization':'Bearer wrong'}
+    elif failure == 'ordinary': owners.actor = replace(owners.actor,subject='ordinary')
+    elif failure == 'revoked': owners.facts = replace(owners.facts,root_entitled=False)
+    elif failure == 'wrong_tenant': owners.actor = replace(owners.actor,tenant='tenant:other')
+    elif failure == 'policy': owners.policy_down = True
+    else: owners.audit_down = True
+    for prefix in ['/api/v2','']:
+        assert client.post(prefix+'/api-consumers',headers=headers,json={'name':'Never created'}).status_code == status
+    assert not rows(store,compat_api_consumers)
+    assert not rows(store,runtime_outcome_outbox)
+
+
+def test_read_only_and_disabled_group_still_prevent_writes(compatibility):
+    client,store,_ = compatibility
+    client.app.state.settings = replace(client.app.state.settings,v2_write_groups=frozenset())
+    assert client.post('/api-consumers',headers=HEADERS,json={'name':'Blocked'}).status_code == 503
+    client.app.state.settings = replace(client.app.state.settings,v2_groups=frozenset())
+    assert client.post('/api-consumers',headers=HEADERS,json={'name':'Blocked'}).status_code == 404
+    assert not rows(store,runtime_outcome_outbox)