conftest.py

tests/conftest.py

1. """Test setup: server.py reads its configuration at import time and exits when
2. BITRIX_WEBHOOK_URL is unset, so the environment is fixed here, before any test
3. module imports it. Data (logs, schema snapshots) goes to a throwaway directory."""
5. import os
6. import sys
7. import tempfile
8. from pathlib import Path
10. import pytest
12. os.environ["BITRIX_WEBHOOK_URL"] = "https://example.invalid/rest/1/x"
13. os.environ["BITRIX_DATA_DIR"] = tempfile.mkdtemp(prefix="bitrix-mcp-tests-")
14. os.environ["BITRIX_LOCAL_TZ"] = "America/Chicago"
15. os.environ["BITRIX_WRITES"] = "on"
16. os.environ["BITRIX_WRITE_PARTIES"] = "on"
17. os.environ["BITRIX_LOG_ARGS"] = "redact"
19. sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
21. import server # noqa: E402
23. # Per-process caches that must not leak between tests. Listed by name and
24. # reset only if present, so the fixture does not depend on how the server
25. # organises them internally.
26. _CACHES = ("_STATUS_CACHE", "_PIPELINES_CACHE", "_FIELDS_CACHE", "_USER_CACHE",
27. "_USER_NAMES", "_ALLOWED_CATS", "_FUNNEL_NAMES")
30. @pytest.fixture(autouse=True)
31. def _fresh_caches():
32. def reset():
33. for name in _CACHES:
34. if not hasattr(server, name):
35. continue
36. value = getattr(server, name)
37. if isinstance(value, dict):
38. value.clear()
39. else:
40. setattr(server, name, None)
41. reset()
42. yield
43. reset()
46. class FakeBitrix:
47. """Stands in for server.bitrix_call: records every (method, params) pair
48. and answers from a script keyed by method name. A scripted value may be a
49. list (consumed in order), an exception instance (raised), or a callable
50. taking params."""
52. def __init__(self, script: dict | None = None):
53. self.script = dict(script or {})
54. self.calls: list[tuple[str, dict]] = []
56. async def __call__(self, method, params=None, envelope=False):
57. self.calls.append((method, params))
58. if method not in self.script:
59. raise AssertionError(f"unexpected Bitrix call: {method} {params!r}")
60. answer = self.script[method]
61. if isinstance(answer, list) and answer and isinstance(answer[0], _Seq):
62. answer = answer.pop(0).value
63. if isinstance(answer, BaseException):
64. raise answer
65. if callable(answer):
66. return answer(params)
67. return answer
69. @property
70. def methods(self) -> list[str]:
71. return [m for m, _ in self.calls]
74. class _Seq:
75. def __init__(self, value):
76. self.value = value
79. def seq(*values):
80. """Script helper: successive answers for repeated calls to one method."""
81. return [_Seq(v) for v in values]
84. @pytest.fixture
85. def fake_bitrix(monkeypatch):
86. def install(script=None):
87. fake = FakeBitrix(script)
88. monkeypatch.setattr(server, "bitrix_call", fake)
89. return fake
90. return install
93. @pytest.fixture
94. def logged(monkeypatch):
95. """Capture log_call records instead of writing calls.jsonl."""
96. records: list[tuple[str, dict, str]] = []
97. monkeypatch.setattr(server, "log_call", lambda tool, args, summary: records.append((tool, args, summary)))
98. return records