developer(ET): auto-commit from developer run_id=363
All checks were successful
CI / test (push) Successful in 23s
CI / test (pull_request) Successful in 23s

This commit is contained in:
2026-06-08 10:19:42 +00:00
parent 2087475d0d
commit 9e1fe82ef0
16 changed files with 1207 additions and 65 deletions

View File

@@ -0,0 +1,216 @@
"""ORCH-067 — Group B: the Plane-status line on the live card (AC-5..AC-9).
The card now carries an explicit '📍 <Plane status>' line under the header that
follows the ORCH-066 status model. The OFFLINE core (stage->status + In Review
from the brd-clock + Awaiting Deploy) is pure/deterministic and never touches the
network; a best-effort LIVE overlay draws the branch statuses that are
indistinguishable offline (Needs Input / Blocked / …). Everything degrades to the
stage default and NEVER raises (AC-9). Network is isolated: the live-state read
(`_live_state_uuid_cached`) and `get_project_states` are patched per case; the DB
is a temp SQLite.
Test ids TC-05..TC-09 from 04-test-plan.yaml.
"""
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_orchestrator_status_line.db")
os.environ["ORCH_DB_PATH"] = _test_db
from types import SimpleNamespace # noqa: E402
import pytest # noqa: E402
import src.db as db_module # noqa: E402
import src.projects as projects_mod # noqa: E402
from src.db import init_db, get_db # noqa: E402
from src import notifications as N # noqa: E402
import src.plane_sync as plane_sync # noqa: E402
_ORCH_PROJECT_ID = "8da6aa25-a60e-44d6-a1e2-d8ae59aa7d6a"
@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()
# Live overlay OFF by default for the offline-core tests; cases that need it
# turn it back on explicitly. Keep the per-issue cache clean between cases.
monkeypatch.setattr(N._get_settings(), "tracker_live_status", False, raising=False)
N._LIVE_STATE_CACHE.clear()
# Pin repo->project resolution (cross-file ORCH_PROJECTS_JSON reloads must not
# strip 'orchestrator' and disable the live overlay under us).
monkeypatch.setattr(
projects_mod, "get_project_by_repo",
lambda repo: (SimpleNamespace(plane_project_id=_ORCH_PROJECT_ID)
if repo == "orchestrator" else None),
)
yield
if os.path.exists(_test_db):
os.unlink(_test_db)
def _mk_task(stage="development", wid="ORCH-067", repo="orchestrator",
plane_issue_id="issue-uuid-1", brd_started=None, brd_ended=None,
title="status line"):
conn = get_db()
cur = conn.execute(
"INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, title, "
"plane_issue_id, brd_review_started_at, brd_review_ended_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
("p1", wid, repo, "feature/ORCH-067-x", stage, title, plane_issue_id,
brd_started, brd_ended),
)
tid = cur.lastrowid
conn.commit()
conn.close()
return tid
def _status_line(text):
"""Extract the single '📍 ...' status line from rendered card text."""
for ln in text.splitlines():
if ln.startswith("\U0001f4cd"):
return ln
return None
# --------------------------------------------------------------------------- #
# TC-05 / AC-5 — render carries an explicit Plane-status line
# --------------------------------------------------------------------------- #
def test_tc05_render_has_status_line():
tid = _mk_task(stage="development")
text = N.render_task_tracker(tid)
line = _status_line(text)
assert line is not None # '📍 ...' present
assert line == "\U0001f4cd Development" # stage -> Plane status
# --------------------------------------------------------------------------- #
# TC-06 / AC-6 — stage -> Plane status mapping (ТЗ §2.2), parametrized
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("stage,expected", [
("created", "To Analyse"),
("analysis", "Analysis"),
("architecture", "Architecture"),
("development", "Development"),
("review", "Code-Review"),
("testing", "Testing"),
("deploy", "⏸️ Awaiting Deploy — ожидание Confirm Deploy"),
("done", "Done"),
])
def test_tc06_stage_to_plane_status(stage, expected):
# plane_status_label is pure/offline -> assert directly off a row-like dict.
assert N.plane_status_label({"stage": stage}) == expected
def test_tc06_unknown_stage_degrades_to_default():
# Anything unknown -> the safe stage default (To Analyse), never an error.
assert N.plane_status_label({"stage": "weird-stage"}) == "To Analyse"
assert N.plane_status_label({}) == "To Analyse"
# --------------------------------------------------------------------------- #
# TC-07 / AC-7 — In Review from the brd-clock, OFFLINE (no network)
# --------------------------------------------------------------------------- #
def test_tc07_in_review_from_brd_clock(monkeypatch):
# analysis + brd started + not ended -> '⏸️ In Review' (waiting BRD approve).
# Guard: any network read would fail this test -> prove it stays offline.
def _boom(*a, **k):
raise AssertionError("In Review must be resolved OFFLINE (no network)")
monkeypatch.setattr(N, "_live_state_uuid_cached", _boom)
tid = _mk_task(stage="analysis", brd_started="2026-06-08 10:00:00",
brd_ended=None)
text = N.render_task_tracker(tid)
assert _status_line(text) == "\U0001f4cd " + N._IN_REVIEW_LABEL
# The human-gate 'Подтверждение BRD' line with ⏸️/⏳ is still rendered.
assert N._BRD_LABEL in text
assert "" in text # ⏳ still-waiting marker
def test_tc07b_in_review_clears_once_brd_ended():
# Once the BRD review ended, analysis is back to the plain 'Analysis' status.
tid = _mk_task(stage="analysis", brd_started="2026-06-08 10:00:00",
brd_ended="2026-06-08 10:30:00")
assert _status_line(N.render_task_tracker(tid)) == "\U0001f4cd Analysis"
# --------------------------------------------------------------------------- #
# TC-08 / AC-8 — Awaiting Deploy (offline) + Needs Input (live overlay)
# --------------------------------------------------------------------------- #
def test_tc08_awaiting_deploy_offline():
# stage=deploy -> '⏸️ Awaiting Deploy' purely offline (no overlay needed).
tid = _mk_task(stage="deploy")
line = _status_line(N.render_task_tracker(tid))
assert line == "\U0001f4cd ⏸️ Awaiting Deploy — ожидание Confirm Deploy"
def test_tc08_needs_input_via_live_overlay(monkeypatch):
# Needs Input is NOT derivable offline -> drawn by the best-effort overlay
# reading the LIVE Plane status. Patch the live read + the state map.
monkeypatch.setattr(N._get_settings(), "tracker_live_status", True,
raising=False)
monkeypatch.setattr(N, "_live_state_uuid_cached",
lambda issue_id, project_id: "uuid-needs-input")
monkeypatch.setattr(
plane_sync, "get_project_states",
lambda project_id: {"needs_input": "uuid-needs-input"},
)
# repo='orchestrator' resolves to a real registry project_id -> overlay runs.
tid = _mk_task(stage="development", repo="orchestrator")
line = _status_line(N.render_task_tracker(tid))
assert line == "\U0001f4cd ❓ Needs Input — нужны уточнения"
def test_tc08b_overlay_no_match_keeps_offline_base(monkeypatch):
# Live status maps to no branch key -> the offline stage base is kept.
monkeypatch.setattr(N._get_settings(), "tracker_live_status", True,
raising=False)
monkeypatch.setattr(N, "_live_state_uuid_cached",
lambda issue_id, project_id: "uuid-in-progress")
monkeypatch.setattr(
plane_sync, "get_project_states",
lambda project_id: {"in_progress": "uuid-in-progress",
"needs_input": "uuid-needs-input"},
)
tid = _mk_task(stage="development", repo="orchestrator")
assert _status_line(N.render_task_tracker(tid)) == "\U0001f4cd Development"
# --------------------------------------------------------------------------- #
# TC-09 / AC-9, AC-16 — render never raises on broken/unreachable status data
# --------------------------------------------------------------------------- #
def test_tc09_render_survives_overlay_exception(monkeypatch):
# The live overlay blowing up must NOT escape render -> degrade to stage base.
monkeypatch.setattr(N._get_settings(), "tracker_live_status", True,
raising=False)
def _boom(*a, **k):
raise RuntimeError("plane down")
monkeypatch.setattr(N, "_live_state_uuid_cached", _boom)
tid = _mk_task(stage="development", repo="orchestrator")
text = N.render_task_tracker(tid) # must not raise
assert _status_line(text) == "\U0001f4cd Development"
def test_tc09b_card_status_label_never_raises(monkeypatch):
# _card_status_label swallows everything -> a usable default, never an error.
def _boom(*a, **k):
raise RuntimeError("boom")
monkeypatch.setattr(N, "plane_status_label", _boom)
assert N._card_status_label({"stage": "development"}) == "To Analyse"
def test_tc09c_plane_status_label_never_raises():
# Garbage row (None / object without keys) -> safe default, no exception.
assert N.plane_status_label(None) == "To Analyse"
assert N.plane_status_label(object()) == "To Analyse"