"""Tests for seed_pearl_workspace. The seeding logic splits into a pure planner (given a workspace snapshot, decide what to create or reconcile) and a thin Linear GraphQL client (the only network boundary). The planner carries the behavior and gets exhaustive pure tests; the client and seed() orchestration get fake-transport tests. The state reconcile is shaped by what Linear actually allows, verified live: teamCreate seeds its own default columns, a team must keep an unstarted state, and the Duplicate state is reserved. So the script never archives -- it renames Linear's defaults into the six Agile columns (repurposing the unstarted Todo as Triage Queue), creates only the genuinely-new Icebox, and leaves Duplicate alone. Positions are set with distinct nonzero values, since Linear ignores a 0.0 position and appends new states at a high one. """ 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(): assert [s["name"] for s in seed.TARGET_STATES] == [ "Icebox", "Triage Queue", "Backlog", "In Progress", "Done", "Canceled"] def test_target_state_types_and_distinct_nonzero_positions(): by_name = {s["name"]: s for s in seed.TARGET_STATES} assert by_name["Icebox"]["type"] == "backlog" assert by_name["Triage Queue"]["type"] == "unstarted" # repurposed from Todo assert by_name["Backlog"]["type"] == "backlog" assert by_name["In Progress"]["type"] == "started" assert by_name["Done"]["type"] == "completed" assert by_name["Canceled"]["type"] == "canceled" positions = [s["position"] for s in seed.TARGET_STATES] assert positions == sorted(positions) assert all(p > 0 for p in positions) # Linear ignores a 0.0 position assert len(set(positions)) == 6 # distinct def test_triage_queue_sources_the_default_todo(): tq = next(s for s in seed.TARGET_STATES if s["name"] == "Triage Queue") assert "Todo" in tq["sources"] 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_column(): 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 state reconcile planner ---------------------------------------- LINEAR_DEFAULTS = [ {"id": "d1", "name": "Backlog", "type": "backlog"}, {"id": "d2", "name": "Todo", "type": "unstarted"}, {"id": "d3", "name": "In Progress", "type": "started"}, {"id": "d4", "name": "Done", "type": "completed"}, {"id": "d5", "name": "Canceled", "type": "canceled"}, {"id": "d6", "name": "Duplicate", "type": "duplicate"}, ] def _plan_by_target(existing): return {p["target"]["name"]: p["match_id"] for p in seed.reconcile_state_plan(existing)} def test_reconcile_against_linear_defaults_renames_and_creates(): plan = _plan_by_target(LINEAR_DEFAULTS) # Icebox is genuinely new; the rest map onto a default (Triage Queue <- Todo). assert plan["Icebox"] is None assert plan["Triage Queue"] == "d2" # the unstarted Todo assert plan["Backlog"] == "d1" assert plan["In Progress"] == "d3" assert plan["Done"] == "d4" assert plan["Canceled"] == "d5" def test_reconcile_is_idempotent_against_already_named_columns(): existing = [{"id": f"s{i}", "name": n, "type": "x"} for i, n in enumerate( ["Icebox", "Triage Queue", "Backlog", "In Progress", "Done", "Canceled"])] plan = _plan_by_target(existing) assert all(mid is not None for mid in plan.values()) # everything matches, nothing created def test_reconcile_matches_case_insensitively(): # the half-seeded board carried a lowercase "icebox" existing = [{"id": "s1", "name": "icebox", "type": "backlog"}] plan = _plan_by_target(existing) assert plan["Icebox"] == "s1" def test_leftover_state_names_reports_unmatched(): # Duplicate isn't a target and stays; report it for the user. assert seed.leftover_state_names(LINEAR_DEFAULTS) == ["Duplicate"] # --- boundary: a fake GraphQL transport --------------------------------------- class FakeTransport: """Records GraphQL calls and replies from a scripted op-name -> 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 _ok(field, payload): return {"data": {field: {"success": True, **payload}}} def test_seed_on_empty_workspace_reconciles_then_builds(): fake = FakeTransport({ "Teams": {"data": {"teams": {"nodes": []}}}, "TeamCreate": _ok("teamCreate", {"team": {"id": "t1", "key": "PEARL", "name": "Pearl"}}), # states re-fetched after create: Linear's defaults "TeamStates": {"data": {"team": {"states": {"nodes": LINEAR_DEFAULTS}}}}, "StateCreate": _ok("workflowStateCreate", {"workflowState": {"id": "new1"}}), "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}), "Projects": {"data": {"projects": {"nodes": []}}}, "ProjectCreate": _ok("projectCreate", {"project": {"id": "p1", "name": "Pearl"}}), "CustomViews": {"data": {"customViews": {"nodes": []}}}, "CustomViewCreate": _ok("customViewCreate", {"customView": {"id": "v1"}}), }) summary = seed.seed(seed.LinearClient("key", transport=fake)) ops = fake.ops() assert ops.count("TeamCreate") == 1 assert ops.count("StateCreate") == 1 # only Icebox is new assert ops.count("StateUpdate") == 6 # six columns positioned/renamed assert ops.count("ProjectCreate") == 1 assert ops.count("CustomViewCreate") == 3 assert summary["team"] == "t1" assert summary["project"] == "p1" assert summary["leftover_states"] == ["Duplicate"] assert "workflowStateArchive" not in ops # never archives def test_seed_is_idempotent_against_a_seeded_workspace(): seeded_states = [{"id": f"s{i}", "name": n, "type": "x"} for i, n in enumerate( ["Icebox", "Triage Queue", "Backlog", "In Progress", "Done", "Canceled", "Duplicate"])] fake = FakeTransport({ "Teams": {"data": {"teams": {"nodes": [ {"id": "t1", "key": "PEARL", "name": "Pearl", "states": {"nodes": seeded_states}}]}}}, "TeamStates": {"data": {"team": {"states": {"nodes": seeded_states}}}}, "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}), "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"}]}}}, }) seed.seed(seed.LinearClient("key", transport=fake)) ops = fake.ops() for create_op in ("TeamCreate", "StateCreate", "ProjectCreate", "CustomViewCreate"): assert create_op not in ops # nothing re-created 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 { teams { nodes { id } } }", {})