feat(plane): unified status-comment format with duration line (ORCH-016)
All checks were successful
CI / test (push) Successful in 11s
CI / test (pull_request) Successful in 12s

Все агенты (analyst..deployer) теперь пишут финальный коммент через единый
хелпер usage.build_status_comment(...) — заголовок «{icon} {Role} — {описание}»,
опциональная строка Verdict/Status из YAML-frontmatter, строка
«Длительность: 4m 12s» (явный duration_s от launcher, fallback из agent_runs
для аналитика), HTML-блок Документы, тех-хвост <sub>tokens · cost</sub>.

- Новые публичные функции в src/usage.py: build_status_comment, fmt_duration,
  get_agent_duration. usage_comment(...) → тонкая deprecated-обёртка (legacy
  тесты в tests/test_usage.py продолжают работать). artifact_links(...)
  переписан на HTML <li><a>…</a></li> (breaking change для внутреннего API,
  но единственный внешний клиент — _post_usage_comments — мигрирован).
- Новый модуль src/frontmatter.py: defensive YAML reader, никогда не raise.
- stage_engine._build_analyst_ready_comment(...) теперь тонкая обёртка над
  build_status_comment(agent="analyst", ...); task_id пробрасывается из
  _handle_analysis_approved_flow для DB-фоллбэка длительности (AC-14).
- launcher._post_usage_comments(...) принимает duration_s, резолвит stage из
  tasks для deployer и worktree_root для AC-8 graceful skipping.

Тесты (16 файлов, 56 новых тестовых функций, покрывают TC-01..TC-25):
fmt_duration table, build_status_comment по всем агентам, DB-фоллбэк,
authorship под per-agent ботами, дедуп-инвариант, regression на
status-only verdict аналитика и финальный notify_done, snapshot
QG_CHECKS + STAGE_TRANSITIONS.

Документация: docs/architecture/README.md (раздел Plane Sync),
CHANGELOG.md (Unreleased Added/Changed).

Refs: ORCH-016

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-05 12:39:06 +00:00
parent 1150cd9144
commit 0663da6e4c
16 changed files with 2071 additions and 112 deletions

View File

@@ -0,0 +1,126 @@
"""ORCH-016 / TC-11 + AC-6: analyst status-comment regression.
Status-only verdict model from PR #12 / #13 must be preserved exactly:
- the analyst comment still asks the stakeholder for the **Approved** status,
- it still rejects the obsolete ``:approved:`` reaction and "move to In Progress",
- it still links the documents that actually exist (BRD / TRZ / AC / Test Plan,
skipping anything not on disk),
- it now also carries the new «Длительность: …» line when an agent_runs row
exists for (task_id, analyst).
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_analyst_regression.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, get_db # noqa: E402
@pytest.fixture(autouse=True)
def setup_db(monkeypatch):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
def _seed_task_and_analyst_run(task_id=42, agent="analyst", duration_seconds=180):
"""Insert a task and a finished analyst run with a measurable duration."""
conn = get_db()
conn.execute(
"INSERT INTO tasks (id, repo, branch, stage, work_item_id) "
"VALUES (?, 'orchestrator', 'feature/ORCH-016', 'analysis', 'ORCH-016')",
(task_id,),
)
conn.execute(
"INSERT INTO agent_runs (task_id, agent, started_at, finished_at) "
"VALUES (?, ?, datetime('now', ?), datetime('now'))",
(task_id, agent, f"-{duration_seconds} seconds"),
)
conn.commit()
conn.close()
def test_tc11_analyst_text_preserved_with_links(monkeypatch, tmp_path):
"""Analyst comment must keep all existing assertions from PR #12 / #13."""
from src import stage_engine as SE
from src.config import settings
wt = tmp_path / "wt"
docs = wt / "docs" / "work-items" / "ET-011"
docs.mkdir(parents=True)
for fname in (
"00-business-request.md", "01-brd.md", "02-trz.md",
"03-acceptance-criteria.md", "04-test-plan.yaml",
):
(docs / fname).write_text("x")
# 04b-ui-test-cases.md intentionally absent
monkeypatch.setattr(SE, "get_worktree_path", lambda repo, branch: str(wt))
monkeypatch.setattr(settings, "gitea_url", "http://localhost:3000", raising=False)
monkeypatch.setattr(
settings, "gitea_public_url", "https://git.mva154.duckdns.org", raising=False
)
monkeypatch.setattr(settings, "gitea_owner", "admin", raising=False)
html = SE._build_analyst_ready_comment(
"enduro-trails", "ET-011", "feature/ET-011-gpx-upload-feature",
)
# Status-only verdict text (PR #12 contract).
assert "Approved" in html
assert "Rejected" in html
assert ":approved:" not in html
assert "In Progress" not in html
# Clickable links via public URL only.
assert "<a href=" in html
base = ("https://git.mva154.duckdns.org/admin/enduro-trails/src/branch/"
"feature/ET-011-gpx-upload-feature/docs/work-items/ET-011/")
assert base + "01-brd.md" in html
assert base + "04-test-plan.yaml" in html
# Missing file NOT linked.
assert "04b-ui-test-cases.md" not in html
# Internal URL must NOT leak into clickable links.
assert "localhost:3000" not in html
def test_tc11_analyst_includes_duration_when_db_has_run(monkeypatch, tmp_path):
"""When an agent_runs row exists for (task_id, analyst), the comment carries
a «Длительность:» line populated via the DB fallback (AC-14)."""
from src import stage_engine as SE
from src.config import settings
wt = tmp_path / "wt"
(wt / "docs" / "work-items" / "ORCH-016").mkdir(parents=True)
(wt / "docs" / "work-items" / "ORCH-016" / "01-brd.md").write_text("x")
_seed_task_and_analyst_run(task_id=42, agent="analyst", duration_seconds=125)
monkeypatch.setattr(SE, "get_worktree_path", lambda repo, branch: str(wt))
monkeypatch.setattr(settings, "gitea_url", "http://localhost:3000", raising=False)
monkeypatch.setattr(settings, "gitea_public_url", "", raising=False)
monkeypatch.setattr(settings, "gitea_owner", "admin", raising=False)
html = SE._build_analyst_ready_comment(
"orchestrator", "ORCH-016", "feature/ORCH-016", task_id=42,
)
# Two-digit seconds rounding may shave ~1s — accept either neighbour.
assert any(
s in html
for s in ("Длительность: 2m 05s", "Длительность: 2m 04s", "Длительность: 2m 06s")
), html

View File

@@ -0,0 +1,135 @@
"""ORCH-016 / TC-16 + AC-6: analyst status-only regression.
Status-only verdict model (PR #12 / #13):
- analyst finishes its run -> Plane state becomes In Review,
- ONE status comment is posted asking the stakeholder to flip the status to
Approved (or write a reason and switch to Rejected),
- NO auto-advance happens — the next stage waits for human approval.
The ORCH-016 PR refactors the comment text into the unified status-comment
helper. This regression test guards against:
(a) the analyst path silently auto-advancing,
(b) the analyst comment losing the «Approved» / «Rejected» instruction text,
(c) the comment switching authorship away from the analyst bot.
We exercise `_handle_analysis_approved_flow` directly (the launcher path).
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_analyst_so.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, get_db # noqa: E402
REPO = "enduro-trails"
BRANCH = "feature/ET-016-x"
WID = "ET-016"
@pytest.fixture(autouse=True)
def setup_db(monkeypatch, tmp_path):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
conn = get_db()
conn.execute(
"INSERT INTO tasks (id, repo, branch, stage, work_item_id) "
"VALUES (1, ?, ?, 'analysis', ?)",
(REPO, BRANCH, WID),
)
conn.commit()
conn.close()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
@pytest.fixture
def fake_worktree(monkeypatch, tmp_path):
base = tmp_path / "wt"
docs = base / "docs" / "work-items" / WID
docs.mkdir(parents=True)
# All analyst artifacts present -> "files_check" returns True.
for f in ("01-brd.md", "02-trz.md", "03-acceptance-criteria.md",
"04-test-plan.yaml"):
(docs / f).write_text("x")
monkeypatch.setattr("src.git_worktree.get_worktree_path", lambda r, b: str(base))
monkeypatch.setattr("src.stage_engine.get_worktree_path", lambda r, b: str(base))
monkeypatch.setattr("src.qg.checks.get_worktree_path", lambda r, b: str(base))
return base
@pytest.fixture
def collect_calls(monkeypatch):
calls = {"in_review": 0, "advance": 0, "comments": [], "enqueued": []}
monkeypatch.setattr(
"src.stage_engine.set_issue_in_review",
lambda wid: calls.__setitem__("in_review", calls["in_review"] + 1),
)
monkeypatch.setattr(
"src.stage_engine.notify_approve_requested", lambda tid: None
)
def _add_comment(wid, body, author=None, **kw):
calls["comments"].append({"wid": wid, "body": body, "author": author})
monkeypatch.setattr("src.stage_engine.plane_add_comment", _add_comment)
# advance_stage isn't directly hit; if anything calls update_task_stage to
# 'architecture', we'd see it here.
def _update_task_stage(task_id, stage):
calls["advance"] += 1
monkeypatch.setattr("src.stage_engine.update_task_stage", _update_task_stage)
def _enqueue(*a, **k):
calls["enqueued"].append((a, k))
return 1
monkeypatch.setattr("src.stage_engine.enqueue_job", _enqueue)
return calls
def test_tc16_analyst_goes_to_in_review_no_advance(fake_worktree, collect_calls):
"""When the analyst finishes with complete artifacts, the task goes to In
Review and NO advance/enqueue happens — the human approves via Plane status.
"""
from src.stage_engine import _handle_analysis_approved_flow, AdvanceResult
result = AdvanceResult(from_stage="analysis")
_handle_analysis_approved_flow(
task_id=1, current_stage="analysis", repo=REPO, work_item_id=WID,
branch=BRANCH, agent="analyst", result=result,
)
# In Review state requested in Plane.
assert collect_calls["in_review"] == 1, collect_calls
# NO stage-machine advance.
assert collect_calls["advance"] == 0, collect_calls
# NO new job enqueued by the analyst path.
assert collect_calls["enqueued"] == [], collect_calls
# Exactly one comment posted, authored by analyst, with required text bits.
assert len(collect_calls["comments"]) == 1, collect_calls["comments"]
c = collect_calls["comments"][0]
assert c["wid"] == WID
assert c["author"] == "analyst"
body = c["body"]
assert "Approved" in body
assert "Rejected" in body
assert ":approved:" not in body
assert "In Progress" not in body
# AC-6 +: the new unified format adds a Длительность line (DB fallback).
# No agent_runs row exists in this test, so the line should be ABSENT.
assert "Длительность" not in body

View File

@@ -0,0 +1,68 @@
"""ORCH-016 / AC-13 + AC-22: fmt_duration formatting contract.
Pure-function tests for the duration formatter used by build_status_comment.
No DB, no I/O — just the table in ADR-001 §8 / AC-13.
"""
import os
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
from src.usage import fmt_duration # noqa: E402
# ---------------------------------------------------------------------------
# TC-21: table-driven happy path (AC-13)
# ---------------------------------------------------------------------------
def test_fmt_duration_boundary_table():
cases = [
(0, "0s"),
(12, "12s"),
(59, "59s"),
(60, "1m 00s"),
(252, "4m 12s"),
(3599, "59m 59s"),
(3600, "1h 00m"),
(3780, "1h 03m"),
(10020, "2h 47m"),
]
for seconds, expected in cases:
assert fmt_duration(seconds) == expected, (
f"fmt_duration({seconds}) -> {fmt_duration(seconds)!r}; expected {expected!r}"
)
# ---------------------------------------------------------------------------
# TC-22: None / negative -> empty string (caller drops the line) (AC-13)
# ---------------------------------------------------------------------------
def test_fmt_duration_none_returns_empty():
assert fmt_duration(None) == ""
def test_fmt_duration_negative_returns_empty():
assert fmt_duration(-1) == ""
assert fmt_duration(-3600) == ""
def test_fmt_duration_garbage_returns_empty():
# Non-coercible input must not raise (defensive).
assert fmt_duration("abc") == ""
assert fmt_duration([1, 2]) == ""
def test_fmt_duration_float_seconds_truncated():
# int(12.9) == 12 — integer truncation, not rounding.
assert fmt_duration(12.9) == "12s"
assert fmt_duration(61.4) == "1m 01s"
# ---------------------------------------------------------------------------
# Caller contract: empty string => the 'Длительность:' line is NOT printed.
# build_status_comment is unit-tested in test_status_comment_format; here we
# just sanity-check the helper used to gate that decision.
# ---------------------------------------------------------------------------
def test_empty_string_is_falsy():
assert not fmt_duration(None)
assert not fmt_duration(-5)
assert fmt_duration(0) # "0s" IS truthy: AC-13 wants the line printed

View File

@@ -0,0 +1,79 @@
"""ORCH-016 / TC-18 + AC-7: notify_done / set_issue_done not regressed.
The final deploy -> done transition still posts the «✅ Task completed!»
comment under the deployer bot, alongside the new ORCH-016 status comment
the deployer publishes when it finishes the stage. The two comments are
independent — the status comment doesn't replace `notify_done`.
"""
import os
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
from src import plane_sync as PS # noqa: E402
def test_notify_done_constants_unchanged():
# Emoji + message body — pinned to lock the contract.
assert PS.EMOJI_DONE == ""
def test_notify_done_posts_completed_comment(monkeypatch):
"""plane_sync.notify_done still posts the ✅ Task completed! comment
authored by the deployer."""
captured = {}
def _spy_update(work_item_id, state, project_id=None):
captured["update"] = (work_item_id, state, project_id)
def _spy_add(work_item_id, body, project_id=None, author=None, **kw):
captured.setdefault("comments", []).append(
{"wid": work_item_id, "body": body, "author": author}
)
monkeypatch.setattr(PS, "update_issue_state", _spy_update)
monkeypatch.setattr(PS, "add_comment", _spy_add)
monkeypatch.setattr(PS, "_resolve_project_id", lambda wid, pid=None: "p-1")
PS.notify_done("ET-016")
assert captured["update"] == ("ET-016", "done", "p-1")
assert len(captured["comments"]) == 1
c = captured["comments"][0]
assert c["wid"] == "ET-016"
assert c["author"] == "deployer"
# Body untouched: emoji + canonical Russian/English copy.
assert "" in c["body"]
assert "Task completed" in c["body"]
def test_set_issue_done_still_exported():
"""set_issue_done must remain importable from plane_sync — stage_engine
line ~269 invokes it on deploy->done. ORCH-016 must not remove or rename it.
"""
assert callable(getattr(PS, "set_issue_done", None))
# And stage_engine still imports it at the module level (regression: ORCH-016
# touches stage_engine to wire the new analyst comment helper).
from src import stage_engine as SE
assert getattr(SE, "set_issue_done", None) is PS.set_issue_done
def test_orch016_does_not_steal_done_signal(monkeypatch):
"""build_status_comment is just a comment — it must NOT call set_issue_done
or notify_done as a side effect (that's stage_engine's job)."""
from src import usage as U
called = {"done": 0, "in_review": 0}
def _fail(*a, **k):
called["done"] += 1
monkeypatch.setattr(PS, "set_issue_done", _fail)
monkeypatch.setattr(PS, "notify_done", _fail)
html = U.build_status_comment(
"deployer", repo="enduro-trails", branch="b", work_item_id="ET-016",
stage="deploy", duration_s=12,
)
assert "\U0001f680 Deployer" in html
assert called["done"] == 0

View File

@@ -0,0 +1,199 @@
"""ORCH-016 / TC-13..TC-15: _post_usage_comments integration tests.
End-to-end (DB + filesystem worktree, no network) verification that
AgentLauncher._post_usage_comments:
- resolves the task by (repo, branch),
- threads the explicit duration_s into build_status_comment,
- posts exactly ONE status comment authored by the finishing agent,
- for deployer: ALSO posts the per-task usage summary (deployer authorship).
The actual Plane HTTP call (plane_sync.add_comment) is patched out; we only
check the (work_item_id, body, author) tuples the launcher passes to it.
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_post_usage.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, get_db # noqa: E402
from src.agents.launcher import AgentLauncher # noqa: E402
REPO = "enduro-trails"
BRANCH = "feature/ET-016-x"
WID = "ET-016"
@pytest.fixture
def setup_db(monkeypatch):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
conn = get_db()
conn.execute(
"INSERT INTO tasks (id, repo, branch, stage, work_item_id) "
"VALUES (1, ?, ?, 'review', ?)",
(REPO, BRANCH, WID),
)
conn.commit()
conn.close()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
@pytest.fixture
def fake_worktree(monkeypatch, tmp_path):
"""Stub get_worktree_path inside the launcher module to a tmp_path location."""
wt = tmp_path / "wt"
(wt / "docs" / "work-items" / WID).mkdir(parents=True)
def _get_wt(repo, branch):
return str(wt)
# The launcher imports get_worktree_path lazily inside the function body
# (`from ..git_worktree import get_worktree_path`); patch the source module.
monkeypatch.setattr("src.git_worktree.get_worktree_path", _get_wt)
monkeypatch.setattr("src.usage._input_total", lambda u: 0) # quiet <sub> tail
return wt
@pytest.fixture
def capture_comments(monkeypatch):
posts = []
def _spy(work_item_id, body, author=None, **kwargs):
posts.append({"wid": work_item_id, "body": body, "author": author})
monkeypatch.setattr("src.agents.launcher.plane_add_comment", _spy)
return posts
@pytest.fixture
def public_url(monkeypatch):
from src.config import settings
monkeypatch.setattr(
settings, "gitea_public_url", "https://git.mva154.duckdns.org", raising=False
)
monkeypatch.setattr(settings, "gitea_url", "http://localhost:3000", raising=False)
monkeypatch.setattr(settings, "gitea_owner", "admin", raising=False)
# ---------------------------------------------------------------------------
# TC-13: reviewer comment.
# ---------------------------------------------------------------------------
def test_tc13_reviewer_posts_one_status_comment(
setup_db, fake_worktree, capture_comments, public_url
):
(fake_worktree / "docs" / "work-items" / WID / "12-review.md").write_text(
"---\nverdict: APPROVE\n---\nReviewed.",
)
AgentLauncher()._post_usage_comments(
run_id=99, agent="reviewer", repo=REPO, branch=BRANCH,
usage={"input_tokens": 1, "output_tokens": 1, "cost_usd": 0.01},
duration_s=180,
)
assert len(capture_comments) == 1
post = capture_comments[0]
assert post["wid"] == WID
assert post["author"] == "reviewer"
body = post["body"]
assert "\U0001f50e Reviewer" in body
assert "Verdict: APPROVE" in body
assert "Длительность: 3m 00s" in body
assert "12-review.md" in body
# ---------------------------------------------------------------------------
# TC-14: tester comment.
# ---------------------------------------------------------------------------
def test_tc14_tester_posts_one_status_comment(
setup_db, fake_worktree, capture_comments, public_url
):
(fake_worktree / "docs" / "work-items" / WID / "13-test-report.md").write_text(
"---\nverdict: PASS\n---\n",
)
AgentLauncher()._post_usage_comments(
run_id=100, agent="tester", repo=REPO, branch=BRANCH,
usage=None, duration_s=42,
)
assert len(capture_comments) == 1
post = capture_comments[0]
assert post["author"] == "tester"
body = post["body"]
assert "\U0001f9ea Tester" in body
assert "Verdict: PASS" in body
assert "Длительность: 42s" in body
# ---------------------------------------------------------------------------
# TC-15: deployer comment + per-task summary (two comments, both from deployer).
# ---------------------------------------------------------------------------
def test_tc15_deployer_posts_status_then_summary(
setup_db, fake_worktree, capture_comments, public_url
):
# Task stage = 'deploy' so build_status_comment uses 14-deploy-log.md.
conn = get_db()
conn.execute("UPDATE tasks SET stage='deploy' WHERE id=1")
conn.commit()
conn.close()
(fake_worktree / "docs" / "work-items" / WID / "14-deploy-log.md").write_text(
"---\ndeploy_status: SUCCESS\n---\nDeployed.",
)
AgentLauncher()._post_usage_comments(
run_id=101, agent="deployer", repo=REPO, branch=BRANCH,
usage={"input_tokens": 1, "output_tokens": 1, "cost_usd": 0.01},
duration_s=300,
)
# 2 comments: status + per-task summary.
assert len(capture_comments) == 2
status, summary = capture_comments
assert status["author"] == "deployer"
assert "Status: SUCCESS" in status["body"]
assert "Длительность: 5m 00s" in status["body"]
assert "14-deploy-log.md" in status["body"]
assert summary["author"] == "deployer"
# task_summary_comment header (Russian "Итого по задаче").
assert "\U0001f4ca" in summary["body"]
assert "Итого" in summary["body"]
def test_deployer_staging_picks_15_log(
setup_db, fake_worktree, capture_comments, public_url
):
conn = get_db()
conn.execute("UPDATE tasks SET stage='deploy-staging' WHERE id=1")
conn.commit()
conn.close()
(fake_worktree / "docs" / "work-items" / WID / "15-staging-log.md").write_text(
"---\nstaging_status: SUCCESS\n---\n",
)
AgentLauncher()._post_usage_comments(
run_id=102, agent="deployer", repo=REPO, branch=BRANCH,
usage=None, duration_s=10,
)
# deployer always also posts the summary; check the FIRST comment is status.
assert len(capture_comments) == 2
status = capture_comments[0]
assert "Status: SUCCESS" in status["body"]
assert "15-staging-log.md" in status["body"]
assert "14-deploy-log.md" not in status["body"]
assert "staging-деплой" in status["body"]

View File

@@ -0,0 +1,64 @@
"""ORCH-016 / TC-20 + AC-11: Quality Gates + stage machine are unchanged.
Smoke / change-detector test: the ORCH-016 PR touches comment formatting only.
The QG registry (src/qg/checks.QG_CHECKS) and the stage-machine table
(src/stages.STAGE_TRANSITIONS) MUST remain bit-identical to the contracts the
pipeline depends on. If a future change moves the comment hot path into these
files by accident, this guard breaks first.
"""
import os
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
from src.qg.checks import QG_CHECKS # noqa: E402
from src.stages import STAGE_TRANSITIONS # noqa: E402
# The set of QG names the pipeline DEPLOYS on. Order doesn't matter, identity does.
_EXPECTED_QGS = {
"check_analysis_approved",
"check_analysis_complete",
"check_architecture_done",
"check_ci_green",
"check_review_approved",
"check_tests_passed",
"check_reviewer_verdict",
"check_tests_local",
"check_deploy_status",
"check_staging_status",
}
def test_tc20_qg_registry_unchanged():
assert set(QG_CHECKS.keys()) == _EXPECTED_QGS
def test_tc20_qg_callables_unchanged():
# All entries must be callable — no stub / lambda / None.
for name, fn in QG_CHECKS.items():
assert callable(fn), f"QG {name} is not callable"
# Reference snapshot of STAGE_TRANSITIONS (mirrors what's in docs/architecture
# and src/stages.py — duplicated here on purpose as a regression yardstick).
_EXPECTED_TRANSITIONS = {
"created": {"next": "analysis", "agent": "analyst", "qg": None},
"analysis": {"next": "architecture", "agent": "architect", "qg": "check_analysis_approved"},
"architecture": {"next": "development", "agent": "developer", "qg": "check_architecture_done"},
"development": {"next": "review", "agent": "reviewer", "qg": "check_ci_green"},
"review": {"next": "testing", "agent": "tester", "qg": "check_reviewer_verdict"},
"testing": {"next": "deploy-staging", "agent": "deployer", "qg": "check_tests_passed"},
"deploy-staging": {"next": "deploy", "agent": "deployer", "qg": "check_staging_status"},
"deploy": {"next": "done", "agent": None, "qg": "check_deploy_status"},
"done": {"next": None, "agent": None, "qg": None},
}
def test_tc20_stage_transitions_unchanged():
assert STAGE_TRANSITIONS == _EXPECTED_TRANSITIONS, (
"STAGE_TRANSITIONS drift detected — ORCH-016 must not change the "
"stage machine. Touched stage_engine or stages.py? Update the snapshot "
"in a separate, intentional PR."
)

View File

@@ -0,0 +1,122 @@
"""ORCH-016 / TC-19 + AC-1..AC-5 authorship: status comments use per-agent bots.
When a status comment is posted by AgentLauncher._post_usage_comments, the
underlying plane_sync.add_comment must be invoked with ``author=<agent>`` so
plane_sync._headers_for(<agent>) picks the agent's bot token
(PLANE_BOT_TOKENS[role]) — falling back to PLANE_HEADERS when the bot token
is empty / role unknown. Comment FORMAT changes (ORCH-016) must not affect
that authorship contract.
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_authorship.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, get_db # noqa: E402
from src.agents.launcher import AgentLauncher # noqa: E402
REPO = "enduro-trails"
BRANCH = "feature/ET-016-x"
WID = "ET-016"
@pytest.fixture
def db(monkeypatch):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
conn = get_db()
conn.execute(
"INSERT INTO tasks (id, repo, branch, stage, work_item_id) "
"VALUES (1, ?, ?, 'review', ?)",
(REPO, BRANCH, WID),
)
conn.commit()
conn.close()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
@pytest.fixture
def fake_wt(monkeypatch, tmp_path):
base = tmp_path / "wt"
(base / "docs" / "work-items" / WID).mkdir(parents=True)
monkeypatch.setattr("src.git_worktree.get_worktree_path", lambda r, b: str(base))
return base
@pytest.fixture
def capture(monkeypatch):
posts = []
def _spy(work_item_id, body, author=None, **kwargs):
posts.append({"wid": work_item_id, "body": body, "author": author})
monkeypatch.setattr("src.agents.launcher.plane_add_comment", _spy)
return posts
@pytest.mark.parametrize("agent", ["architect", "developer", "reviewer", "tester"])
def test_tc19_status_comment_carries_agent_author(agent, db, fake_wt, capture):
"""Each agent's status comment must be POST-ed under that agent's bot."""
AgentLauncher()._post_usage_comments(
run_id=1, agent=agent, repo=REPO, branch=BRANCH,
usage=None, duration_s=10,
)
assert len(capture) >= 1
assert capture[0]["author"] == agent, (
f"Expected author={agent!r}, got {capture[0]['author']!r}"
)
def test_tc19_deployer_status_and_summary_both_authored_by_deployer(db, fake_wt, capture):
"""Deployer posts TWO comments (status + per-task summary) — both ``author='deployer'``."""
conn = get_db()
conn.execute("UPDATE tasks SET stage='deploy' WHERE id=1")
conn.commit()
conn.close()
AgentLauncher()._post_usage_comments(
run_id=2, agent="deployer", repo=REPO, branch=BRANCH,
usage=None, duration_s=10,
)
assert len(capture) == 2
assert {c["author"] for c in capture} == {"deployer"}
def test_tc19_headers_for_unknown_role_falls_back(monkeypatch):
"""Ensure plane_sync._headers_for handles unknown agents (fallback contract)."""
from src import plane_sync
h = plane_sync._headers_for("unknown_role_xyz")
# PLANE_HEADERS fallback uses settings.plane_api_token (set to 'test-token').
assert isinstance(h, dict) and "X-API-Key" in h
def test_tc19_status_comment_format_preserves_author_contract(db, fake_wt, capture):
"""The ORCH-016 format change must not strip the author= kw from the call site."""
(fake_wt / "docs" / "work-items" / WID / "12-review.md").write_text(
"---\nverdict: APPROVE\n---\n",
)
AgentLauncher()._post_usage_comments(
run_id=3, agent="reviewer", repo=REPO, branch=BRANCH,
usage={"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0},
duration_s=180,
)
assert len(capture) == 1
post = capture[0]
assert post["author"] == "reviewer"
# And the new format is present in the body (sanity).
assert "\U0001f50e Reviewer" in post["body"]
assert "Verdict: APPROVE" in post["body"]
assert "Длительность: 3m 00s" in post["body"]

View File

@@ -0,0 +1,124 @@
"""ORCH-016 / TC-17 + AC-7: status-comment de-dup contract.
The «one comment per agent per stage» guarantee is enforced upstream of
build_status_comment by:
- the webhook event-dedup table (events.delivery_id PARTIAL UNIQUE, ORCH-5 /
src.db.insert_event_dedup),
- the job queue claim-once contract (src.db.claim_next_job, ORCH-1).
The ORCH-016 PR introduces a new comment FORMAT but must not weaken these
guarantees. This regression test:
1. exercises insert_event_dedup directly to confirm the same delivery_id is
accepted exactly once (sanity for the dedup primitive),
2. exercises build_status_comment to confirm it is a PURE function (same
inputs -> same output), so a retried call from a poorly-isolated test or a
misbehaving caller doesn't silently produce two different comment bodies.
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_dedup_regression.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, insert_event_dedup # noqa: E402
@pytest.fixture(autouse=True)
def setup_db(monkeypatch):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
# ---------------------------------------------------------------------------
# Primitive: event-dedup still rejects a re-delivered webhook.
# ---------------------------------------------------------------------------
def test_tc17_event_dedup_inserts_once_for_same_delivery_id():
"""Two webhook deliveries with the same delivery_id -> one row inserted.
First call returns True (new row); second call returns False (rejected).
This is the primitive every status-comment trigger relies on.
"""
assert insert_event_dedup("plane", "issue.updated", "{}", "delivery-XYZ") is True
assert insert_event_dedup("plane", "issue.updated", "{}", "delivery-XYZ") is False
def test_tc17_event_dedup_distinguishes_delivery_ids():
"""Distinct delivery IDs are independent — two different webhooks both go through."""
assert insert_event_dedup("plane", "issue.updated", "{}", "delivery-A") is True
assert insert_event_dedup("plane", "issue.updated", "{}", "delivery-B") is True
# ---------------------------------------------------------------------------
# Format: build_status_comment is deterministic. A double-fire from buggy code
# still produces an IDENTICAL body -- so the upstream dedup primitive can
# safely treat the second call as no-op without comparing prose.
# ---------------------------------------------------------------------------
def test_tc17_build_status_comment_is_pure(tmp_path):
"""Same inputs produce byte-identical output (deterministic / side-effect free)."""
from src import usage as U
wt = tmp_path / "wt"
(wt / "docs" / "work-items" / "ET-016").mkdir(parents=True)
(wt / "docs" / "work-items" / "ET-016" / "12-review.md").write_text(
"---\nverdict: APPROVE\n---\n",
)
args = dict(
repo="enduro-trails",
branch="feature/ET-016-x",
work_item_id="ET-016",
duration_s=120,
worktree_root=str(wt),
usage={"input_tokens": 100, "output_tokens": 50, "cost_usd": 0.05},
)
a = U.build_status_comment("reviewer", **args)
b = U.build_status_comment("reviewer", **args)
c = U.build_status_comment("reviewer", **args)
assert a == b == c
def test_tc17_build_status_comment_no_db_side_effects(tmp_path):
"""A status-comment build must NOT write to the DB.
Otherwise a webhook-dedup hit would still touch state via the comment
builder. We check by counting rows in `tasks`/`agent_runs`/`jobs` before
and after.
"""
from src import usage as U
from src.db import get_db
conn = get_db()
counts_before = [
conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0],
conn.execute("SELECT COUNT(*) FROM agent_runs").fetchone()[0],
conn.execute("SELECT COUNT(*) FROM jobs").fetchone()[0],
]
conn.close()
U.build_status_comment(
"developer", repo="enduro-trails", branch="b",
work_item_id="ET-016", pr_number=1, duration_s=10,
usage={"input_tokens": 1, "output_tokens": 1, "cost_usd": 0.01},
)
conn = get_db()
counts_after = [
conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0],
conn.execute("SELECT COUNT(*) FROM agent_runs").fetchone()[0],
conn.execute("SELECT COUNT(*) FROM jobs").fetchone()[0],
]
conn.close()
assert counts_before == counts_after

View File

@@ -0,0 +1,145 @@
"""ORCH-016 / TC-24 + TC-25 + AC-14: DB fallback for the duration line.
When build_status_comment is called WITHOUT an explicit duration_s but with a
task_id, it must:
- read the last finished agent_runs row for (task_id, agent),
- compute (julianday(finished_at) - julianday(started_at)) * 86400 in seconds,
- format it via fmt_duration and inject the «Длительность: …» line.
Failure modes (DB locked / row missing / NULL finished_at / negative diff) must
NEVER raise; they simply suppress the duration line and let the rest of the
comment publish.
"""
import os
import tempfile
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
_test_db = os.path.join(tempfile.gettempdir(), "test_orch016_duration_fallback.db")
os.environ["ORCH_DB_PATH"] = _test_db
import pytest # noqa: E402
from src import db as db_module # noqa: E402
from src.db import init_db, get_db # noqa: E402
from src import usage as U # noqa: E402
@pytest.fixture(autouse=True)
def setup_db(monkeypatch):
monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
if os.path.exists(_test_db):
os.unlink(_test_db)
init_db()
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
def _insert_run(task_id, agent, *, seconds_ago_start=None, finished=True):
"""Insert an agent_runs row with controllable timestamps."""
conn = get_db()
if seconds_ago_start is None:
conn.execute(
"INSERT INTO agent_runs (task_id, agent) VALUES (?, ?)",
(task_id, agent),
)
else:
if finished:
conn.execute(
"INSERT INTO agent_runs (task_id, agent, started_at, finished_at) "
"VALUES (?, ?, datetime('now', ?), datetime('now'))",
(task_id, agent, f"-{seconds_ago_start} seconds"),
)
else:
conn.execute(
"INSERT INTO agent_runs (task_id, agent, started_at) "
"VALUES (?, ?, datetime('now', ?))",
(task_id, agent, f"-{seconds_ago_start} seconds"),
)
conn.commit()
conn.close()
# ---------------------------------------------------------------------------
# TC-24: explicit duration_s missing -> DB lookup populates the line.
# ---------------------------------------------------------------------------
def test_tc24_fallback_reads_agent_runs_for_last_finished():
_insert_run(7, "reviewer", seconds_ago_start=240)
secs = U.get_agent_duration(7, "reviewer")
# SQLite's julianday math can be off by a second on either side.
assert secs is not None and abs(secs - 240) <= 1, secs
html = U.build_status_comment("reviewer", task_id=7)
assert any(
s in html for s in (
"Длительность: 4m 00s",
"Длительность: 4m 01s",
"Длительность: 3m 59s",
)
), html
def test_tc24_fallback_picks_last_run_when_multiple():
_insert_run(11, "developer", seconds_ago_start=120)
_insert_run(11, "developer", seconds_ago_start=10)
secs = U.get_agent_duration(11, "developer")
assert secs is not None and abs(secs - 10) <= 1, secs
def test_tc24_no_row_returns_none():
assert U.get_agent_duration(999, "tester") is None
def test_tc24_finished_at_null_returns_none():
_insert_run(13, "tester", seconds_ago_start=100, finished=False)
assert U.get_agent_duration(13, "tester") is None
def test_tc24_missing_args_returns_none():
assert U.get_agent_duration(None, "tester") is None
assert U.get_agent_duration(7, "") is None
assert U.get_agent_duration(0, "tester") is None
# ---------------------------------------------------------------------------
# TC-25: read failure -> logged at debug, NO exception, comment still ships.
# ---------------------------------------------------------------------------
def test_tc25_db_read_failure_no_raise(monkeypatch, caplog):
"""A locked / broken DB must not crash the status comment hot path."""
import logging
def _boom():
raise RuntimeError("simulated DB outage")
monkeypatch.setattr(U, "get_db", _boom)
with caplog.at_level(logging.DEBUG, logger="orchestrator.usage"):
assert U.get_agent_duration(1, "developer") is None
# build_status_comment must still publish (no duration line, no crash).
html = U.build_status_comment("developer", task_id=1, repo="r", branch="b")
assert "Длительность" not in html
assert "\U0001f4bb Developer" in html
# ---------------------------------------------------------------------------
# Sanity: explicit duration_s wins over DB fallback (no SELECT at all).
# ---------------------------------------------------------------------------
def test_explicit_duration_wins_over_db_fallback(monkeypatch):
called = {"n": 0}
real = U.get_agent_duration
def _spy(task_id, agent):
called["n"] += 1
return real(task_id, agent)
monkeypatch.setattr(U, "get_agent_duration", _spy)
_insert_run(5, "architect", seconds_ago_start=300)
html = U.build_status_comment(
"architect", task_id=5, duration_s=12, repo="r", branch="b",
)
assert "Длительность: 12s" in html
# Explicit value supplied -> DB fallback is short-circuited.
assert called["n"] == 0

View File

@@ -0,0 +1,354 @@
"""ORCH-016 / TC-01..TC-10, TC-12, TC-23: unified status comment format.
Unit tests for src.usage.build_status_comment(...) — the single hot path for
every agent's "I just finished a stage" comment in Plane (ADR-001).
Covers:
* Header per agent (icon + role + description from AC-1..AC-5).
* Verdict / Status line read from frontmatter (reviewer / tester / deployer).
* Длительность line shown when duration_s is supplied; suppressed otherwise.
* <a href="..."> link items per agent.
* URL base picks gitea_public_url, falls back to gitea_url.
* Graceful behaviour when files are missing / no frontmatter (AC-8).
No DB / no network — only the worktree filesystem (via tmp_path).
"""
import os
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
import pytest # noqa: E402
from src import usage as U # noqa: E402
WID = "ET-016"
REPO = "enduro-trails"
BRANCH = "feature/ET-016-status-comments"
@pytest.fixture(autouse=True)
def _set_urls(monkeypatch):
"""gitea_public_url is the canonical clickable base (AC-9)."""
monkeypatch.setattr(U, "logger", U.logger)
from src.config import settings
monkeypatch.setattr(settings, "gitea_url", "http://localhost:3000", raising=False)
monkeypatch.setattr(
settings, "gitea_public_url", "https://git.mva154.duckdns.org", raising=False
)
monkeypatch.setattr(settings, "gitea_owner", "admin", raising=False)
yield
def _wt_with_files(tmp_path, files: dict) -> str:
"""Create a worktree skeleton with given files. `files` maps rel-path -> body."""
base = tmp_path / "wt"
docs = base / "docs" / "work-items" / WID
docs.mkdir(parents=True)
for rel, body in files.items():
p = docs / rel if not rel.startswith("/") else base / rel.lstrip("/")
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(body)
return str(base)
# ---------------------------------------------------------------------------
# TC-01: architect comment
# ---------------------------------------------------------------------------
def test_tc01_architect_comment(tmp_path):
wt = _wt_with_files(tmp_path, {"06-adr/ADR-001-x.md": "x"})
html = U.build_status_comment(
"architect",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=312,
worktree_root=wt,
)
# Header
assert "\U0001f4d0 Architect — " in html, html
assert "архитектурную" in html
assert "См. ADR ниже" in html
# Duration: 312s -> 5m 12s
assert "Длительность: 5m 12s" in html
# ADR link via gitea_public_url
assert ("https://git.mva154.duckdns.org/admin/enduro-trails/src/branch/"
f"{BRANCH}/docs/work-items/{WID}/06-adr") in html
# No Verdict for architect
assert "Verdict" not in html
assert "Status:" not in html
# ---------------------------------------------------------------------------
# TC-02: developer comment with PR + branch
# ---------------------------------------------------------------------------
def test_tc02_developer_comment_links_branch_and_pr():
html = U.build_status_comment(
"developer",
repo=REPO, branch=BRANCH, work_item_id=WID,
pr_number=42, duration_s=600,
)
assert "\U0001f4bb Developer — " in html
assert "разработку" in html
# Both branch and PR links
assert f"https://git.mva154.duckdns.org/admin/{REPO}/src/branch/{BRANCH}" in html
assert f"https://git.mva154.duckdns.org/admin/{REPO}/pulls/42" in html
assert f"PR #42" in html
assert "Длительность: 10m 00s" in html
assert "Verdict" not in html
# ---------------------------------------------------------------------------
# TC-03 / TC-04: reviewer verdict via frontmatter
# ---------------------------------------------------------------------------
def test_tc03_reviewer_verdict_approve(tmp_path):
wt = _wt_with_files(tmp_path, {
"12-review.md": "---\nverdict: APPROVE\n---\nbody...",
})
html = U.build_status_comment(
"reviewer",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=120, worktree_root=wt,
)
assert "\U0001f50e Reviewer — " in html
assert "Verdict: APPROVE" in html
assert "Длительность: 2m 00s" in html
assert "12-review.md" in html
def test_tc04_reviewer_verdict_request_changes(tmp_path):
wt = _wt_with_files(tmp_path, {
"12-review.md": "---\nverdict: REQUEST_CHANGES\n---\nblockers...",
})
html = U.build_status_comment(
"reviewer",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=45, worktree_root=wt,
)
assert "Verdict: REQUEST_CHANGES" in html
assert "Длительность: 45s" in html
# ---------------------------------------------------------------------------
# TC-05: reviewer with NO 12-review.md -> graceful (no Verdict, no Review link)
# but Длительность and header still present.
# ---------------------------------------------------------------------------
def test_tc05_reviewer_missing_artifact_graceful(tmp_path):
wt = _wt_with_files(tmp_path, {}) # empty docs dir
html = U.build_status_comment(
"reviewer",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=30, worktree_root=wt,
)
assert "\U0001f50e Reviewer — " in html
assert "Verdict" not in html
# Link to 12-review.md is dropped (AC-8 graceful).
assert "12-review.md" not in html
# Duration still printed when known.
assert "Длительность: 30s" in html
# ---------------------------------------------------------------------------
# TC-06 / TC-07: tester verdict via frontmatter (verdict OR status)
# ---------------------------------------------------------------------------
def test_tc06_tester_pass(tmp_path):
wt = _wt_with_files(tmp_path, {
"13-test-report.md": "---\nverdict: PASS\n---\n",
})
html = U.build_status_comment(
"tester",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=240, worktree_root=wt,
)
assert "\U0001f9ea Tester — " in html
assert "Verdict: PASS" in html
assert "Длительность: 4m 00s" in html
assert "13-test-report.md" in html
def test_tc07_tester_fail(tmp_path):
wt = _wt_with_files(tmp_path, {
"13-test-report.md": "---\nverdict: FAIL\n---\n",
})
html = U.build_status_comment(
"tester",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=240, worktree_root=wt,
)
assert "Verdict: FAIL" in html
assert "Длительность: 4m 00s" in html
def test_tc07b_tester_falls_back_to_status_key(tmp_path):
# Some testers used `status:` instead of `verdict:` (ET-006 / ET-008 pattern).
wt = _wt_with_files(tmp_path, {
"13-test-report.md": "---\nstatus: PASSED\n---\n",
})
html = U.build_status_comment(
"tester",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=10, worktree_root=wt,
)
assert "Verdict: PASSED" in html
# ---------------------------------------------------------------------------
# TC-08 / TC-09: deployer status via frontmatter
# ---------------------------------------------------------------------------
def test_tc08_deployer_deploy_status_success(tmp_path):
wt = _wt_with_files(tmp_path, {
"14-deploy-log.md": "---\ndeploy_status: SUCCESS\n---\n",
})
html = U.build_status_comment(
"deployer",
repo=REPO, branch=BRANCH, work_item_id=WID,
stage="deploy", duration_s=120, worktree_root=wt,
)
assert "\U0001f680 Deployer — " in html
assert "Status: SUCCESS" in html
assert "Длительность: 2m 00s" in html
assert "14-deploy-log.md" in html
def test_tc09_deployer_staging_status_success(tmp_path):
wt = _wt_with_files(tmp_path, {
"15-staging-log.md": "---\nstaging_status: SUCCESS\n---\n",
})
html = U.build_status_comment(
"deployer",
repo=REPO, branch=BRANCH, work_item_id=WID,
stage="deploy-staging", duration_s=60, worktree_root=wt,
)
assert "Status: SUCCESS" in html
assert "Длительность: 1m 00s" in html
# The staging-stage helper links 15-staging-log.md, not 14-deploy-log.md.
assert "15-staging-log.md" in html
assert "14-deploy-log.md" not in html
def test_deployer_status_failed_drives_status_line(tmp_path):
wt = _wt_with_files(tmp_path, {
"14-deploy-log.md": "---\ndeploy_status: FAILED\n---\n",
})
html = U.build_status_comment(
"deployer",
repo=REPO, branch=BRANCH, work_item_id=WID,
stage="deploy", duration_s=5, worktree_root=wt,
)
assert "Status: FAILED" in html
# ---------------------------------------------------------------------------
# TC-10: gitea_public_url is preferred; falls back to gitea_url when empty.
# ---------------------------------------------------------------------------
def test_tc10_url_fallback_to_gitea_url(monkeypatch):
from src.config import settings
monkeypatch.setattr(settings, "gitea_public_url", "", raising=False)
monkeypatch.setattr(settings, "gitea_url", "http://localhost:3000", raising=False)
html = U.build_status_comment(
"developer",
repo=REPO, branch=BRANCH, work_item_id=WID,
pr_number=7, duration_s=15,
)
assert "http://localhost:3000/admin/enduro-trails/pulls/7" in html
# And the public URL is not there because it was empty.
assert "git.mva154.duckdns.org" not in html
# ---------------------------------------------------------------------------
# TC-12: frontmatter parser is graceful — missing file / empty / bad YAML -> None
# (the comment still publishes the header + duration, just no Verdict / Status).
# ---------------------------------------------------------------------------
def test_tc12_frontmatter_missing_file_no_crash(tmp_path):
from src.frontmatter import read_frontmatter_value
assert read_frontmatter_value(str(tmp_path / "nope.md"), "verdict") is None
def test_tc12_frontmatter_empty_no_crash(tmp_path):
p = tmp_path / "empty.md"
p.write_text("")
from src.frontmatter import read_frontmatter_value
assert read_frontmatter_value(str(p), "verdict") is None
def test_tc12_frontmatter_bad_yaml_no_crash(tmp_path):
p = tmp_path / "bad.md"
p.write_text("---\nverdict: [unterminated\n---\nbody")
from src.frontmatter import read_frontmatter_value
assert read_frontmatter_value(str(p), "verdict") is None
def test_tc12_frontmatter_missing_key_returns_none(tmp_path):
p = tmp_path / "ok.md"
p.write_text("---\nother: value\n---\nbody")
from src.frontmatter import read_frontmatter_value
assert read_frontmatter_value(str(p), "verdict") is None
# ---------------------------------------------------------------------------
# TC-23: duration_s=None and no task_id -> the Длительность line is OMITTED.
# Header / description / artifact links remain.
# ---------------------------------------------------------------------------
def test_tc23_no_duration_no_line(tmp_path):
wt = _wt_with_files(tmp_path, {"06-adr/ADR-001-x.md": "x"})
html_none = U.build_status_comment(
"architect",
repo=REPO, branch=BRANCH, work_item_id=WID,
duration_s=None, worktree_root=wt,
)
html_default = U.build_status_comment(
"architect",
repo=REPO, branch=BRANCH, work_item_id=WID,
worktree_root=wt,
)
for html in (html_none, html_default):
assert "Длительность" not in html
# But the header, description and ADR link are still there.
assert "\U0001f4d0 Architect — " in html
assert "архитектурную" in html
assert "06-adr" in html
# ---------------------------------------------------------------------------
# Extra: usage tail is rendered as <sub> when non-zero, suppressed otherwise.
# (Backs up ADR-001 §3 and keeps the old usage_comment test contract.)
# ---------------------------------------------------------------------------
def test_usage_tail_rendered_when_non_zero():
html = U.build_status_comment(
"developer",
repo=REPO, branch=BRANCH, work_item_id=WID,
usage={"input_tokens": 45231, "output_tokens": 12100, "cost_usd": 0.21},
)
assert "<sub>" in html and "</sub>" in html
assert "45.2k in" in html
assert "12.1k out" in html
assert "$0.21" in html
def test_usage_tail_suppressed_when_all_zero():
html = U.build_status_comment("developer", repo=REPO, branch=BRANCH)
assert "<sub>" not in html
# ---------------------------------------------------------------------------
# AC-1 / AC-5 literal strings — fixed wording per role.
# ---------------------------------------------------------------------------
def test_ac1_architect_header_literal():
html = U.build_status_comment("architect", repo=REPO, branch=BRANCH,
work_item_id=WID, duration_s=10)
assert "\U0001f4d0 Architect — " in html
def test_ac5_deployer_deploy_description():
html = U.build_status_comment(
"deployer", repo=REPO, branch=BRANCH, work_item_id=WID, stage="deploy",
)
assert "прод-деплой" in html
def test_ac5_deployer_staging_description():
html = U.build_status_comment(
"deployer", repo=REPO, branch=BRANCH, work_item_id=WID, stage="deploy-staging",
)
assert "staging-деплой" in html