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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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 } } }", {})
|
