test_party_writes.py

tests/test_party_writes.py

1. """Pins the payloads and replies of create_contact, update_contact,
2. create_company and update_company, which share one internal implementation."""
4. import asyncio
6. import pytest
7. from conftest import seq
9. import server
11. run = asyncio.run
13. NO_DUP: dict = {}
14. SAM = [{"ID": "7", "NAME": "Sam", "LAST_NAME": "Lee"}]
16. CONTACT_FIELDS = {
17. "fields": {
18. "NAME": {"type": "string", "title": "First name"},
19. "UF_CRM_MARKET": {
20. "type": "enumeration",
21. "title": "Market",
22. "items": [{"ID": "44", "VALUE": "Government"}, {"ID": "45", "VALUE": "Retail"}],
23. },
24. }
25. }
26. COMPANY_FIELDS = {
27. "fields": {
28. "TITLE": {"type": "string", "title": "Company name"},
29. "UF_CRM_TIER": {
30. "type": "enumeration",
31. "title": "Partner tier",
32. "items": [{"ID": "61", "VALUE": "Palladium"}, {"ID": "62", "VALUE": "Iridium"}],
33. },
34. }
35. }
38. def fm(value):
39. return [{"VALUE": value, "VALUE_TYPE": "WORK"}]
42. # --------------------------------------------------------------------------- create_contact
44. def test_create_contact_full_payload(fake_bitrix, logged):
45. bx = fake_bitrix({
46. "crm.duplicate.findbycomm": NO_DUP,
47. "user.get": SAM,
48. "crm.contact.add": 501,
49. })
50. out = run(server.create_contact(" Ana ", last_name="Reyes ", email=" ana@buyer.example ",
51. phone="+1 555 010 2000", assignee="7"))
52. assert out == "Created contact 501: Ana Reyes."
53. assert bx.calls == [
54. ("crm.duplicate.findbycomm", {"entity_type": "CONTACT", "type": "EMAIL", "values": ["ana@buyer.example"]}),
55. ("crm.duplicate.findbycomm", {"entity_type": "CONTACT", "type": "PHONE", "values": ["+1 555 010 2000"]}),
56. ("user.get", {"ID": "7"}),
57. ("crm.contact.add", {"fields": {
58. "NAME": "Ana",
59. "LAST_NAME": "Reyes",
60. "EMAIL": fm("ana@buyer.example"),
61. "PHONE": fm("+1 555 010 2000"),
62. "ASSIGNED_BY_ID": "7",
63. }}),
64. ]
65. assert list(bx.calls[-1][1]["fields"]) == ["NAME", "LAST_NAME", "EMAIL", "PHONE", "ASSIGNED_BY_ID"]
66. assert logged == [("create_contact", {"name": "Ana", "id": 501}, "created")]
69. def test_create_contact_name_only(fake_bitrix, logged):
70. bx = fake_bitrix({"crm.contact.add": {"id": 88}})
71. out = run(server.create_contact("Ana"))
72. assert out == "Created contact 88: Ana."
73. assert bx.calls == [("crm.contact.add", {"fields": {"NAME": "Ana"}})]
74. assert logged == [("create_contact", {"name": "Ana", "id": 88}, "created")]
77. def test_create_contact_passes_through_unexpected_add_result(fake_bitrix, logged):
78. fake_bitrix({"crm.contact.add": "90"})
79. assert run(server.create_contact("Ana")) == "Created contact 90: Ana."
82. def test_create_contact_returns_existing_duplicate(fake_bitrix, logged):
83. bx = fake_bitrix({"crm.duplicate.findbycomm": seq(NO_DUP, {"CONTACT": [42, 43]})})
84. out = run(server.create_contact("Ana", email="ana@buyer.example", phone="5550102000"))
85. assert out == "Contact already exists (id 42, phone match); not creating a duplicate. Link it instead."
86. assert bx.methods == ["crm.duplicate.findbycomm", "crm.duplicate.findbycomm"]
87. assert logged == []
90. def test_create_contact_dup_check_error_is_skipped(fake_bitrix, logged):
91. bx = fake_bitrix({
92. "crm.duplicate.findbycomm": server.BitrixError("ACCESS_DENIED: no"),
93. "crm.contact.add": 3,
94. })
95. assert run(server.create_contact("Ana", email="ana@buyer.example")) == "Created contact 3: Ana."
96. assert bx.methods == ["crm.duplicate.findbycomm", "crm.contact.add"]
99. def test_create_contact_requires_name(fake_bitrix, logged):
100. bx = fake_bitrix({})
101. assert run(server.create_contact(" \x00 ")) == "Refused: a contact name is required."
102. assert bx.calls == []
105. def test_create_contact_unknown_assignee(fake_bitrix, logged):
106. bx = fake_bitrix({"user.search": []})
107. out = run(server.create_contact("Ana", assignee="Nobody"))
108. assert out == "No Bitrix user found matching 'Nobody'."
109. assert bx.calls == [("user.search", {"FILTER": {"FIND": "Nobody"}})]
110. assert logged == []
113. def test_create_contact_add_error(fake_bitrix, logged):
114. fake_bitrix({"crm.contact.add": server.BitrixError("INVALID: bad")})
115. assert run(server.create_contact("Ana")) == "Error creating contact: INVALID: bad"
116. assert logged == [("create_contact", {"name": "Ana"}, "error: INVALID: bad")]
119. # --------------------------------------------------------------------------- create_company
121. def test_create_company_full_payload(fake_bitrix, logged):
122. bx = fake_bitrix({
123. "crm.duplicate.findbycomm": NO_DUP,
124. "user.current": {"ID": "9", "NAME": "Me"},
125. "crm.company.add": 700,
126. })
127. out = run(server.create_company(" Harbor Signs ", email="hello@harbor.example",
128. phone="555-010-3000", assignee="me"))
129. assert out == "Created company 700: Harbor Signs."
130. assert bx.calls == [
131. ("crm.duplicate.findbycomm", {"entity_type": "COMPANY", "type": "EMAIL", "values": ["hello@harbor.example"]}),
132. ("crm.duplicate.findbycomm", {"entity_type": "COMPANY", "type": "PHONE", "values": ["555-010-3000"]}),
133. ("user.current", None),
134. ("crm.company.add", {"fields": {
135. "TITLE": "Harbor Signs",
136. "EMAIL": fm("hello@harbor.example"),
137. "PHONE": fm("555-010-3000"),
138. "ASSIGNED_BY_ID": "9",
139. }}),
140. ]
141. assert list(bx.calls[-1][1]["fields"]) == ["TITLE", "EMAIL", "PHONE", "ASSIGNED_BY_ID"]
142. assert logged == [("create_company", {"title": "Harbor Signs", "id": 700}, "created")]
145. def test_create_company_returns_existing_duplicate(fake_bitrix, logged):
146. fake_bitrix({"crm.duplicate.findbycomm": {"COMPANY": ["12"]}})
147. out = run(server.create_company("Harbor Signs", email="hello@harbor.example"))
148. assert out == "Company already exists (id 12, email match); not creating a duplicate. Link it instead."
149. assert logged == []
152. def test_create_company_requires_title(fake_bitrix, logged):
153. bx = fake_bitrix({})
154. assert run(server.create_company(" ")) == "Refused: a company name is required."
155. assert bx.calls == []
158. def test_create_company_numeric_assignee_is_taken_as_id(fake_bitrix, logged):
159. # A numeric assignee is used as the id even when user.get finds nobody.
160. bx = fake_bitrix({"user.get": [], "crm.company.add": 5})
161. assert run(server.create_company("Harbor", assignee="31")) == "Created company 5: Harbor."
162. assert bx.calls[-1] == ("crm.company.add", {"fields": {"TITLE": "Harbor", "ASSIGNED_BY_ID": "31"}})
165. def test_create_company_add_error(fake_bitrix, logged):
166. fake_bitrix({"crm.company.add": server.BitrixError("INVALID: bad")})
167. assert run(server.create_company("Harbor")) == "Error creating company: INVALID: bad"
168. assert logged == [("create_company", {"title": "Harbor"}, "error: INVALID: bad")]
171. # --------------------------------------------------------------------------- update_contact
173. def test_update_contact_full_payload(fake_bitrix, logged):
174. bx = fake_bitrix({
175. "user.get": SAM,
176. "crm.item.fields": CONTACT_FIELDS,
177. "crm.contact.update": True,
178. })
179. out = run(server.update_contact(" 12 ", name="Ana", last_name="Reyes", email="ana@buyer.example",
180. phone="5550102000", assignee="7", fields={"Market": "Government"}))
181. assert out == "Updated contact 12: NAME, LAST_NAME, EMAIL, PHONE, ASSIGNED_BY_ID, Market=Government."
182. assert bx.calls == [
183. ("user.get", {"ID": "7"}),
184. ("crm.item.fields", {"entityTypeId": 3, "useOriginalUfNames": "Y"}),
185. ("crm.contact.update", {"id": "12", "fields": {
186. "NAME": "Ana",
187. "LAST_NAME": "Reyes",
188. "EMAIL": fm("ana@buyer.example"),
189. "PHONE": fm("5550102000"),
190. "ASSIGNED_BY_ID": "7",
191. "UF_CRM_MARKET": "44",
192. }}),
193. ]
194. assert list(bx.calls[-1][1]["fields"]) == [
195. "NAME", "LAST_NAME", "EMAIL", "PHONE", "ASSIGNED_BY_ID", "UF_CRM_MARKET"]
196. fields_logged = ["NAME", "LAST_NAME", "EMAIL", "PHONE", "ASSIGNED_BY_ID", "UF_CRM_MARKET"]
197. assert logged == [("update_contact", {"record_id": "12", "fields": fields_logged}, "updated")]
200. def test_update_contact_single_field(fake_bitrix, logged):
201. bx = fake_bitrix({"crm.contact.update": True})
202. assert run(server.update_contact("12", phone="5550102000")) == "Updated contact 12: PHONE."
203. assert bx.calls == [("crm.contact.update", {"id": "12", "fields": {"PHONE": fm("5550102000")}})]
206. def test_update_contact_rejects_standard_field_in_fields(fake_bitrix, logged):
207. bx = fake_bitrix({"crm.item.fields": CONTACT_FIELDS})
208. out = run(server.update_contact("12", fields={"First name": "Ana"}))
209. assert out == ("Refused: NAME are standard fields; use the name/email/phone args "
210. "instead of `fields`.")
211. assert bx.methods == ["crm.item.fields"]
214. def test_update_contact_unknown_enum_option(fake_bitrix, logged):
215. fake_bitrix({"crm.item.fields": CONTACT_FIELDS})
216. out = run(server.update_contact("12", fields={"Market": "Space"}))
217. assert out == "'Space' is not an option of UF_CRM_MARKET. Options: Government | Retail"
220. def test_update_contact_nothing_to_update(fake_bitrix, logged):
221. bx = fake_bitrix({})
222. out = run(server.update_contact("12", name=" "))
223. assert out == "Nothing to update: pass a name/email/phone/assignee or fields."
224. assert bx.calls == []
227. def test_update_contact_invalid_id(fake_bitrix, logged):
228. bx = fake_bitrix({})
229. out = run(server.update_contact("12a", name="Ana"))
230. assert out == "Invalid record_id '12a': a CRM record ID is a positive integer."
231. assert bx.calls == []
234. def test_update_contact_unknown_assignee(fake_bitrix, logged):
235. bx = fake_bitrix({"user.search": []})
236. out = run(server.update_contact("12", name="Ana", assignee="Nobody"))
237. assert out == "No Bitrix user found matching 'Nobody'."
238. assert bx.methods == ["user.search"]
241. def test_update_contact_update_error(fake_bitrix, logged):
242. fake_bitrix({"crm.contact.update": server.BitrixError("NOT_FOUND: gone")})
243. out = run(server.update_contact("12", name="Ana"))
244. assert out == "Error updating contact 12: NOT_FOUND: gone"
245. assert logged == [("update_contact", {"record_id": "12", "fields": ["NAME"]}, "error: NOT_FOUND: gone")]
248. # --------------------------------------------------------------------------- update_company
250. def test_update_company_full_payload(fake_bitrix, logged):
251. bx = fake_bitrix({
252. "user.search": [{"ID": "7", "NAME": "Sam", "LAST_NAME": "Lee"}],
253. "crm.item.fields": COMPANY_FIELDS,
254. "crm.company.update": True,
255. })
256. out = run(server.update_company("40", title=" Harbor Signs ", email="hello@harbor.example",
257. phone="555-010-3000", assignee="Sam", fields={"Partner tier": "Iridium"}))
258. assert out == "Updated company 40: TITLE, EMAIL, PHONE, ASSIGNED_BY_ID, Partner tier=Iridium."
259. assert bx.calls == [
260. ("user.search", {"FILTER": {"FIND": "Sam"}}),
261. ("crm.item.fields", {"entityTypeId": 4, "useOriginalUfNames": "Y"}),
262. ("crm.company.update", {"id": "40", "fields": {
263. "TITLE": "Harbor Signs",
264. "EMAIL": fm("hello@harbor.example"),
265. "PHONE": fm("555-010-3000"),
266. "ASSIGNED_BY_ID": "7",
267. "UF_CRM_TIER": "62",
268. }}),
269. ]
270. fields_logged = ["TITLE", "EMAIL", "PHONE", "ASSIGNED_BY_ID", "UF_CRM_TIER"]
271. assert logged == [("update_company", {"record_id": "40", "fields": fields_logged}, "updated")]
274. def test_update_company_rejects_standard_field_in_fields(fake_bitrix, logged):
275. fake_bitrix({"crm.item.fields": COMPANY_FIELDS})
276. out = run(server.update_company("40", fields={"Company name": "X"}))
277. assert out == ("Refused: TITLE are standard fields; use the title/email/phone args "
278. "instead of `fields`.")
281. def test_update_company_nothing_to_update(fake_bitrix, logged):
282. out = run(server.update_company("40"))
283. assert out == "Nothing to update: pass a title/email/phone/assignee or fields."
286. def test_update_company_invalid_id(fake_bitrix, logged):
287. bx = fake_bitrix({})
288. out = run(server.update_company("-4", title="Harbor"))
289. assert out == "Invalid record_id '-4': a CRM record ID is a positive integer."
290. assert bx.calls == []
293. def test_update_company_update_error(fake_bitrix, logged):
294. fake_bitrix({"crm.company.update": server.BitrixError("NOT_FOUND: gone")})
295. out = run(server.update_company("40", title="Harbor"))
296. assert out == "Error updating company 40: NOT_FOUND: gone"
297. assert logged == [("update_company", {"record_id": "40", "fields": ["TITLE"]}, "error: NOT_FOUND: gone")]
300. # --------------------------------------------------------------------------- gates
302. @pytest.mark.parametrize("call", [
303. lambda: server.create_contact("Ana"),
304. lambda: server.update_contact("12", name="Ana"),
305. lambda: server.create_company("Harbor"),
306. lambda: server.update_company("40", title="Harbor"),
307. ])
308. def test_party_writes_respect_both_switches(call, fake_bitrix, logged, monkeypatch):
309. bx = fake_bitrix({})
310. monkeypatch.setattr(server, "WRITES_ENABLED", False)
311. assert run(call()) == "Writes are disabled on this server (BITRIX_WRITES=off). No change was made."
312. monkeypatch.setattr(server, "WRITES_ENABLED", True)
313. monkeypatch.setattr(server, "PARTY_WRITES_ENABLED", False)
314. assert run(call()) == ("Contact/company writes are disabled (BITRIX_WRITE_PARTIES=off). "
315. "No change was made.")
316. assert bx.calls == []