test: gate releases on enforced conformance and installed package checks

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

tests/test_enforced_conformance.py

@@ -0,0 +1,89 @@
+"""Disposable enforcement-mode journeys; synthetic owners, real signed IAM token.
+
+These exercise the complete Tier 2/3 workload through AccessBoundary. They do
+not establish issuer registration, deployed custody or platform acceptance.
+"""
+import asyncio
+import json
+import time
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi.testclient import TestClient
+
+from hub_core.conformance import ConformanceHarness
+from hub_core.runtime.app import create_app
+from hub_core.runtime.config import RuntimeSettings
+from test_access_boundary import Owners
+from test_access_identity import setup
+
+
+@pytest.fixture(scope='module')
+def signing_key():
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+@pytest.fixture
+def enforced(signing_key):
+    token, identity, _, upstream = setup(signing_key)
+    owners = Owners()
+    owners.facts = replace(owners.facts, producer_addresses=frozenset({'hub:ops-hub'}))
+    async def current_facts(actor, resource):
+        return replace(owners.facts, checked_at=time.time())
+    owners.resolve = current_facts
+    controller = owners.controller()
+    controller.identity = identity
+    app = create_app(settings=RuntimeSettings(environment='test', access_mode='enforce'),
+                     access_controller=controller)
+    with TestClient(app, base_url='https://hub.example', headers={'Authorization': 'Bearer ' + token}) as client:
+        yield client, owners
+    asyncio.run(upstream.aclose())
+
+
+def test_tier_2_and_3_workload_passes_through_enforcement(enforced):
+    client, owners = enforced
+    report = ConformanceHarness(client).run()
+    assert report.passed, report.to_dict()
+    assert report.passed_count == 12
+    assert owners.requests
+    assert all(r.actor.subject == 'immutable-root' for r in owners.requests)
+    records = {r['correlation_id']: r for r in owners.records if r['outcome'] == 'authorized'}
+    for family in ('progress', 'interaction'):
+        result = client.get('/ports/projections/' + family + '_events')
+        event = result.json()['data']['items'][0]
+        attribution = event['payload']['_hub_access']
+        record = records[attribution['correlation_id']]
+        assert record['subject'] == 'immutable-root'
+    assert 'Projection rebuild conformance fixture.' not in json.dumps(owners.records)
+
+
+@pytest.mark.parametrize('failure,status', [('anonymous',401), ('invalid',401),
+    ('revoked',403), ('policy_denied',403), ('policy_outage',503), ('audit_outage',503)])
+def test_denied_reads_and_writes_leave_no_business_mutation(enforced, failure, status):
+    client, owners = enforced
+    assert ConformanceHarness(client).run().passed
+    before = client.get('/ports/projections/messages').json()['data']['items']
+    saved = client.headers['authorization']
+    if failure == 'anonymous':
+        del client.headers['authorization']
+    elif failure == 'invalid':
+        client.headers['authorization'] = 'Bearer invalid'
+    elif failure == 'revoked':
+        owners.facts = replace(owners.facts, root_entitled=False)
+    elif failure == 'policy_denied':
+        owners.allow = False
+    elif failure == 'policy_outage':
+        owners.policy_down = True
+    else:
+        owners.audit_down = True
+    message = {'schema_version':'0.1.0', 'correlation_id':str(uuid4()),
+               'from_address':'hub:ops-hub', 'to_addresses':['agent:conformance'],
+               'body':'must never commit'}
+    assert client.get('/ports/projections/messages').status_code == status
+    assert client.post('/ports/messaging/messages', json=message).status_code == status
+    client.headers['authorization'] = saved
+    owners.facts = replace(owners.facts, root_entitled=True)
+    owners.allow, owners.policy_down, owners.audit_down = True, False, False
+    assert client.get('/ports/projections/messages').json()['data']['items'] == before