frontend-patterns/tests/test_reference_consume.py

"""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()