"""ORCH-089 — plane_sync: label reading + Approved setter (offline, httpx mocked). Covers (04-test-plan.yaml): TC-07 fetch_issue_labels parses the issue's `labels` field; get_project_labels resolves {normalized_name -> uuid}. TC-08 the project label-map is cached with a TTL (a repeat inside the TTL window makes no second GET). TC-09 set_issue_approved PATCHes the issue to the Approved UUID; never-raise. """ import os import tempfile import pytest os.environ.setdefault("ORCH_DB_PATH", os.path.join(tempfile.gettempdir(), "test_ps_labels.db")) os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token") os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token") from unittest.mock import MagicMock # noqa: E402 from src import plane_sync as ps # noqa: E402 def _resp(json_body): m = MagicMock() m.json.return_value = json_body m.raise_for_status.return_value = None return m @pytest.fixture(autouse=True) def fresh_cache(monkeypatch): ps.reload_project_labels() monkeypatch.setattr(ps, "_resolve_project_id", lambda w=None, p=None: "proj-1") monkeypatch.setattr(ps.settings, "auto_label_states_ttl_s", 300, raising=False) yield ps.reload_project_labels() # --- TC-07: fetch_issue_labels + get_project_labels ------------------------ def test_tc07_fetch_issue_labels(monkeypatch): monkeypatch.setattr(ps, "find_issue_id", lambda w, p=None: "issue-uuid") monkeypatch.setattr( ps.httpx, "get", lambda *a, **k: _resp({"labels": ["uuid-A", "uuid-B"]}), ) assert ps.fetch_issue_labels("ORCH-1") == ["uuid-A", "uuid-B"] def test_tc07_fetch_issue_labels_not_found(monkeypatch): # Issue not resolvable -> None (distinct from [] = "no labels"). monkeypatch.setattr(ps, "find_issue_id", lambda w, p=None: None) assert ps.fetch_issue_labels("ORCH-404") is None def test_tc07_fetch_issue_labels_api_error(monkeypatch): monkeypatch.setattr(ps, "find_issue_id", lambda w, p=None: "issue-uuid") monkeypatch.setattr(ps.httpx, "get", MagicMock(side_effect=Exception("boom"))) assert ps.fetch_issue_labels("ORCH-1") is None # never-raise def test_tc07_get_project_labels_normalized(monkeypatch): monkeypatch.setattr( ps.httpx, "get", lambda *a, **k: _resp({"results": [ {"id": "uuid-A", "name": "autoApprove"}, {"id": "uuid-B", "name": "Auto Deploy"}, ]}), ) m = ps.get_project_labels("proj-1") assert m["autoapprove"] == "uuid-A" assert m["auto deploy"] == "uuid-B" def test_tc07_get_project_labels_ambiguous(monkeypatch): # Two distinct labels collapse to the same normalized name -> sentinel. monkeypatch.setattr( ps.httpx, "get", lambda *a, **k: _resp([ {"id": "uuid-A", "name": "autoApprove"}, {"id": "uuid-B", "name": "AUTOAPPROVE"}, ]), ) m = ps.get_project_labels("proj-1") assert m["autoapprove"] == "__AMBIGUOUS__" def test_tc07_get_project_labels_api_error_empty(monkeypatch): monkeypatch.setattr(ps.httpx, "get", MagicMock(side_effect=Exception("down"))) assert ps.get_project_labels("proj-1") == {} # never-raise, no cache -> {} # --- TC-08: TTL cache ------------------------------------------------------ def test_tc08_label_map_cached_within_ttl(monkeypatch): clock = {"t": 1000.0} monkeypatch.setattr(ps.time, "monotonic", lambda: clock["t"]) mock_get = MagicMock(side_effect=lambda *a, **k: _resp( {"results": [{"id": "uuid-A", "name": "autoApprove"}]} )) monkeypatch.setattr(ps.httpx, "get", mock_get) ps.get_project_labels("proj-1") ps.get_project_labels("proj-1") # within TTL -> served from cache assert mock_get.call_count == 1 # Past the TTL -> refetch. clock["t"] += 301 ps.get_project_labels("proj-1") assert mock_get.call_count == 2 def test_tc08_ttl_zero_lifetime_cache(monkeypatch): monkeypatch.setattr(ps.settings, "auto_label_states_ttl_s", 0, raising=False) clock = {"t": 1000.0} monkeypatch.setattr(ps.time, "monotonic", lambda: clock["t"]) mock_get = MagicMock(side_effect=lambda *a, **k: _resp( [{"id": "uuid-A", "name": "autoApprove"}] )) monkeypatch.setattr(ps.httpx, "get", mock_get) ps.get_project_labels("proj-1") clock["t"] += 100000 ps.get_project_labels("proj-1") assert mock_get.call_count == 1 # lifetime cache, never expires def test_tc08_stale_served_on_refresh_failure(monkeypatch): clock = {"t": 1000.0} monkeypatch.setattr(ps.time, "monotonic", lambda: clock["t"]) responses = iter([ _resp({"results": [{"id": "uuid-A", "name": "autoApprove"}]}), Exception("transient"), ]) def flaky(*a, **k): r = next(responses) if isinstance(r, Exception): raise r return r monkeypatch.setattr(ps.httpx, "get", flaky) ps.get_project_labels("proj-1") clock["t"] += 301 # force a refresh that fails -> stale map served m = ps.get_project_labels("proj-1") assert m["autoapprove"] == "uuid-A" # --- TC-09: set_issue_approved --------------------------------------------- def test_tc09_set_issue_approved_patches_approved_uuid(monkeypatch): monkeypatch.setattr(ps, "get_project_states", lambda pid: {"approved": "approved-uuid"}) monkeypatch.setattr(ps, "find_issue_id", lambda w, p=None: "issue-uuid") patch_spy = MagicMock(return_value=_resp({})) monkeypatch.setattr(ps.httpx, "patch", patch_spy) ps.set_issue_approved("ORCH-1") patch_spy.assert_called_once() assert patch_spy.call_args.kwargs["json"] == {"state": "approved-uuid"} def test_tc09_set_issue_approved_never_raises(monkeypatch): monkeypatch.setattr(ps, "get_project_states", lambda pid: {"approved": "approved-uuid"}) monkeypatch.setattr(ps, "find_issue_id", lambda w, p=None: "issue-uuid") monkeypatch.setattr(ps.httpx, "patch", MagicMock(side_effect=Exception("boom"))) # Must not raise. ps.set_issue_approved("ORCH-1")