activity-core/tests/test_run_artifacts.py
tegwick 7711bf9c70
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 20s
feat(ops): run artefacts in API/UI + llm-connect host access design
ACTIVITY-WP-0027: document durable ClusterIP host access for llm-connect;
join activity_runs to ops_runs; surface Forgejo artefact links on
/ops/ui automation detail and new run detail pages.
2026-08-05 17:23:02 +02:00

90 lines
2.5 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 (
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