"""Tests for run↔ops_run join and Forgejo artefact URLs (ACTIVITY-WP-0027).""" from __future__ import annotations import uuid from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock from activity_core.run_artifacts import ( _ops_summary, artifacts_from_ops_result, build_forgejo_blob_url, match_ops_runs_to_activity_run, ) def test_build_forgejo_blob_url_commit() -> None: url = build_forgejo_blob_url( target_repo="freedom-intelligence", path="briefs/2026/08/2026-08-05.md", ref="33a5affbee7dafe40eb7b20a53e18b6efad60e3b", web_base="https://forgejo.coulomb.social", org="coulomb", ) assert url is not None assert url.startswith("https://forgejo.coulomb.social/coulomb/freedom-intelligence/") assert "src/commit/" in url assert "briefs/2026/08/2026-08-05.md" in url def test_build_forgejo_blob_url_rejects_traversal() -> None: assert ( build_forgejo_blob_url( target_repo="freedom-intelligence", path="../etc/passwd", ref="main", ) is None ) def test_artifacts_from_ops_result() -> None: arts = artifacts_from_ops_result( { "ok": True, "path": "briefs/2026/08/2026-08-05.md", "head_after": "33a5affbee7dafe40eb7b20a53e18b6efad60e3b", "target_repo": "freedom-intelligence", }, title="FI brief", ) assert arts assert arts[0]["kind"] == "forgejo_blob" assert "freedom-intelligence" in arts[0]["url"] def test_match_ops_runs_prefers_run_id() -> None: run_id = uuid.uuid4() ar = MagicMock() ar.run_id = run_id ar.fired_at = datetime(2026, 8, 5, 5, 30, tzinfo=timezone.utc) exact = MagicMock() exact.triggering_event_id = str(run_id) exact.created_at = ar.fired_at other = MagicMock() other.triggering_event_id = "scheduled" other.created_at = ar.fired_at matched = match_ops_runs_to_activity_run(ar, [other, exact]) assert matched == [exact] def test_match_ops_runs_window_fallback() -> None: run_id = uuid.uuid4() ar = MagicMock() ar.run_id = run_id ar.fired_at = datetime(2026, 8, 5, 5, 30, tzinfo=timezone.utc) near = MagicMock() near.triggering_event_id = "scheduled" near.created_at = ar.fired_at + timedelta(seconds=5) far = MagicMock() far.triggering_event_id = "other" far.created_at = ar.fired_at + timedelta(hours=2) matched = match_ops_runs_to_activity_run(ar, [far, near]) assert near in matched assert far not in matched def test_ops_summary_exposes_compact_execution_constellation() -> None: now = datetime(2026, 8, 22, 20, 0, tzinfo=timezone.utc) row = MagicMock() row.id = uuid.uuid4() row.state = "succeeded" row.title = "Profiled run" row.target_repo = "activity-core" row.claim_owner = "glas-worker@railiance01" row.attempt = 1 row.triggering_event_id = "event-1" row.source_id = "profiled" row.approach_hint = "legacy-only" row.harness_profile_ref = "harness.agent-dev@1.0.0" row.execution_refs = {"goal_refs": ["goal:42@1"]} row.created_at = now row.updated_at = now row.result = { "ok": True, "execution_evidence": { "profile_ref": "harness.agent-dev@1.0.0", "rein_id": "rein-aharness", "rein_version": "0.1.0", "resolved_model": "claude-sonnet-4-6", "sandbox_profile": "profile.bwrap-local", "outcome": "succeeded", "duration_s": 5.2, }, "raw_output": "must-drop-even-for-historic-rows", } summary = _ops_summary(row) assert summary["execution_evidence"]["rein_id"] == "rein-aharness" assert summary["result"]["execution_evidence"]["profile_ref"] == ( "harness.agent-dev@1.0.0" ) assert "raw_output" not in str(summary)