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.
90 lines
2.5 KiB
Python
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
|