"""Tests for seed_pearl_workspace. The seeding logic splits into a pure planner (given a workspace snapshot, decide what to create, recreate, 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, the Duplicate state is reserved, and a state's type is immutable. Because the board orders columns by type category first, the columns need specific types (Icebox and Triage backlog, Backlog unstarted) to render in board order, and a column found with the wrong type has to be recreated -- create with the right type, move its issues, archive the old one. Positions are set with distinct nonzero values, since Linear ignores a 0.0 position and appends new states high. """ 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", "Backlog", "In Progress", "Done", "Canceled"] def test_target_state_types_put_backlog_last_in_the_pre_work_group(): by_name = {s["name"]: s for s in seed.TARGET_STATES} # Icebox and Triage are backlog-type; Backlog is the unstarted state, which # Linear renders after the backlog columns -> board order Icebox, Triage, Backlog. assert by_name["Icebox"]["type"] == "backlog" assert by_name["Triage"]["type"] == "backlog" assert by_name["Backlog"]["type"] == "unstarted" 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_sources_take_over_the_right_defaults_on_a_fresh_team(): by_name = {s["name"]: s for s in seed.TARGET_STATES} # On a fresh team Icebox takes over the default Backlog and Backlog takes # over the unstarted Todo (keeping its type). assert "Backlog" in by_name["Icebox"]["sources"] assert "Todo" in by_name["Backlog"]["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" # --- 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 _entries_by_target(existing): return {p["target"]["name"]: p for p in seed.reconcile_state_plan(existing)} def test_reconcile_against_linear_defaults_reuses_defaults_and_creates_triage(): plan = _entries_by_target(LINEAR_DEFAULTS) # Icebox takes the default backlog Backlog; Backlog takes the unstarted Todo # (both type-correct -> update). Triage is genuinely new. assert plan["Icebox"]["action"] == "update" and plan["Icebox"]["match_id"] == "d1" assert plan["Backlog"]["action"] == "update" and plan["Backlog"]["match_id"] == "d2" assert plan["Triage"]["action"] == "create" assert plan["In Progress"]["match_id"] == "d3" assert plan["Done"]["match_id"] == "d4" assert plan["Canceled"]["match_id"] == "d5" def test_reconcile_exact_name_wins_over_a_source_alias(): # A live "Backlog" must be claimed by the Backlog target, not stolen by # Icebox's "Backlog" source alias. existing = [{"id": "s1", "name": "Backlog", "type": "unstarted"}, {"id": "s2", "name": "Icebox", "type": "backlog"}] plan = _entries_by_target(existing) assert plan["Backlog"]["match_id"] == "s1" assert plan["Icebox"]["match_id"] == "s2" def test_reconcile_marks_a_wrong_type_match_for_recreate(): # The live workspace before the fix: Triage is unstarted, Backlog is backlog # -- both the wrong type, so both must be recreated. existing = [ {"id": "s1", "name": "Icebox", "type": "backlog"}, {"id": "s2", "name": "Triage", "type": "unstarted"}, {"id": "s3", "name": "Backlog", "type": "backlog"}, {"id": "s4", "name": "In Progress", "type": "started"}, {"id": "s5", "name": "Done", "type": "completed"}, {"id": "s6", "name": "Canceled", "type": "canceled"}, ] plan = _entries_by_target(existing) assert plan["Triage"]["action"] == "recreate" and plan["Triage"]["match_id"] == "s2" assert plan["Backlog"]["action"] == "recreate" and plan["Backlog"]["match_id"] == "s3" assert plan["Icebox"]["action"] == "update" def test_reconcile_is_idempotent_against_correctly_typed_columns(): correct = [("Icebox", "backlog"), ("Triage", "backlog"), ("Backlog", "unstarted"), ("In Progress", "started"), ("Done", "completed"), ("Canceled", "canceled")] existing = [{"id": f"s{i}", "name": n, "type": t} for i, (n, t) in enumerate(correct)] plan = _entries_by_target(existing) assert all(e["action"] == "update" for e in plan.values()) def test_reconcile_matches_case_insensitively(): existing = [{"id": "s1", "name": "icebox", "type": "backlog"}] plan = _entries_by_target(existing) assert plan["Icebox"]["match_id"] == "s1" def test_leftover_state_names_reports_unmatched(): # Duplicate isn't a target and stays; Todo is claimed (by Backlog), so only # Duplicate is reported. 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"}}), "TeamUpdate": _ok("teamUpdate", {}), "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 Triage is new assert ops.count("StateUpdate") == 6 # six columns named/positioned assert ops.count("StateArchive") == 0 # nothing to recreate on a fresh team assert ops.count("TeamUpdate") == 1 # default new-issue state -> Triage assert ops.count("ProjectCreate") == 1 assert ops.count("CustomViewCreate") == 3 assert summary["team"] == "t1" assert summary["project"] == "p1" assert summary["states_created"] == ["Triage"] assert summary["states_recreated"] == [] assert summary["leftover_states"] == ["Duplicate"] assert summary["default_issue_state"] == "Triage" # the default points at the freshly-created Triage team_update = next(v for op, v in fake.calls if op == "TeamUpdate") assert team_update["input"]["defaultIssueStateId"] == "new1" def test_seed_heals_type_drift_by_recreating_and_migrating(): # The live workspace before the fix: Triage unstarted (the old default, # empty), Backlog backlog (holds two issues). Both wrong type -> recreate. live_states = [ {"id": "s1", "name": "Icebox", "type": "backlog", "position": 1.0}, {"id": "s2", "name": "Triage", "type": "unstarted", "position": 2.0}, {"id": "s3", "name": "Backlog", "type": "backlog", "position": 3.0}, {"id": "s4", "name": "In Progress", "type": "started", "position": 4.0}, {"id": "s5", "name": "Done", "type": "completed", "position": 5.0}, {"id": "s6", "name": "Canceled", "type": "canceled", "position": 6.0}, {"id": "s7", "name": "Duplicate", "type": "duplicate", "position": 5.0}, ] new_ids = iter(["nTriage", "nBacklog"]) def make_state(_v): return _ok("workflowStateCreate", {"workflowState": {"id": next(new_ids)}}) def state_issues(v): nodes = [{"id": "i1"}, {"id": "i2"}] if v["id"] == "s3" else [] return {"data": {"workflowState": {"issues": { "nodes": nodes, "pageInfo": {"hasNextPage": False, "endCursor": None}}}}} fake = FakeTransport({ "Teams": {"data": {"teams": {"nodes": [ {"id": "t1", "key": "PEARL", "name": "Pearl", "defaultIssueState": {"id": "s2"}, "states": {"nodes": live_states}}]}}}, "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}), "StateCreate": make_state, "StateIssues": state_issues, "IssueMove": _ok("issueUpdate", {"issue": {"id": "i"}}), "StateArchive": _ok("workflowStateArchive", {}), "TeamUpdate": _ok("teamUpdate", {}), "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"}]}}}, }) summary = seed.seed(seed.LinearClient("key", transport=fake)) ops = fake.ops() assert ops.count("StateCreate") == 2 # new Triage + new Backlog assert ops.count("StateArchive") == 2 # old Triage + old Backlog assert ops.count("IssueMove") == 2 # the two issues off old Backlog # every create happens before any archive, so a new unstarted state exists # before the old unstarted one is archived assert max(i for i, op in enumerate(ops) if op == "StateCreate") \ < min(i for i, op in enumerate(ops) if op == "StateArchive") assert summary["states_recreated"] == ["Triage", "Backlog"] assert summary["issues_moved"] == 2 # the issues land on the new Backlog, and the default moves to the new Triage moves = [v for op, v in fake.calls if op == "IssueMove"] assert all(m["state"] == "nBacklog" for m in moves) team_update = next(v for op, v in fake.calls if op == "TeamUpdate") assert team_update["input"]["defaultIssueStateId"] == "nTriage" def test_seed_is_idempotent_against_a_correctly_seeded_workspace(): correct = [("Icebox", "backlog"), ("Triage", "backlog"), ("Backlog", "unstarted"), ("In Progress", "started"), ("Done", "completed"), ("Canceled", "canceled"), ("Duplicate", "duplicate")] seeded_states = [{"id": f"s{i}", "name": n, "type": t, "position": float(i + 1)} for i, (n, t) in enumerate(correct)] fake = FakeTransport({ # default already points at Triage (s1), so no team update fires either "Teams": {"data": {"teams": {"nodes": [ {"id": "t1", "key": "PEARL", "name": "Pearl", "defaultIssueState": {"id": "s1"}, "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 absent in ("TeamCreate", "StateCreate", "StateArchive", "ProjectCreate", "CustomViewCreate", "TeamUpdate"): assert absent not in ops # nothing created, recreated, or repointed 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 } } }", {})