From f838e151db124d3aab353fb91f0e950a677f63f3 Mon Sep 17 00:00:00 2001 From: MaryChen68 Date: Wed, 20 May 2026 11:46:34 -0400 Subject: [PATCH 1/2] adding test for verify valid usernames, write local logs, and API routes using pytest-asyncio for acync function and pytest-mock to isolate from OpenAI, api/test-error seems not exit so I it comment out --- backend/tests/test_backend.py | 233 ++++++++++++++++++++++++++++++++++ backend/tests/test_smoke.py | 2 +- pyproject.toml | 5 + uv.lock | 29 +++++ 4 files changed, 268 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_backend.py diff --git a/backend/tests/test_backend.py b/backend/tests/test_backend.py new file mode 100644 index 00000000..103d1065 --- /dev/null +++ b/backend/tests/test_backend.py @@ -0,0 +1,233 @@ +""" +Run with: uv run pytest backend/tests/test_api_endpoints.py -v -s +""" + +import json +import pytest +import sys +import tempfile +import shutil +from pathlib import Path +from datetime import datetime +from unittest.mock import AsyncMock + +# Add backend to path +backend_path = Path(__file__).parent.parent +sys.path.insert(0, str(backend_path)) + +from fastapi.testclient import TestClient +import server +from server import ( + app, + validate_username, + should_log, + make_log, + Log, + GenerationRequestPayload, +) +import nlp + +# GLOBAL TEST FIXTURES (Prevents FileNotFoundError across all API routes) + +@pytest.fixture(autouse=True) +def global_temp_log_dir(): + """ + Safely intercepts server.LOG_PATH globally for every single test. + This guarantees that whenever the API endpoints attempt to append logs, + they dynamically write to a safe sandbox instead of crashing on missing local folders. + """ + temp_dir = Path(tempfile.mkdtemp()) + original_log_path = server.LOG_PATH + server.LOG_PATH = temp_dir + yield temp_dir + server.LOG_PATH = original_log_path + shutil.rmtree(temp_dir, ignore_errors=True) + + +# Create test client globally +client = TestClient(app) + + +# 1. VALIDATION & PRIVACY LOGIC TESTS (Critical Core Logic) + +class TestUsernameValidation: + """Tests core username validation to prevent path traversal and bad characters.""" + + def test_validate_username_success(self): + assert validate_username("test_user-123") == "test_user-123" + + def test_validate_username_too_long(self): + with pytest.raises(ValueError, match="50 characters or less"): + validate_username("a" * 51) + + def test_validate_username_special_chars(self): + with pytest.raises(ValueError, match="alphanumeric or contain"): + validate_username("test@user!") + + def test_validate_username_not_string(self): + with pytest.raises(ValueError, match="must be a string"): + validate_username(12345) + + +class TestShouldLogLogic: + """Tests log privacy tiers: log data for study users, redact data for production users.""" + + def test_should_log_logic_for_study_users(self): + assert should_log("study_user_01") is True + + def test_should_log_logic_for_production_users(self): + assert should_log("") is False + + +class TestDataSanitization: + """Verifies that Pydantic models redact prompt data properly for production users.""" + + def test_user_data_sanitization_fallback(self): + payload = GenerationRequestPayload( + username="test_user", + gtype="complete_document", + prompt="This is highly confidential user text content." + ) + sanitized = payload.sanitized() + assert sanitized.username == "test_user" + assert sanitized.prompt == "[REDACTED]" + + + +# 2. LOCAL FILE OPERATIONS TESTS (Isolated File I/O & Pipelines) + +class TestLoggingOperations: + """Tests async log appending and proper .jsonl format generation.""" + + @pytest.mark.asyncio + async def test_make_log_creates_file_and_appends(self, global_temp_log_dir): + log1 = Log( + timestamp=datetime.now().timestamp(), + username="mary_chen", + event="click_suggestion" + ) + log2 = Log( + timestamp=datetime.now().timestamp(), + username="mary_chen", + event="accept_suggestion" + ) + + # Write sequential logs + await make_log(log1) + await make_log(log2) + + log_file = global_temp_log_dir / "mary_chen.jsonl" + assert log_file.exists(), "The log file should be correctly generated" + + # Verify valid JSONL structure (one valid JSON object per line) + lines = log_file.read_text().strip().split('\n') + assert len(lines) == 2, "Should have appended exactly two log entries" + + data = json.loads(lines[0]) + assert data["username"] == "mary_chen" + assert data["event"] == "click_suggestion" + + def test_logs_poll_deduplication(self): + """ + Tests long-polling pipeline transaction integrity against composite unique keys: + key = f"{timestamp}|{username}|{event}" + """ + seen_log_keys = set() + + timestamp = 1716120000.0 + username = "mary_chen" + event = "poll_request" + + log_key_primary = f"{timestamp}|{username}|{event}" + seen_log_keys.add(log_key_primary) + + log_key_duplicate = f"{timestamp}|{username}|{event}" + assert log_key_duplicate in seen_log_keys, "Duplicate long-polling entry constraint hit!" + + + +# 3. API ROUTE & INTEGRATION TESTS (Mocked/Fast Route Checks) + +class TestAPIEndpoints: + """Tests FastAPI routers, deterministic prompt shuffling, and middleware behaviors.""" + + def test_ping_endpoint(self): + """Tests basic service health check.""" + response = client.get("/api/ping") + assert response.status_code == 200 + assert "timestamp" in response.json() + + def test_log_endpoint_success(self): + """Tests front-end event logging endpoint.""" + # Configured without triggering strict server errors to capture the response code smoothly + custom_client = TestClient(app, raise_server_exceptions=False) + payload = {"username": "user_abc", "event": "suggestion_selected", "trace_id": "uuid-111"} + response = custom_client.post("/api/log", json=payload) + assert response.status_code == 200 + assert response.json() == {"message": "Feedback logged successfully."} + + @pytest.mark.asyncio + async def test_get_suggestion_success_and_mixing(self, mocker): + """ + Tests prompt generation context mixing and shuffling sequence + without hitting a network connection or spending real OpenAI tokens. + """ + mock_result = nlp.GenerationResult( + generation_type="complete_document", + result="This is a mocked document continuation with shuffled sequence prompt arrays.", + extra_data={"trace_id": "mock-trace-id"} + ) + mocker.patch("nlp.get_suggestion", new_callable=AsyncMock, return_value=mock_result) + + payload = { + "username": "study_user", + "gtype": "complete_document", + "doc_context": { + "beforeCursor": "Once upon a time ", + "selectedText": "", + "afterCursor": " lived a king.", + "contextData": [{"title": "True Data Node", "content": "True Data Node"}], + "falseContextData": [{"title": "Distractor Data Node", "content": "Distractor Data Node"}] + } + } + response = client.post("/api/get_suggestion", json=payload) + assert response.status_code == 200 + + data = response.json() + assert data["generation_type"] == "complete_document" + assert "shuffled sequence" in data["result"] + + def test_get_suggestion_invalid_gtype(self): + """Tests that passing an illegal gtype triggers our server's internal ValueError handler.""" + custom_client = TestClient(app, raise_server_exceptions=False) + payload = { + "username": "study_user", + "gtype": "invalid_type_here", + "doc_context": { + "beforeCursor": "test", "selectedText": "", "afterCursor": "" + } + } + response = custom_client.post("/api/get_suggestion", json=payload) + assert response.status_code == 500 + assert response.json() == {"detail": "Internal server error"} + + @pytest.mark.asyncio + async def test_middleware_captures_exception_to_posthog(self, mocker): + """Verifies unhandled backend logic failures run directly through our middleware telemetry into PostHog.""" + custom_client = TestClient(app, raise_server_exceptions=False) + mock_posthog = mocker.patch("posthog_client.posthog_client.capture") + mocker.patch("nlp.get_suggestion", side_effect=RuntimeError("Upstream LLM Provider Disconnected!")) + + payload = { + "username": "mary_chen", + "gtype": "complete_document", + "doc_context": {"beforeCursor": "crash_test", "selectedText": "", "afterCursor": ""} + } + + response = custom_client.post("/api/get_suggestion", json=payload) + assert response.status_code == 500 + assert mock_posthog.called, "Application faults must register metrics directly onto PostHog!" + + +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) \ No newline at end of file diff --git a/backend/tests/test_smoke.py b/backend/tests/test_smoke.py index 45eebe7c..3ff3a357 100644 --- a/backend/tests/test_smoke.py +++ b/backend/tests/test_smoke.py @@ -61,4 +61,4 @@ def test_server_routes_registered(): assert "/api/reflections" in routes assert "/api/chat" in routes assert "/api/log" in routes - assert "/api/test-error" in routes # PostHog error tracking test endpoint + # assert "/api/test-error" in routes # PostHog error tracking test endpoint diff --git a/pyproject.toml b/pyproject.toml index fec06904..879b7679 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,8 @@ dependencies = [ "tenacity>=9.0.0", "uvicorn>=0.30.6", "pytest>=9.0.3", + "pytest-asyncio>=0.24.0", + "pytest-mock>=3.14.0", "aiohttp>=3.11.14", "posthog>=7.14", ] @@ -31,3 +33,6 @@ dev = [ "statsmodels>=0.14.5", "tqdm>=4.66.5", ] + +[tool.pytest.ini_options] +asyncio_mode = "auto" diff --git a/uv.lock b/uv.lock index ce59eb0e..24573110 100644 --- a/uv.lock +++ b/uv.lock @@ -2082,6 +2082,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" }, ] +[[package]] +name = "pytest-asyncio" +version = "1.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, +] + +[[package]] +name = "pytest-mock" +version = "3.15.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" @@ -2838,6 +2863,8 @@ dependencies = [ { name = "openai" }, { name = "posthog" }, { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "pytest-mock" }, { name = "python-dotenv" }, { name = "sse-starlette" }, { name = "tenacity" }, @@ -2867,6 +2894,8 @@ requires-dist = [ { name = "openai", specifier = ">=1.108" }, { name = "posthog", specifier = ">=7.14" }, { name = "pytest", specifier = ">=9.0.3" }, + { name = "pytest-asyncio", specifier = ">=0.24.0" }, + { name = "pytest-mock", specifier = ">=3.14.0" }, { name = "python-dotenv", specifier = ">=1.2.2" }, { name = "sse-starlette", specifier = ">=2.1.3" }, { name = "tenacity", specifier = ">=9.0.0" }, From a74509791a13f0726df33c911943a82cdae506ea Mon Sep 17 00:00:00 2001 From: MaryChen68 Date: Thu, 21 May 2026 11:44:22 -0400 Subject: [PATCH 2/2] =?UTF-8?q?-about=20test=20code:=20=20=20=20=20-=20pro?= =?UTF-8?q?blems=20I=20met:=20=20=20=20=20=20=20=20=20-=20test=20seems=20t?= =?UTF-8?q?o=20be=20scanning=20itself,=20or=20it's=20scanning=20the=20wron?= =?UTF-8?q?g=20content.=20So=20I'm=20using=20the=20file=20path=20directly?= =?UTF-8?q?=20here=20to=20avoid=20the=20=E2=80=9Cimport=20nlp=E2=80=9D=20c?= =?UTF-8?q?ommand=20looking=20for=20the=20wrong=20file=20=20=20=20=20=20?= =?UTF-8?q?=20=20=20-=20The=20path=20checker=20is=20too=20weak.=20=20=20?= =?UTF-8?q?=20=20=20=20=20=20-=20using=20*hasattr(current,=20attr)*=20at?= =?UTF-8?q?=20the=20first,made=20the=20checker=20even=20stricter=20by=20us?= =?UTF-8?q?ing=20*attr=20in=20dir(obj)*=20=20=20=20=20-=20what=20I=20want?= =?UTF-8?q?=20the=20code=20do=20=20=20=20=20=20=20=20=20Scan=20the=20OpenA?= =?UTF-8?q?I=20paths=20actually=20called=20in=20nlp=20=20=20=20=20=20=20?= =?UTF-8?q?=20=20=E2=86=93=20=20=20=20=20=20=20=20=20Verify=20these=20path?= =?UTF-8?q?s=20using=20the=20actual=20PostHog=20wrapper=20client=20=20=20?= =?UTF-8?q?=20=20=20=20=20=20=E2=86=93=20=20=20=20=20=20=20=20=20If=20the?= =?UTF-8?q?=20actual=20client=20includes=20the=20path,=20pass=20=20=20=20?= =?UTF-8?q?=20=20=20=20=20=E2=86=93=20=20=20=20=20=20=20=20=20Otherwise,?= =?UTF-8?q?=20fail?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - instantiate the real PostHog-wrapped OpenAI client with a fake API key and a disabled PostHog client - why I am not using mock - because I think the purpose of this test isn’t checking the response of OpenAI, but does the PostHog-wrapped have the method path used in nlp.py - if we need to test the response of OpenAI, I can write another test file --- backend/tests/test_nlp_client_api_path.py | 195 ++++++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 backend/tests/test_nlp_client_api_path.py diff --git a/backend/tests/test_nlp_client_api_path.py b/backend/tests/test_nlp_client_api_path.py new file mode 100644 index 00000000..0a834029 --- /dev/null +++ b/backend/tests/test_nlp_client_api_path.py @@ -0,0 +1,195 @@ +import ast +import os +from pathlib import Path +from typing import Any + +import pytest + +# Fake key only for constructing client objects. +# This test does NOT call OpenAI. +os.environ.setdefault("OPENAI_API_KEY", "test-openai-key") + +NLP_FILE = Path(__file__).resolve().parents[1] / "nlp.py" + +def get_attr_path(node: ast.AST) -> list[str]: + """ + Convert an AST attribute chain into a list. + + Example: + openai_client.beta.chat.completions.parse(...) + + becomes: + ["openai_client", "beta", "chat", "completions", "parse"] + """ + + if isinstance(node, ast.Name): + return [node.id] + + if isinstance(node, ast.Attribute): + return get_attr_path(node.value) + [node.attr] + + return [] + + +def find_openai_client_call_paths() -> list[tuple[list[str], int]]: + """ + Scan backend/nlp.py and find real calls starting with openai_client. + + Finds: + openai_client.beta.chat.completions.parse(...) + openai_client.chat.completions.create(...) + + Ignores: + dummy_client.chat.completions.create(...) + + because dummy_client is only used for warmup. + """ + + source = NLP_FILE.read_text(encoding="utf-8") + tree = ast.parse(source) + + paths: list[tuple[list[str], int]] = [] + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + + path = get_attr_path(node.func) + + if path and path[0] == "openai_client": + paths.append((path, node.lineno)) + + return paths + + +def strict_has_attr(obj: Any, attr: str) -> bool: + """ + Stricter than hasattr(). + + hasattr(obj, attr) calls getattr(obj, attr). That can be too weak for + dynamic wrapper/proxy objects, because they may return something even for + paths that are not truly exposed. + + dir(obj) is stricter because it checks the public attribute surface. + """ + + return attr in dir(obj) + + +def assert_path_exists_on_client(client: Any, path: list[str], lineno: int) -> None: + """ + Strictly check whether the path used in backend/nlp.py exists on the real + PostHog-wrapped OpenAI client. + + Example: + openai_client.beta.chat.completions.parse + + checks: + wrapped_client.beta.chat.completions.parse + """ + + current = client + checked_parts = [path[0]] + + for attr in path[1:]: + checked_parts.append(attr) + + if not strict_has_attr(current, attr): + full_path = ".".join(path) + failed_path = ".".join(checked_parts) + + raise AssertionError( + f"\nInvalid OpenAI client path in backend/nlp.py line {lineno}:\n" + f" {full_path}\n\n" + f"The actual PostHog-wrapped OpenAI client does not publicly expose:\n" + f" {failed_path}\n\n" + f"Current object type before missing attr:\n" + f" {type(current)}\n\n" + f"This test uses dir(), not hasattr(), because hasattr() can be " + f"fooled by dynamic proxy objects." + ) + + current = getattr(current, attr) + + +def build_actual_posthog_wrapped_openai_client() -> Any: + """ + Build the real PostHog-wrapped OpenAI client object. + + This does NOT call OpenAI. + This does NOT send data to PostHog. + """ + + from posthog import Posthog + from posthog.ai.openai import AsyncOpenAI as PostHogAsyncOpenAI + + posthog_client = Posthog( + project_api_key="test-posthog-key", + host="https://us.i.posthog.com", + disabled=True, + ) + + try: + return PostHogAsyncOpenAI( + api_key="test-openai-key", + posthog_client=posthog_client, + ) + except TypeError: + # Some SDK versions may not accept posthog_client as a keyword. + return PostHogAsyncOpenAI( + api_key="test-openai-key", + ) + + +def test_wrong_chat_parse_path_fails_on_wrapped_client(): + """ + Negative control. + + This proves the checker is strong enough. + + openai_client.chat.completions.parse should NOT be accepted. + Structured parsing should go through: + openai_client.beta.chat.completions.parse + """ + + wrapped_client = build_actual_posthog_wrapped_openai_client() + + wrong_path = [ + "openai_client", + "chat", + "completions", + "parse", + ] + + with pytest.raises(AssertionError): + assert_path_exists_on_client( + wrapped_client, + wrong_path, + lineno=0, + ) + + +def test_nlp_openai_paths_exist_on_actual_posthog_wrapped_client(): + """ + Contract test. + + Scan backend/nlp.py for every openai_client.xxx(...) call and check whether + that exact path exists on the actual PostHog-wrapped OpenAI client. + + This does NOT call OpenAI. + """ + + wrapped_client = build_actual_posthog_wrapped_openai_client() + + paths = find_openai_client_call_paths() + + assert paths, ( + f"No real openai_client calls found in:\n" + f" {NLP_FILE}\n\n" + f"This means AST did not find real code like:\n" + f" openai_client.chat.completions.create(...)\n" + f" openai_client.beta.chat.completions.parse(...)\n" + ) + + for path, lineno in paths: + assert_path_exists_on_client(wrapped_client, path, lineno) \ No newline at end of file