1 files changed, 217 insertions, 0 deletions
diff --git a/scripts/tests/test_seed_pearl_workspace.py b/scripts/tests/test_seed_pearl_workspace.py
new file mode 100644
index 0000000..2e7cb4a
--- /dev/null
+++ b/scripts/tests/test_seed_pearl_workspace.py
@@ -0,0 +1,217 @@
+"""Tests for seed_pearl_workspace.
+
+The seeding logic splits into a pure planner (given a workspace snapshot, decide
+what to create) and a thin Linear GraphQL client (the only network boundary).
+The planner carries the behavior, so it gets exhaustive pure tests; the client
+and the seed() orchestration get fake-transport tests that assert the right
+mutations go out and that a second run is a no-op (idempotence).
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import seed_pearl_workspace as seed
+
+
+# --- pure: target definitions -------------------------------------------------
+
+def test_target_states_are_the_six_agile_columns_in_order():
+    names = [s["name"] for s in seed.TARGET_STATES]
+    assert names == ["icebox", "triage queue", "backlog",
+                     "in-progress", "done", "cancelled"]
+
+
+def test_early_columns_share_backlog_type_with_ascending_positions():
+    # icebox / triage queue / backlog all sort by position within one type, so
+    # pearl's (type-rank, position) grouping reproduces the board column order.
+    early = [s for s in seed.TARGET_STATES
+             if s["name"] in ("icebox", "triage queue", "backlog")]
+    assert all(s["type"] == "backlog" for s in early)
+    positions = [s["position"] for s in early]
+    assert positions == sorted(positions)
+    assert len({s["position"] for s in seed.TARGET_STATES}) == 6  # all distinct
+
+
+def test_later_columns_map_to_distinct_linear_types():
+    by_name = {s["name"]: s["type"] for s in seed.TARGET_STATES}
+    assert by_name["in-progress"] == "started"
+    assert by_name["done"] == "completed"
+    assert by_name["cancelled"] == "canceled"
+
+
+def test_every_state_has_a_color():
+    assert all(s.get("color") for s in seed.TARGET_STATES)
+
+
+def test_view_specs_names_and_project_scope():
+    specs = seed.desired_views("proj-1")
+    assert [v["name"] for v in specs] == [
+        "Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]
+    for v in specs:
+        assert v["filterData"]["project"]["id"]["eq"] == "proj-1"
+
+
+def test_open_view_excludes_completed_and_cancelled():
+    open_view = next(v for v in seed.desired_views("p")
+                     if v["name"] == "Pearl Open Issues")
+    nin = open_view["filterData"]["state"]["type"]["nin"]
+    assert "completed" in nin and "canceled" in nin
+
+
+def test_icebox_and_inbox_views_filter_to_their_state():
+    views = {v["name"]: v for v in seed.desired_views("p")}
+    assert views["Pearl Icebox"]["filterData"]["state"]["name"]["eq"] == "icebox"
+    assert views["Pearl Inbox"]["filterData"]["state"]["name"]["eq"] == "triage queue"
+
+
+# --- pure: the planner --------------------------------------------------------
+
+EMPTY = {"team": None, "states": [], "project": None, "views": []}
+
+
+def test_plan_on_empty_workspace_creates_everything():
+    plan = seed.plan_seed(EMPTY)
+    assert plan.create_team is True
+    assert [s["name"] for s in plan.create_states] == [s["name"] for s in seed.TARGET_STATES]
+    assert plan.create_project is True
+    assert [v["name"] for v in plan.create_views] == [
+        "Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]
+
+
+def test_plan_is_idempotent_when_everything_exists():
+    current = {
+        "team": {"id": "t1", "name": "Pearl"},
+        "states": [{"name": n} for n in
+                   ["icebox", "triage queue", "backlog", "in-progress", "done", "cancelled"]],
+        "project": {"id": "p1", "name": "Pearl"},
+        "views": [{"name": n} for n in
+                  ["Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]],
+    }
+    plan = seed.plan_seed(current)
+    assert plan.create_team is False
+    assert plan.create_states == []
+    assert plan.create_project is False
+    assert plan.create_views == []
+
+
+def test_plan_creates_only_missing_states_and_views():
+    current = {
+        "team": {"id": "t1", "name": "Pearl"},
+        "states": [{"name": "backlog"}, {"name": "done"}],
+        "project": {"id": "p1", "name": "Pearl"},
+        "views": [{"name": "Pearl Icebox"}],
+    }
+    plan = seed.plan_seed(current)
+    assert plan.create_team is False
+    assert plan.create_project is False
+    assert [s["name"] for s in plan.create_states] == [
+        "icebox", "triage queue", "in-progress", "cancelled"]
+    assert [v["name"] for v in plan.create_views] == [
+        "Pearl Open Issues", "Pearl Inbox"]
+
+
+# --- boundary: a fake GraphQL transport ---------------------------------------
+
+class FakeTransport:
+    """Records GraphQL calls and replies from a scripted op->response map."""
+
+    def __init__(self, responses):
+        self.responses = responses
+        self.calls = []
+
+    def __call__(self, query, variables):
+        op = next(name for name in self.responses if name in query)
+        self.calls.append((op, variables))
+        resp = self.responses[op]
+        return resp(variables) if callable(resp) else resp
+
+    def ops(self):
+        return [op for op, _ in self.calls]
+
+
+def test_seed_on_empty_workspace_issues_creates_in_order():
+    state_n = {"i": 0}
+
+    def make_state(_v):
+        state_n["i"] += 1
+        return {"data": {"workflowStateCreate":
+                         {"success": True, "workflowState": {"id": f"s{state_n['i']}"}}}}
+
+    fake = FakeTransport({
+        # snapshot queries: empty workspace
+        "teams": {"data": {"teams": {"nodes": []}}},
+        "teamCreate": {"data": {"teamCreate": {"success": True,
+                       "team": {"id": "t1", "key": "PEARL", "name": "Pearl",
+                                "states": {"nodes": []}}}}},
+        "workflowStateCreate": make_state,
+        "projects": {"data": {"projects": {"nodes": []}}},
+        "projectCreate": {"data": {"projectCreate":
+                          {"success": True, "project": {"id": "p1", "name": "Pearl"}}}},
+        "customViews": {"data": {"customViews": {"nodes": []}}},
+        "customViewCreate": {"data": {"customViewCreate":
+                             {"success": True, "customView": {"id": "v1"}}}},
+    })
+    client = seed.LinearClient("key", transport=fake)
+    summary = seed.seed(client)
+
+    assert fake.ops().count("teamCreate") == 1
+    assert fake.ops().count("workflowStateCreate") == 6
+    assert fake.ops().count("projectCreate") == 1
+    assert fake.ops().count("customViewCreate") == 3
+    assert summary["team"] == "t1"
+    assert summary["project"] == "p1"
+
+
+def test_seed_is_idempotent_against_a_populated_workspace():
+    populated = {
+        "teams": {"data": {"teams": {"nodes": [
+            {"id": "t1", "key": "PEARL", "name": "Pearl",
+             "states": {"nodes": [{"id": f"s{i}", "name": n}
+                                  for i, n in enumerate(
+                                      ["icebox", "triage queue", "backlog",
+                                       "in-progress", "done", "cancelled"])]}}]}}},
+        "projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}},
+        "customViews": {"data": {"customViews": {"nodes": [
+            {"id": "v1", "name": "Pearl Open Issues"},
+            {"id": "v2", "name": "Pearl Icebox"},
+            {"id": "v3", "name": "Pearl Inbox"}]}}},
+    }
+    fake = FakeTransport(populated)
+    client = seed.LinearClient("key", transport=fake)
+    seed.seed(client)
+    for create_op in ("teamCreate", "workflowStateCreate",
+                      "projectCreate", "customViewCreate"):
+        assert create_op not in fake.ops()
+
+
+def test_seed_reports_leftover_default_states_without_archiving():
+    # An existing team carrying the six columns plus Linear's leftover defaults:
+    # seed creates nothing and surfaces the defaults for manual tidy-up, and
+    # never issues an archive.
+    populated = {
+        "teams": {"data": {"teams": {"nodes": [
+            {"id": "t1", "key": "PEARL", "name": "Pearl",
+             "states": {"nodes": [{"id": f"s{i}", "name": n} for i, n in enumerate(
+                 ["icebox", "triage queue", "backlog", "in-progress", "done",
+                  "cancelled", "Todo", "In Review"])]}}]}}},
+        "projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}},
+        "customViews": {"data": {"customViews": {"nodes": [
+            {"id": "v1", "name": "Pearl Open Issues"},
+            {"id": "v2", "name": "Pearl Icebox"},
+            {"id": "v3", "name": "Pearl Inbox"}]}}},
+    }
+    fake = FakeTransport(populated)
+    summary = seed.seed(seed.LinearClient("key", transport=fake))
+    assert summary["leftover_states"] == ["Todo", "In Review"]
+    assert "workflowStateArchive" not in fake.ops()
+
+
+def test_client_raises_on_graphql_errors():
+    fake = FakeTransport({"teams": {"errors": [{"message": "bad key"}]}})
+    client = seed.LinearClient("key", transport=fake)
+    with pytest.raises(seed.LinearError, match="bad key"):
+        client.execute("query { teams { nodes { id } } }", {})