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
|
"""Tests for import_org_backlog.
The importer splits into a pure org parser + mapper (no I/O) and a thin Linear
GraphQL client. The parser pulls top-level TODO/DOING tasks out of the "Pearl
Open Work" section; the mapper turns each into an issueCreate input (priority,
state, labels); the client creates missing labels and the issues, skipping a
task whose title already exists so a re-run is safe.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import import_org_backlog as imp
SAMPLE = """\
* Pearl Priority Scheme
- =[#A]= urgent (not a task)
* Pearl Open Work
** DOING [#A] Personal Account Dogfooding
Meta task, skipped.
*** TODO a child
child body
** TODO [#B] Fix the thing :bug:discuss:
:PROPERTIES:
:LAST_REVIEWED: 2026-06-06
:END:
The body explaining the bug.
More body.
*** 2026-06-06 dated log
log entry, excluded from the description
** TODO [#C] A docs task :docs:quick:
Body here.
** TODO [#B] Manual testing and validation :test:
checklist umbrella, skipped
*** TODO a verify
verify body
** TODO [#D] Someday feature :feature:
Someday body.
* Pearl Resolved
** DONE [#A] old finished thing
done body, excluded (resolved section)
"""
# --- pure: parser ------------------------------------------------------------
def test_parse_pulls_open_work_tasks_skipping_meta_and_resolved():
tasks = imp.parse_open_work_tasks(SAMPLE)
assert [t["title"] for t in tasks] == [
"Fix the thing", "A docs task", "Someday feature"]
def test_parse_captures_priority_and_tags():
tasks = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)}
assert tasks["Fix the thing"]["priority"] == "B"
assert tasks["Fix the thing"]["tags"] == ["bug", "discuss"]
assert tasks["Someday feature"]["priority"] == "D"
assert tasks["A docs task"]["tags"] == ["docs", "quick"]
def test_parse_body_excludes_drawer_and_subtrees():
tasks = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)}
body = tasks["Fix the thing"]["body"]
assert body == "The body explaining the bug.\nMore body."
assert "LAST_REVIEWED" not in body # drawer dropped
assert "dated log" not in body # sub-entry dropped
def test_parse_handles_a_colon_laden_title():
text = ("* Pearl Open Work\n\n"
"** TODO [#C] Reconsider how =#+title:= displays :bug:quick:\nBody.\n")
t = imp.parse_open_work_tasks(text)[0]
assert t["title"] == "Reconsider how =#+title:= displays"
assert t["tags"] == ["bug", "quick"]
# --- pure: label set + mapping -----------------------------------------------
def test_label_names_needed_drops_the_four_tags():
tasks = imp.parse_open_work_tasks(SAMPLE)
# bug/docs/quick/feature survive (capitalized); discuss is dropped
assert imp.label_names_needed(tasks) == {"Bug", "Docs", "Quick", "Feature"}
def test_task_to_issue_maps_priority_state_and_labels():
label_ids = {"bug": "L-bug", "feature": "L-feat", "docs": "L-docs", "quick": "L-quick"}
states = {"Backlog": "S-back", "Icebox": "S-ice"}
by = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)}
bug = imp.task_to_issue(by["Fix the thing"], states, label_ids)
assert bug["priority"] == 2 # [#B]
assert bug["stateId"] == "S-back"
assert bug["labelIds"] == ["L-bug"] # discuss dropped
assert "explaining the bug" in bug["description"]
someday = imp.task_to_issue(by["Someday feature"], states, label_ids)
assert someday["priority"] == 4 # [#D]
assert someday["stateId"] == "S-ice" # [#D] -> Icebox
# --- boundary: a fake GraphQL transport --------------------------------------
class FakeTransport:
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_import_creates_missing_labels_and_new_issues_only():
created_labels = {"i": 0}
def make_label(_v):
created_labels["i"] += 1
return {"data": {"issueLabelCreate":
{"success": True, "issueLabel": {"id": f"NL{created_labels['i']}"}}}}
fake = FakeTransport({
"Teams": {"data": {"teams": {"nodes": [
{"id": "t1",
"labels": {"nodes": [{"id": "L-bug", "name": "Bug"},
{"id": "L-feat", "name": "Feature"}]},
"states": {"nodes": [{"id": "S-back", "name": "Backlog"},
{"id": "S-ice", "name": "Icebox"}]}}]}}},
"Projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}},
"ProjectIssues": {"data": {"project": {"issues": {"nodes": [{"title": "A docs task"}]}}}},
"LabelCreate": make_label,
"IssueCreate": {"data": {"issueCreate": {"success": True, "issue": {"id": "i1"}}}},
})
summary = imp.import_backlog(imp.LinearClient("key", transport=fake), SAMPLE)
ops = fake.ops()
# Bug + Feature already exist; only Docs + Quick get created (discuss dropped)
assert ops.count("LabelCreate") == 2
# "A docs task" already exists -> skipped; the other two created
assert ops.count("IssueCreate") == 2
assert set(summary["created"]) == {"Fix the thing", "Someday feature"}
assert summary["skipped"] == ["A docs task"]
def test_client_raises_on_graphql_errors():
fake = FakeTransport({"Teams": {"errors": [{"message": "nope"}]}})
with pytest.raises(imp.LinearError, match="nope"):
imp.LinearClient("key", transport=fake).execute("query Teams { teams { nodes { id } } }", {})
|
