activity-core/tests/test_frontend_patterns.py

import copy
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
import pytest
import yaml
from activity_core.context_resolvers import frontend_patterns as f
from activity_core.definition_parser import parse_file


def event(kind='real'):
    return dict(schema_version=1, use_id='demo:task:decision', event_id='first',
        previous_event_sha256=None, recorded_at='2026-08-20T12:00:00Z', kind=kind,
        consumer=dict(repo='demo', revision='a'*40, task_ref='workplans/task', producer='agent', run_ref='run:1'),
        reference=dict(version='0.1', packaging_revision='b'*40, manifest_sha256='c'*64),
        context=dict(problem='Choose frontend boundaries', constraints=[], considered=['CAND-0001'], selected=['CAND-0001']),
        decision=dict(action='applied', reason='Fits the task'),
        outcome=dict(value='unknown', basis='consumer-report', stage='decision', expectation='Fewer coupling defects', observation=None, evidence_refs=[]))


def fixtures(monkeypatch, events, duplicate=False):
    blobs = [yaml.safe_dump(e).encode() for e in events]
    entries = [dict(path=f'receipts/{i}.yaml', sha256=hashlib.sha256(b).hexdigest()) for i,b in enumerate(blobs)]
    if duplicate and entries:
        entries.append(entries[0])
    def raw(repo, sha, path):
        if repo == 'frontend-patterns':
            return json.dumps(dict(schema_version=1,sources=[dict(repo='demo', branch='main', index='receipts/index.json')])).encode()
        if path.endswith('index.json'):
            return json.dumps(dict(schema_version=1,receipts=entries)).encode()
        return blobs[int(path.split('/')[-1].split('.')[0])]
    monkeypatch.setattr(f,'revision',lambda *args: 'd'*40)
    monkeypatch.setattr(f,'raw',raw)
    return entries


def test_unknown_real_duplicate_and_synthetic_separate(monkeypatch):
    synthetic = event('synthetic'); synthetic['use_id']='fixture:task:decision'; synthetic['event_id']='fixture'
    fixtures(monkeypatch,[event(),synthetic],True)
    report=f.collect()
    assert report['real_uses']==1 and report['synthetic_uses']==1
    assert report['duplicates']==1 and report['outcomes']=={'unknown':1}
    assert report['model_calls']==0 and report['execution_ready'] is False


def test_delayed_outcome_and_month_boundary(monkeypatch):
    root=event(); follow=copy.deepcopy(root)
    follow.update(event_id='later',previous_event_sha256=hashlib.sha256(yaml.safe_dump(root).encode()).hexdigest(),recorded_at='2026-09-05T00:00:00Z')
    follow['outcome'].update(value='helpful',observation='Test passed',evidence_refs=['tests/result'])
    fixtures(monkeypatch,[follow,root])
    report=f.collect('monthly',datetime(2026,9,10,tzinfo=timezone.utc))
    assert report['outcomes']=={'unknown':1}
    assert f.collect()['outcomes']=={'helpful':1}
    assert f.collect()['signals'][0]['verification_required'] is True


def test_identity_collision_quarantines_use(monkeypatch):
    conflict=copy.deepcopy(event()); conflict['decision']['reason']='different'
    fixtures(monkeypatch,[event(),conflict])
    report=f.collect()
    assert report['real_uses']==0 and report['status']=='degraded'
    assert 'quarantined_event_lineage' in report['errors']


def test_missing_predecessor_quarantines_use(monkeypatch):
    e=event(); e['previous_event_sha256']='e'*64
    fixtures(monkeypatch,[e])
    assert f.collect()['real_uses']==0


@pytest.mark.parametrize('value',['helpful','harmful','mixed','no-effect'])
def test_outcomes_need_evidence(value):
    e=event(); e['outcome']['value']=value
    with pytest.raises(ValueError): f.validate_event(e,'demo')
    e['outcome'].update(observation='Observed result',evidence_refs=['result'])
    f.validate_event(e,'demo')


@pytest.mark.parametrize('action',['rejected','no-match'])
def test_empty_selection_valid_for_abstention(action):
    e=event(); e['context']['selected']=[]; e['decision']['action']=action
    f.validate_event(e,'demo')


@pytest.mark.parametrize('path',['../secret','/etc/passwd','x//y','x/./y','x?token=a','https://evil/x'])
def test_reject_unsafe_path(path):
    assert not f.path_ok(path)


def test_unavailable_is_not_no_data(monkeypatch):
    monkeypatch.setattr(f,'revision',lambda *args: (_ for _ in ()).throw(ValueError('unavailable')))
    report=f.collect()
    assert report['status']=='degraded' and report['errors']


def test_empty_roster_is_no_consumers(monkeypatch):
    monkeypatch.setattr(f,'revision',lambda *args:'a'*40)
    monkeypatch.setattr(f,'raw',lambda *args:b'{"schema_version":1,"sources":[]}')
    report=f.collect()
    assert report['status']=='no_consumers' and report['eligible_tasks'] is None
    assert len(f.FrontendPatternsResolver().resolve('daily',None,{})) < 3900


def test_over_limit_is_explicit(monkeypatch):
    entries=fixtures(monkeypatch,[event()])
    entries.extend([entries[0]]*100)
    report=f.collect()
    assert report['status']=='degraded' and report['receipt_events']==0


def test_domain_definitions_parse():
    manifest = Path(__file__).parents[1] / 'k8s/railiance/20-runtime.yaml'
    cm = next(d for d in yaml.safe_load_all(manifest.read_text()) if d and d.get('metadata', {}).get('name') == 'actcore-external-activity-definitions')
    import tempfile
    for name, body in cm['data'].items():
        if not name.startswith('frontend-patterns-'): continue
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / name
            path.write_text(body)
            d=parse_file(path)
        assert d.instructions[0]['model']=='deterministic'