"""Exercise qualified lookup, immutable event production and owner compatibility.""" from copy import deepcopy import hashlib import json from pathlib import Path import shutil import subprocess import sys import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'scripts')) import reference_consume as consume @pytest.fixture(scope='module') def published(): return consume.published() @pytest.fixture def decision(): return json.loads((ROOT / 'feedback/decision-input.example.json').read_text()) def test_lookup_explains_filters_and_preserves_exact_read_revisions(published): index, pin, manifest = published result = consume.lookup(index, pin, manifest, ['CAND-0007'], ['recoverability']) candidate, = result['candidates'] assert result['status'] == 'matches' assert candidate['id'] == 'CAND-0007' assert candidate['classification']['forces']['worsens'] assert candidate['gaps'] and candidate['maturity'] == 'hypothesis' assert candidate['contract']['revision'] == manifest['corpus_revision'] assert f'/commit/{manifest["corpus_revision"]}/' in candidate['contract']['url'] assert all(e['locator']['revision'] == pin['packaging_revision'] for e in candidate['evidence']) assert all(e['status'] == 'proposed' and e['condition'] and e['unknown'] for e in result['relations']) assert all(v['revision'] == pin['packaging_revision'] for v in result['entrypoints'].values()) assert result['reference']['manifest_sha256'] == pin['manifest_sha256'] assert consume.lookup(index, pin, manifest, ['CAND-0007'], ['telemetry-quality'])['status'] == 'no-match' def test_missing_context_abstains_and_responsibility_filters_intersect(published): index, pin, manifest = published result = consume.lookup(index, pin, manifest) assert result['status'] == 'needs-context' and result['candidates'] == result['relations'] == [] result = consume.lookup(index, pin, manifest, responsibilities=['mutation'], forces=['recoverability']) assert result['candidates'] assert all('mutation' in c['classification']['responsibilities'] and 'recoverability' in c['classification']['forces']['improves'] for c in result['candidates']) @pytest.mark.parametrize('filters', [ {'candidates': ['CAND-0018']}, {'forces': ['magic']}, {'responsibilities': ['magic']}, ]) def test_unknown_or_unpublished_identity_is_not_silently_selected(published, filters): with pytest.raises(ValueError, match='Unknown'): consume.lookup(*published, **filters) def test_live_research_is_independent_and_tampered_export_fails(tmp_path, published): shutil.copytree(ROOT / 'reference', tmp_path / 'reference') (tmp_path / '.git').symlink_to(ROOT / '.git', target_is_directory=True) live = tmp_path / 'research/candidates/MutationOutcomeTracking.md' live.parent.mkdir(parents=True) live.write_text('Legitimate later research outside the frozen release.\n') assert consume.published(tmp_path) == published export = tmp_path / 'reference/0.1/index.json' export.write_text('{}\n') with pytest.raises(ValueError, match='Published export changed'): consume.published(tmp_path) with pytest.raises(ValueError, match='Unknown reference version'): consume.published(tmp_path, '9.9') def test_receipt_keeps_unknown_outcome_and_pins_verified_reference(published, decision): index, pin, _ = published original = deepcopy(decision) event = consume.receipt(decision, index, pin) assert decision == original assert event['kind'] == 'synthetic' assert event['outcome']['value'] == 'unknown' assert event['outcome']['observation'] is None assert event['outcome']['measured_decision_minutes'] is None assert event['reference']['packaging_revision'] == pin['packaging_revision'] assert 'origin' not in event @pytest.mark.parametrize('mutate', [ lambda d: d.update(kind=''), lambda d: d.update(recorded_at='2999-01-01T00:00:00Z'), lambda d: d.update(recorded_at='2026-01-01T00:00:00'), lambda d: d.update(previous_event_sha256='invalid'), lambda d: d.update(origin={'verified': True}), lambda d: d['consumer'].update(revision='main'), lambda d: d['consumer'].update(run_ref=''), lambda d: d['context'].update(considered=['CAND-0018']), lambda d: d['context'].update(selected=['CAND-0007']), lambda d: d['context'].update(considered=['CAND-0007', 'CAND-0007']), lambda d: d['decision'].update(action='applied'), lambda d: d['outcome'].update(value='helpful'), lambda d: d['outcome'].update(measured_decision_minutes=-1), lambda d: d['outcome'].update(measured_rework_minutes=float('nan')), ]) def test_invalid_producer_claims_or_provenance_are_rejected(published, decision, mutate): mutate(decision) with pytest.raises(ValueError): consume.receipt(decision, *published[:2]) def test_followup_and_observed_result_retain_lineage(published, decision): decision['previous_event_sha256'] = 'a' * 64 decision['event_id'] = 'fixture-outcome-02' decision['outcome'].update(value='mixed', basis='observed-result', stage='test', observation='Disclosed fixture result; no consumer use.', evidence_refs=['fixture:test-01']) event = consume.receipt(decision, *published[:2]) assert event['previous_event_sha256'] == 'a' * 64 assert event['outcome']['value'] == 'mixed' def test_immutable_write_returns_exact_collector_index_entry(tmp_path, published, decision): event = consume.receipt(decision, *published[:2]) path = 'research/pattern-use/fixture/event-01.json' entry = consume.write_receipt(event, path, tmp_path) raw = (tmp_path / path).read_bytes() assert json.loads(raw) == event assert entry == {'path': path, 'sha256': hashlib.sha256(raw).hexdigest()} with pytest.raises(FileExistsError): consume.write_receipt(event, path, tmp_path) assert (tmp_path / path).read_bytes() == raw @pytest.mark.parametrize('path', ['/tmp/event.json', '../event.json', 'x//event.json', 'x/./event.json', 'x\\event.json']) def test_unsafe_paths_are_rejected_before_writing(tmp_path, published, decision, path): with pytest.raises(ValueError, match='Unsafe receipt path'): consume.write_receipt(consume.receipt(decision, *published[:2]), path, tmp_path) assert list(tmp_path.iterdir()) == [] def test_symlink_escape_is_refused(tmp_path, published, decision): (tmp_path / 'out').symlink_to(ROOT, target_is_directory=True) with pytest.raises(ValueError, match='escapes consumer root'): consume.write_receipt(consume.receipt(decision, *published[:2]), 'out/event.json', tmp_path) def test_cli_generates_owner_accepted_event_and_index(tmp_path): result = subprocess.run([sys.executable, str(ROOT / 'scripts/reference_consume.py'), '--version', '0.1', 'receipt', '--input', str(ROOT / 'feedback/decision-input.example.json'), '--output', 'use/event.json'], cwd=tmp_path, capture_output=True, text=True, check=True) entry = json.loads(result.stdout) assert entry['sha256'] == hashlib.sha256((tmp_path / entry['path']).read_bytes()).hexdigest() owner = ROOT.parent / 'activity-core' python = owner / '.venv/bin/python' if not python.exists(): pytest.skip('Actual activity-core contract check requires the sibling owner environment') script = '''import json, sys sys.path.insert(0, sys.argv[1]) from activity_core.context_resolvers.frontend_patterns import validate_event event = json.load(open(sys.argv[2])) validate_event(event, event['consumer']['repo']) ''' subprocess.run([str(python), '-B', '-c', script, str(owner / 'src'), str(tmp_path / entry['path'])], check=True, capture_output=True, text=True) def test_invalid_cli_decision_cannot_create_a_receipt(tmp_path, decision): decision['decision']['action'] = 'applied' # Selection remains empty. source = tmp_path / 'decision.json' source.write_text(json.dumps(decision)) result = subprocess.run([sys.executable, str(ROOT / 'scripts/reference_consume.py'), 'receipt', '--input', str(source), '--output', 'use/event.json'], cwd=tmp_path, capture_output=True, text=True) assert result.returncode == 1 assert 'Action/selection mismatch' in result.stderr assert not (tmp_path / 'use').exists()