activity-core/tests/test_run_artifacts.py
tegwick 26934e25b9
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Container Image / build-and-push (push) Successful in 21s
Enforce bounded operation guardrails
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f
2026-08-23 12:31:13 +02:00

142 lines
4.2 KiB
Python

"""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,
public_context_keys,
)
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_public_context_keys_hide_credential_and_raw_payload_names() -> None:
assert public_context_keys(
{
"safe_summary": {},
"credential": "must-drop",
"tool_output": "must-drop",
"provider_response": {},
}
) == ["safe_summary"]
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)