217 lines
9.3 KiB
Python
217 lines
9.3 KiB
Python
"""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"
|