|
| 1 | +"""Shared fixtures for the contract-level test suite. |
| 2 | +
|
| 3 | +IMPORTANT - import-time side effects in src/ are load-bearing: utils.py, |
| 4 | +chat_api_handler.py and database_operations.py read the CWD-relative |
| 5 | +config.yaml when first imported, and database_operations.py opens a sqlite |
| 6 | +database at the configured path at import time. The module-level code below |
| 7 | +therefore chdirs into a throwaway sandbox seeded with a minimal config.yaml |
| 8 | +BEFORE any test module (and hence any src module) is imported: every |
| 9 | +import-time side effect lands in the sandbox, the repo's real config.yaml and |
| 10 | +chat_sessions/ are never read or written, and the suite works from any |
| 11 | +invocation directory. |
| 12 | +""" |
| 13 | +import atexit |
| 14 | +import os |
| 15 | +import shutil |
| 16 | +import socket |
| 17 | +import tempfile |
| 18 | +from pathlib import Path |
| 19 | + |
| 20 | +import pytest |
| 21 | +import requests |
| 22 | + |
| 23 | +_SANDBOX = Path(tempfile.mkdtemp(prefix="local-multimodal-chat-tests-")) |
| 24 | +(_SANDBOX / "chat_sessions").mkdir() |
| 25 | +(_SANDBOX / "config.yaml").write_text( |
| 26 | + "\n".join( |
| 27 | + [ |
| 28 | + "ollama:", |
| 29 | + ' embedding_model: "sandbox-embed"', |
| 30 | + " base_url: http://sandbox-ollama:11434", |
| 31 | + "", |
| 32 | + 'whisper_model: "sandbox-whisper"', |
| 33 | + "", |
| 34 | + "chromadb:", |
| 35 | + ' chromadb_path: "chroma_db"', |
| 36 | + ' collection_name: "pdfs"', |
| 37 | + "", |
| 38 | + 'chat_sessions_database_path: "./chat_sessions/chat_sessions.db"', |
| 39 | + "", |
| 40 | + ] |
| 41 | + ) |
| 42 | +) |
| 43 | +# Best-effort cleanup; ignore_errors because the module-level singleton in |
| 44 | +# database_operations holds its sandbox db open for the whole session. |
| 45 | +atexit.register(shutil.rmtree, _SANDBOX, ignore_errors=True) |
| 46 | + |
| 47 | + |
| 48 | +def pytest_configure(config): |
| 49 | + # chdir in the configure hook, not at conftest import: pytest resolves |
| 50 | + # testpaths against the CWD first, and collection (which imports the test |
| 51 | + # modules and therefore the src modules) only starts afterwards - so the |
| 52 | + # import-time side effects still land in the sandbox. |
| 53 | + os.chdir(_SANDBOX) |
| 54 | + |
| 55 | + |
| 56 | +class FakeSessionState(dict): |
| 57 | + """Stand-in for st.session_state covering the four access patterns used in |
| 58 | + src/: st.session_state["key"], .get("key", default), attribute access, and |
| 59 | + membership tests. Never rely on Streamlit's bare-mode session_state - it is |
| 60 | + a process-global singleton with undocumented semantics.""" |
| 61 | + |
| 62 | + def __getattr__(self, name): |
| 63 | + try: |
| 64 | + return self[name] |
| 65 | + except KeyError: |
| 66 | + raise AttributeError(name) from None |
| 67 | + |
| 68 | + def __setattr__(self, name, value): |
| 69 | + self[name] = value |
| 70 | + |
| 71 | + |
| 72 | +class DummyResponse: |
| 73 | + """Minimal requests.Response stand-in; .json() is repeatably callable.""" |
| 74 | + |
| 75 | + def __init__(self, payload): |
| 76 | + self._payload = payload |
| 77 | + |
| 78 | + def json(self): |
| 79 | + return self._payload |
| 80 | + |
| 81 | + |
| 82 | +class RecordingPost: |
| 83 | + """Callable that records every requests.post(...) call and returns a |
| 84 | + canned response. The signature mirrors the real |
| 85 | + requests.post(url, data=None, json=None, ...) so that a refactor in src to |
| 86 | + positional arguments cannot silently record the body under the wrong key.""" |
| 87 | + |
| 88 | + def __init__(self, response_payload): |
| 89 | + self.response_payload = response_payload |
| 90 | + self.calls = [] |
| 91 | + |
| 92 | + def __call__(self, url=None, data=None, json=None, headers=None, **kwargs): |
| 93 | + self.calls.append({"url": url, "data": data, "json": json, "headers": headers}) |
| 94 | + return DummyResponse(self.response_payload) |
| 95 | + |
| 96 | + |
| 97 | +@pytest.fixture(autouse=True) |
| 98 | +def no_network(monkeypatch): |
| 99 | + """Fail loudly if any test hits the network without explicitly patching |
| 100 | + it: requests.get/post are replaced for a clear error message, and raw |
| 101 | + socket connects are blocked as a catch-all (this also covers aiohttp and |
| 102 | + requests.Session). captured_post overrides requests.post for tests that |
| 103 | + expect an HTTP call.""" |
| 104 | + |
| 105 | + def _blocked(*args, **kwargs): |
| 106 | + raise RuntimeError( |
| 107 | + "Network access is disabled in tests; use the captured_post " |
| 108 | + "fixture or patch the network call explicitly." |
| 109 | + ) |
| 110 | + |
| 111 | + monkeypatch.setattr(requests, "get", _blocked) |
| 112 | + monkeypatch.setattr(requests, "post", _blocked) |
| 113 | + monkeypatch.setattr(socket.socket, "connect", _blocked) |
| 114 | + |
| 115 | + |
| 116 | +@pytest.fixture |
| 117 | +def captured_post(monkeypatch): |
| 118 | + """Factory: captured_post(payload) patches requests.post to record calls |
| 119 | + and answer with `payload`. Works because src modules do `import requests` |
| 120 | + and look up requests.post at call time.""" |
| 121 | + |
| 122 | + def _install(response_payload): |
| 123 | + recorder = RecordingPost(response_payload) |
| 124 | + monkeypatch.setattr(requests, "post", recorder) |
| 125 | + return recorder |
| 126 | + |
| 127 | + return _install |
| 128 | + |
| 129 | + |
| 130 | +@pytest.fixture |
| 131 | +def fake_session_state(monkeypatch): |
| 132 | + """Replace st.session_state for every src module at once (they all do |
| 133 | + `import streamlit as st`, so patching the module attribute reaches all).""" |
| 134 | + import streamlit as st |
| 135 | + |
| 136 | + state = FakeSessionState() |
| 137 | + monkeypatch.setattr(st, "session_state", state) |
| 138 | + return state |
| 139 | + |
| 140 | + |
| 141 | +@pytest.fixture |
| 142 | +def db_manager(tmp_path): |
| 143 | + """A DatabaseManager on a throwaway sqlite file. close() in teardown is |
| 144 | + mandatory: an open handle makes tmp_path cleanup fail on Windows.""" |
| 145 | + from database_operations import DatabaseManager |
| 146 | + |
| 147 | + manager = DatabaseManager(str(tmp_path / "test.db")) |
| 148 | + yield manager |
| 149 | + manager.close() |
0 commit comments