python153 lines · 4.6 KB

1import json

2

3from mapping import load_mapping

4from notion import NotionClient, NotionError, apply_writes, clip, load_work_from_notion, work_from_page

5

6SAMPLE = """---

7workspace: growth-chef

8platform: slack

9scope: team.growth

10agent: relay-prime

11---

12| id | name | role | direction |

13|----|------|------|-----------|

14| C0MGMT01 | #mgmt | management | in |

15| C0REPT02 | #reports | reports | out |

16| C0OPS03 | #ops | room | both |

17"""

18

19PAGE = {

20 "id": "page-north",

21 "properties": {

22 "Name": {"type": "title", "title": [{"plain_text": "North Star launch"}]},

23 "Scope key": {"type": "rich_text", "rich_text": [{"plain_text": "north-star"}]},

24 "Admin Channel": {"type": "rich_text", "rich_text": [{"plain_text": "C0OPS03"}]},

25 "Status": {"type": "status", "status": {"name": "Active"}},

26 "Broadcast TS": {"type": "rich_text", "rich_text": []},

27 "Owner": {"type": "rich_text", "rich_text": [{"plain_text": "U08AAAA"}]},

28 },

29}

30

31

32def test_work_from_page():

33 w = work_from_page(PAGE)

34 assert w is not None

35 assert w.id == "north-star"

36 assert w.room == "C0OPS03"

37 assert w.state == "active"

38 assert w.anchor == ""

39 assert w.page_id == "page-north"

40

41

42def test_company_status_is_not_treated_as_active():

43 page = {

44 "id": PAGE["id"],

45 "properties": dict(PAGE["properties"]),

46 }

47 page["properties"]["Status"] = {"type": "status", "status": {"name": "Review"}}

48

49 def transport(method, url, headers, body):

50 return {"results": [page], "has_more": False}

51

52 client = NotionClient("secret_test", transport)

53 mapping = load_work_from_notion(load_mapping(SAMPLE), "db1", client)

54 work, action = mapping.plan()[0]

55 assert work.state == "review"

56 assert action == "gap"

57

58

59def test_query_overlays_mapping():

60 calls = []

61

62 def transport(method, url, headers, body):

63 calls.append((method, url))

64 return {"results": [PAGE], "has_more": False}

65

66 client = NotionClient("secret_test", transport)

67 mapping = load_work_from_notion(load_mapping(SAMPLE), "db1", client)

68 assert len(mapping.work) == 1

69 assert mapping.plan()[0][1] == "post"

70 assert calls[0][0] == "POST"

71 assert "/databases/db1/query" in calls[0][1]

72

73

74def test_dry_run_does_not_patch():

75 calls = []

76

77 def transport(method, url, headers, body):

78 calls.append(method)

79 return {"results": [PAGE], "has_more": False}

80

81 client = NotionClient("secret_test", transport)

82 mapping = load_work_from_notion(load_mapping(SAMPLE), "db1", client)

83 calls.clear()

84 lines = apply_writes(client, mapping.plan(), {}, {}, dry_run=True)

85 assert any("POST" in line for line in lines)

86 assert calls == []

87

88

89def test_live_write_patches_anchor():

90 calls = []

91

92 def transport(method, url, headers, body):

93 calls.append((method, url, json.loads(body.decode()) if body else None))

94 return {}

95

96 client = NotionClient("secret_test", transport)

97 w = work_from_page(PAGE)

98 apply_writes(

99 client,

100 ((w, "post"),),

101 {"north-star": ("1000000000.000002", "https://example.slack.com/archives/C0OPS03/p1000000000002")},

102 {},

103 dry_run=False,

104 )

105 assert calls[0][0] == "PATCH"

106 assert "/pages/page-north" in calls[0][1]

107 ts = calls[0][2]["properties"]["Broadcast TS"]["rich_text"][0]["text"]["content"]

108 assert ts == "1000000000.000002"

109 link = calls[0][2]["properties"]["Broadcast Permalink"]["rich_text"][0]["text"]["content"]

110 assert link.startswith("https://example.slack.com/")

111

112

113def test_post_without_ts_does_not_patch():

114 calls = []

115

116 def transport(method, url, headers, body):

117 calls.append(method)

118 return {}

119

120 client = NotionClient("secret_test", transport)

121 w = work_from_page(PAGE)

122 lines = apply_writes(client, ((w, "post"),), {}, {}, dry_run=False)

123 assert calls == []

124 assert any("no ts" in line for line in lines)

125

126

127def test_empty_ts_is_refused():

128 client = NotionClient("secret_test", lambda *a: {})

129 w = work_from_page(PAGE)

130 try:

131 client.write_anchor(w.page_id, "", "https://example.slack.com/p1")

132 raise AssertionError("wrote empty Broadcast TS")

133 except NotionError:

134 pass

135

136

137def test_clip_marks_truncation():

138 out = clip("x" * 3000)

139 assert len(out) == 2000

140 assert out.endswith("(truncated)")

141

142

143if __name__ == "__main__":

144 test_work_from_page()

145 test_company_status_is_not_treated_as_active()

146 test_query_overlays_mapping()

147 test_dry_run_does_not_patch()

148 test_live_write_patches_anchor()

149 test_post_without_ts_does_not_patch()

150 test_empty_ts_is_refused()

151 test_clip_marks_truncation()

152 print("ok")

153