"""Tests for scripts/dev/publish_issue.py — one-way staging→production ticket publish (#186). Uses an in-memory fake HubClient so these tests never shell out to a real `muse hub` CLI or hit a network. Real wiring (MuseHubCliClient) is exercised manually against staging/production, not here. """ import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts" / "dev")) from publish_issue import StagingUrlLeakError, build_mirror_marker, publish_issue # noqa: E402 STAGING_HUB = "https://staging.musehub.ai" PRODUCTION_HUB = "https://musehub.ai" class FakeHubClient: """In-memory stand-in for the real `muse hub issue` CLI calls.""" def __init__(self) -> None: self._issues: dict[tuple[str, int], dict] = {} self._next_number: dict[str, int] = {} def seed(self, hub: str, number: int, *, title: str, body: str, labels: list[str]) -> None: self._issues[(hub, number)] = { "number": number, "title": title, "body": body, "labels": labels, "url": f"{hub}/gabriel/musehub/issues/{number}", } self._next_number[hub] = max(self._next_number.get(hub, 0), number) + 1 def read_issue(self, number: int, hub: str) -> dict: return dict(self._issues[(hub, number)]) def list_issues(self, hub: str, state: str = "all") -> list[dict]: return [dict(v) for (h, _), v in self._issues.items() if h == hub] def create_issue(self, hub: str, *, title: str, body: str, labels: list[str]) -> dict: number = self._next_number.get(hub, 1) self._next_number[hub] = number + 1 issue = { "number": number, "title": title, "body": body, "labels": labels, "url": f"{hub}/gabriel/musehub/issues/{number}", } self._issues[(hub, number)] = issue return dict(issue) def update_issue(self, hub: str, number: int, *, title: str, body: str) -> dict: issue = self._issues[(hub, number)] issue["title"] = title issue["body"] = body return dict(issue) def set_labels(self, hub: str, number: int, labels: list[str]) -> None: self._issues[(hub, number)]["labels"] = labels @pytest.fixture def client() -> FakeHubClient: return FakeHubClient() class TestPublishIssue: def test_fresh_publish_creates_exactly_one_production_issue(self, client: FakeHubClient) -> None: client.seed(STAGING_HUB, 42, title="feat: thing", body="Do the thing.", labels=["enhancement"]) result = publish_issue(client, 42, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert result["action"] == "created" prod_issues = client.list_issues(PRODUCTION_HUB) assert len(prod_issues) == 1 assert prod_issues[0]["title"] == "feat: thing" assert "Do the thing." in prod_issues[0]["body"] assert build_mirror_marker(42) in prod_issues[0]["body"] assert prod_issues[0]["labels"] == ["enhancement"] def test_republishing_same_staging_issue_updates_rather_than_duplicates(self, client: FakeHubClient) -> None: client.seed(STAGING_HUB, 42, title="feat: thing", body="Do the thing.", labels=["enhancement"]) publish_issue(client, 42, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) # Staging ticket gets edited after the first publish. client.seed(STAGING_HUB, 42, title="feat: thing (updated)", body="Do the thing, better.", labels=["enhancement", "bug"]) result = publish_issue(client, 42, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert result["action"] == "updated" prod_issues = client.list_issues(PRODUCTION_HUB) assert len(prod_issues) == 1, "must not create a second production issue for the same staging ticket" assert prod_issues[0]["title"] == "feat: thing (updated)" assert "better" in prod_issues[0]["body"] assert prod_issues[0]["labels"] == ["enhancement", "bug"] def test_different_staging_issues_produce_separate_production_issues(self, client: FakeHubClient) -> None: client.seed(STAGING_HUB, 1, title="A", body="body A", labels=[]) client.seed(STAGING_HUB, 2, title="B", body="body B", labels=[]) publish_issue(client, 1, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) publish_issue(client, 2, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert len(client.list_issues(PRODUCTION_HUB)) == 2 def test_staging_url_in_body_raises_and_blocks_publish(self, client: FakeHubClient) -> None: """A hard block, not a warning — a warning is exactly the thing that gets skimmed past at the moment someone's about to post publicly.""" client.seed( STAGING_HUB, 7, title="bug: X", body="See https://staging.musehub.ai/gabriel/musehub/issues/6 for context.", labels=[], ) with pytest.raises(StagingUrlLeakError, match="staging.musehub.ai"): publish_issue(client, 7, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert client.list_issues(PRODUCTION_HUB) == [], ( "must not publish anything when the body leaks a staging URL" ) def test_clean_body_has_no_warnings(self, client: FakeHubClient) -> None: client.seed(STAGING_HUB, 8, title="bug: Y", body="No staging links here.", labels=[]) result = publish_issue(client, 8, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert result["warnings"] == [] def test_bare_issue_reference_in_body_triggers_warning(self, client: FakeHubClient) -> None: client.seed( STAGING_HUB, 9, title="bug: Z", body="Related to #182 and #183, see also the discussion there.", labels=[], ) result = publish_issue(client, 9, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert any("#182" in w and "#183" in w for w in result["warnings"]) def test_markdown_heading_does_not_trigger_bare_reference_warning(self, client: FakeHubClient) -> None: client.seed( STAGING_HUB, 10, title="bug: W", body="## Summary\n\nNo real issue references here, just headings.", labels=[], ) result = publish_issue(client, 10, staging_hub=STAGING_HUB, production_hub=PRODUCTION_HUB) assert result["warnings"] == [] def test_mirror_marker_is_stable_per_staging_number(self) -> None: assert build_mirror_marker(42) == build_mirror_marker(42) assert build_mirror_marker(42) != build_mirror_marker(43)