test_write_gates.py

tests/test_write_gates.py

1. """Every write tool refuses, without touching Bitrix, when BITRIX_WRITES is off;
2. record ids are validated before anything is read or written."""
4. import asyncio
6. import pytest
8. import server
10. run = asyncio.run
12. WRITE_CALLS = [
13. lambda: server.add_lead_comment("1", "note"),
14. lambda: server.rename_lead("1", "New name"),
15. lambda: server.set_lead_fields("1", {"Rush Job": "yes"}),
16. lambda: server.add_deal_comment("1", "note"),
17. lambda: server.set_deal_title("1", "Booth"),
18. lambda: server.set_deal_stage("1", "New"),
19. lambda: server.set_deal_fields("1", {"Rush Job": "yes"}),
20. lambda: server.assign_deal("1", "me"),
21. lambda: server.link_deal_contact("1", "Ana"),
22. lambda: server.link_deal_company("1", "Harbor"),
23. lambda: server.create_deal("Winback", "Booth"),
24. lambda: server.create_contact("Ana"),
25. lambda: server.update_contact("1", name="Ana"),
26. lambda: server.create_company("Harbor"),
27. lambda: server.update_company("1", title="Harbor"),
28. ]
31. @pytest.mark.parametrize("call", WRITE_CALLS)
32. def test_writes_off_refuses_every_write(call, fake_bitrix, logged, monkeypatch):
33. bx = fake_bitrix({})
34. monkeypatch.setattr(server, "WRITES_ENABLED", False)
35. assert run(call()) == "Writes are disabled on this server (BITRIX_WRITES=off). No change was made."
36. assert bx.calls == []
37. assert logged == []
40. @pytest.mark.parametrize("call", [
41. lambda: server.add_lead_comment("7x", "note"),
42. lambda: server.rename_lead("7x", "New name"),
43. lambda: server.set_lead_fields("7x", {"Rush Job": "yes"}),
44. lambda: server.add_deal_comment("7x", "note"),
45. lambda: server.set_deal_fields("7x", {"Rush Job": "yes"}),
46. lambda: server.get_record("lead", "7x"),
47. lambda: server.get_timeline("lead", "7x"),
48. lambda: server.get_chat_transcript("lead", "7x"),
49. lambda: server.get_products("7x"),
50. lambda: server.get_stage_history("7x"),
51. ])
52. def test_invalid_record_id_is_refused_first(call, fake_bitrix, logged):
53. bx = fake_bitrix({})
54. assert run(call()) == "Invalid record_id '7x': a CRM record ID is a positive integer."
55. assert bx.calls == []
58. def test_add_lead_comment_payload(fake_bitrix, logged):
59. bx = fake_bitrix({"crm.timeline.comment.add": 555})
60. assert run(server.add_lead_comment(" 250 ", " Called back \x07")) == "Posted comment 555 to lead 250 (11 chars)."
61. assert bx.calls == [("crm.timeline.comment.add",
62. {"fields": {"ENTITY_ID": "250", "ENTITY_TYPE": "LEAD", "COMMENT": "Called back"}})]
63. assert logged == [("add_lead_comment", {"record_id": "250", "len": 11}, "posted comment 555")]
66. def test_user_lookups_share_one_cache(fake_bitrix):
67. bx = fake_bitrix({"user.get": [{"ID": "7", "NAME": "Sam", "LAST_NAME": "Lee"}]})
68. assert run(server._user_display(7)) == "Sam Lee"
69. assert run(server._resolve_user("7")) == ("7", "Sam Lee")
70. assert run(server._user_display("7")) == "Sam Lee"
71. assert bx.methods == ["user.get"]
72. assert run(server._user_display(0)) == ""
75. def test_user_display_falls_back_without_caching_the_error(fake_bitrix):
76. fake_bitrix({"user.get": server.BitrixError("ACCESS_DENIED: no")})
77. assert run(server._user_display("8")) == "user 8"
78. with pytest.raises(server.BitrixError):
79. run(server._resolve_user("8"))