"""Tests for the promotion transition (CUST-WP-0061-T03, scripts/promote_intake.py): intake.routed -> workplan | task | decision | engagement. Pure file-manipulation logic (_slugify, _next_number, _append_yaml_block, _inject_field_in_frontmatter) is tested offline. The full promote_intake() flow is tested with the hub API layer mocked (monkeypatched _api_get/ _api_post) so these run without a live server; the real end-to-end proof (actual DB writes) is documented in CUST-WP-0061's progress notes, mirroring the live-verification discipline used for T01/T02. """ from __future__ import annotations import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) import promote_intake as pi # noqa: E402 class TestSlugify: def test_lowercases_and_hyphenates(self): assert pi._slugify("Qonto MCP integration: bank account") == "qonto-mcp-integration-bank-account" def test_truncates_long_titles(self): long_title = "a " * 100 assert len(pi._slugify(long_title)) <= 60 def test_empty_title_gets_placeholder(self): assert pi._slugify("!!!") == "untitled" class TestNextNumber: def test_finds_max_and_increments(self, tmp_path): (tmp_path / "a.md").write_text("id: CUST-WP-0005\nid: CUST-WP-0012\n") (tmp_path / "b.md").write_text("id: CUST-WP-0003\n") import re n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)")) assert n == 13 def test_empty_repo_starts_at_one(self, tmp_path): import re n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)")) assert n == 1 def test_skips_skip_dirs(self, tmp_path): import re skipped = tmp_path / ".git" skipped.mkdir() (skipped / "a.md").write_text("id: CUST-WP-9999\n") n = pi._next_number(tmp_path, re.compile(r"CUST-WP-(\d+)")) assert n == 1 class TestAppendYamlBlock: def test_creates_new_file(self, tmp_path): target = tmp_path / "DecisionQueue.md" pi._append_yaml_block(target, {"id": "X-DEC-2026-001", "title": "test"}, origin_intake_id="abc-123") text = target.read_text() assert "id: \"X-DEC-2026-001\"" in text assert 'origin: "intake:abc-123"' in text def test_appends_to_existing_file(self, tmp_path): target = tmp_path / "DecisionQueue.md" target.write_text("# Decision Queue\n\nSome existing content.\n") pi._append_yaml_block(target, {"id": "X-DEC-2026-002", "title": "second"}, origin_intake_id="def-456") text = target.read_text() assert "Some existing content." in text assert "X-DEC-2026-002" in text class TestInjectFieldInFrontmatter: def test_injects_field_into_frontmatter(self, tmp_path): f = tmp_path / "wp.md" f.write_text("---\nid: CUST-WP-0099\ntitle: x\n---\n\nBody.\n") pi._inject_field_in_frontmatter(f, "state_hub_workstream_id", "019f0000-0000-7000-8000-000000000000") text = f.read_text() assert 'state_hub_workstream_id: "019f0000-0000-7000-8000-000000000000"' in text assert "Body." in text class TestPromoteIntakeMocked: """Full promote_intake() flow with the hub API mocked.""" def _fake_routed_intake(self, **overrides): base = { "id": "019f0000-0000-7000-8000-000000000001", "title": "Qonto MCP mailing — actionable", "status": "routed", "lane": "green", "description": "found in mail triage", } base.update(overrides) return base def test_rejects_intake_not_routed(self, tmp_path, monkeypatch): monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake(status="open")) with pytest.raises(pi.PromotionError, match="not 'routed'"): pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech") def test_promote_to_workplan_writes_file_and_closes_intake(self, tmp_path, monkeypatch): calls = [] def fake_get(api_base, path, *a, **k): if path.startswith("/intakes/"): return self._fake_routed_intake() if path == "/topics": return [{"id": "019ftopic", "domain_slug": "infotech"}] if path == "/repos/testrepo": return {"id": "019frepo", "slug": "testrepo"} return None def fake_post(api_base, path, body): calls.append((path, body)) if path == "/workplans": return {"id": "019fworkplan", "title": body["title"]} if path.startswith("/intakes/") and path.endswith("/close"): return {"id": "019f0000", "status": "closed", "outcome": "promoted"} return {"id": "019fgeneric"} monkeypatch.setattr(pi, "_api_get", fake_get) monkeypatch.setattr(pi, "_api_post", fake_post) result = pi.promote_intake( "http://x", "019f0000-0000-7000-8000-000000000001", "workplan", tmp_path, "testrepo", "infotech", ) assert result["to_kind"] == "workplan" new_file = tmp_path / result["location"] assert new_file.is_file() text = new_file.read_text() assert "Qonto MCP mailing" in text assert 'state_hub_workstream_id: "019fworkplan"' in text assert "intake:019f0000-0000-7000-8000-000000000001" in text close_calls = [c for c in calls if c[0].endswith("/close")] assert len(close_calls) == 1 assert close_calls[0][1]["outcome"] == "promoted" assert close_calls[0][1]["promoted_to"] == result["canonical_id"] def test_promote_to_workplan_raises_loudly_when_repo_lookup_fails(self, tmp_path, monkeypatch): """Regression: the live proof against binky-control (2026-07-21) found /workplans/ requires repo_id, which promote_to_workplan originally never resolved -- the POST 422'd and the failure was silently swallowed (file written, no state_hub_workstream_id, no error). Must raise, not write a half-registered file silently.""" def fake_get(api_base, path, *a, **k): if path.startswith("/intakes/"): return self._fake_routed_intake() if path == "/topics": return [{"id": "019ftopic", "domain_slug": "infotech"}] if path == "/repos/testrepo": return {"_error": "404: not found"} return None monkeypatch.setattr(pi, "_api_get", fake_get) monkeypatch.setattr(pi, "_api_post", lambda *a, **k: {"id": "x"}) with pytest.raises(pi.PromotionError, match="could not look up repo"): pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech") def test_promote_to_workplan_raises_loudly_when_workstream_post_fails(self, tmp_path, monkeypatch): def fake_get(api_base, path, *a, **k): if path.startswith("/intakes/"): return self._fake_routed_intake() if path == "/topics": return [{"id": "019ftopic", "domain_slug": "infotech"}] if path == "/repos/testrepo": return {"id": "019frepo"} return None def fake_post(api_base, path, body): if path == "/workplans": return {"_error": "422: repo_id required"} return {"id": "x"} monkeypatch.setattr(pi, "_api_get", fake_get) monkeypatch.setattr(pi, "_api_post", fake_post) with pytest.raises(pi.PromotionError, match="workstream registration failed"): pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech") def test_promote_to_workplan_refuses_existing_file(self, tmp_path, monkeypatch): monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake()) wp_dir = tmp_path / "workplans" wp_dir.mkdir() # pre-create the file the promotion would try to write prefix = pi.infer_wp_prefix(tmp_path, "testrepo") (wp_dir / f"{prefix}-0001-qonto-mcp-mailing-actionable.md").write_text("already here") def fake_post(api_base, path, body): return {"id": "x"} monkeypatch.setattr(pi, "_api_post", fake_post) with pytest.raises(pi.PromotionError, match="already exists"): pi.promote_intake("http://x", "019f0000", "workplan", tmp_path, "testrepo", "infotech") def test_promote_to_decision_requires_target_file(self, tmp_path, monkeypatch): monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake()) with pytest.raises(pi.PromotionError, match="--target-file"): pi.promote_intake("http://x", "019f0000", "decision", tmp_path, "testrepo", "infotech") def test_promote_to_task_requires_workplan_file(self, tmp_path, monkeypatch): monkeypatch.setattr(pi, "_api_get", lambda *a, **k: self._fake_routed_intake()) with pytest.raises(pi.PromotionError, match="--workplan-file"): pi.promote_intake("http://x", "019f0000", "task", tmp_path, "testrepo", "infotech") def test_promote_to_engagement_is_file_only_no_hub_call(self, tmp_path, monkeypatch): post_calls = [] def fake_get(api_base, path, *a, **k): return self._fake_routed_intake() def fake_post(api_base, path, body): post_calls.append(path) if path.startswith("/intakes/") and path.endswith("/close"): return {"id": "x", "status": "closed"} return {"id": "x"} monkeypatch.setattr(pi, "_api_get", fake_get) monkeypatch.setattr(pi, "_api_post", fake_post) result = pi.promote_intake( "http://x", "019f0000", "engagement", tmp_path, "testrepo", "infotech", target_file="OfficeHourQueue.md", ) assert result["to_kind"] == "engagement" target = tmp_path / "OfficeHourQueue.md" assert target.is_file() assert "queued" in target.read_text() # only the intake-close call should have hit the API — no # engagement-creation endpoint exists yet non_close_calls = [c for c in post_calls if not c.endswith("/close")] assert non_close_calls == []