1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
"""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 } } }", {})
|
