Merge pull request #19 from kundhu-codes/routes_tests

tests: add async route tests

Author: sausi-7

pyproject.toml

@@ -55,3 +55,7 @@ include = ["inkwell*"]
 [tool.ruff]
 line-length = 100
 target-version = "py311"
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"

tests/test_routes/__init__.py

@@ -0,0 +1,125 @@
+import pytest
+import httpx
+from unittest.mock import patch
+
+from inkwell.app import create_app
+from inkwell.routes.api_profile import _is_meaningfully_configured
+
+# ------------------------------------------------------------------
+# Fixtures
+# ------------------------------------------------------------------
+
+
+@pytest.fixture
+async def client():
+    """Create a fresh Async httpx Client for each test."""
+    app = create_app()
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
+        yield ac
+
+
+# ------------------------------------------------------------------
+# Logic Tests
+# ------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "personality_dict, expected",
+    [
+        ({"bio": "I am a dev", "example_comments": ["Comment 1"]}, True),
+        ({"bio": "", "example_comments": ["Comment 1"]}, False),
+        ({"bio": "I am a dev", "example_comments": []}, False),
+        ({"bio": "    ", "example_comments": ["Comment 1"]}, False),
+        ({}, False),
+    ],
+    ids=[
+        "fully_configured_profile",
+        "missing_bio_content",
+        "missing_example_comments",
+        "bio_with_only_whitespace",
+        "totally_empty_dictionary",
+    ],
+)
+def test_is_meaningfully_configured(personality_dict, expected):
+    assert _is_meaningfully_configured(personality_dict) is expected
+
+
+# ------------------------------------------------------------------
+# Tests for GET /api/profile
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_profile.load_personality")
+async def test_get_profile_success(mock_load, client):
+    mock_load.return_value = {
+        "name": "Name",
+        "bio": "A multi-line bio string\nthat represents a persona.\n",
+        "example_comments": ["Example 1"],
+    }
+
+    response = await client.get("/api/profile")
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["profile"]["name"] == "Name"
+    assert data["configured"] is True
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_profile.load_personality")
+async def test_get_profile_not_configured(mock_load, client):
+    mock_load.return_value = {}
+
+    response = await client.get("/api/profile")
+    assert response.status_code == 200
+    assert response.json()["configured"] is False
+
+
+# ------------------------------------------------------------------
+# Tests for POST /api/profile
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload, expected_tone_structure",
+    [
+        (
+            {"name": "Name", "tone": {"style": "witty", "humor": "dry", "formality": "casual"}},
+            {"style": "witty", "humor": "dry", "formality": "casual"},
+        ),
+        ({"name": "Name1", "tone": {"style": "", "humor": "", "formality": ""}}, {}),
+    ],
+    ids=["save_full", "save_empty_tone"],
+)
+@patch("inkwell.routes.api_profile.write_yaml")
+async def test_save_profile_scenarios(mock_write, client, payload, expected_tone_structure):
+    mock_write.return_value = "personality.yml"
+
+    # Await the post call
+    response = await client.post("/api/profile", json=payload)
+
+    assert response.status_code == 200
+    actual_data_saved = mock_write.call_args[0][1]
+    assert actual_data_saved["tone"] == expected_tone_structure
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_profile.write_yaml")
+async def test_save_profile_internal_error_handling(mock_write, client):
+    mock_write.side_effect = Exception("Disk Full")
+
+    response = await client.post("/api/profile", json={"name": "User"})
+
+    assert response.status_code == 500
+    assert "Write failed" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_save_profile_validation_error(client):
+    bad_payload = {"name": "User", "interests": "not-a-list"}
+
+    response = await client.post("/api/profile", json=bad_payload)
+    assert response.status_code == 422

tests/test_routes/test_api_profile.py

@@ -0,0 +1,165 @@
+import pytest
+import json
+import time
+from queue import Queue
+from unittest.mock import patch
+from httpx import AsyncClient, ASGITransport
+import pytest_asyncio
+
+from inkwell.app import create_app
+from inkwell.routes.api_scan import _scan_state
+
+# ------------------------------------------------------------------
+# Fixtures
+# ------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def client():
+    _scan_state["running"] = False
+    _scan_state["started_at"] = None
+    _scan_state["current_subreddit"] = None
+    _scan_state["cancel"] = False
+    _scan_state["queue"] = None
+
+    app = create_app()
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+# ------------------------------------------------------------------
+# Tests for GET /api/scan/status
+# ------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_get_scan_status_idle(client):
+    response = await client.get("/api/scan/status")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["running"] is False
+    assert data["current_subreddit"] is None
+
+
+@pytest.mark.asyncio
+async def test_get_scan_status_running(client):
+    _scan_state["running"] = True
+    _scan_state["current_subreddit"] = "r/python"
+
+    response = await client.get("/api/scan/status")
+    data = response.json()
+    assert data["running"] is True
+    assert data["current_subreddit"] == "r/python"
+
+
+# ------------------------------------------------------------------
+# Tests for POST /api/scan
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_scan._run_scan_with_emit")
+async def test_start_scan_success(mock_run, client):
+
+    def mock_emit_logic(emit, options):
+        emit({"kind": "start", "subreddit_count": 2})
+        emit({"kind": "done", "scanned": 2, "new_signals": 0})
+
+    mock_run.side_effect = mock_emit_logic
+
+    response = await client.post("/api/scan", json={"limit_subreddits": 2})
+
+    assert response.status_code == 200
+    assert "text/event-stream" in response.headers["content-type"]
+
+    events = response.text.strip().split("\n\n")
+    assert 'data: {"kind": "start", "subreddit_count": 2}' in events[0]
+    assert 'data: {"kind": "done"' in events[1]
+
+    assert _scan_state["running"] is False
+
+
+@pytest.mark.asyncio
+async def test_start_scan_conflict(client):
+    _scan_state["running"] = True
+
+    response = await client.post("/api/scan", json={})
+    assert response.status_code == 409
+    assert "already running" in response.json()["detail"]
+
+
+@patch("inkwell.routes.api_scan.logger")
+@patch("inkwell.routes.api_scan._run_scan_with_emit")
+@pytest.mark.asyncio
+async def test_start_scan_exception_handling(mock_run, mock_logger, client):
+    mock_run.side_effect = Exception("Reddit API Down/Timeout")
+
+    response = await client.post("/api/scan", json={})
+
+    assert 'data: {"kind": "error", "message": "Reddit API Down/Timeout"}' in response.text
+    assert _scan_state["running"] is False
+
+
+@patch("inkwell.routes.api_scan._run_scan_with_emit")
+@pytest.mark.asyncio
+async def test_scan_heartbeat_precedes_completion_on_stall(mock_run, client):
+    def stalling_scan(emit, options):
+        # 7s stall > 5s heartbeat interval
+        time.sleep(7.0)
+        emit({"kind": "done", "scanned": 1})
+
+    mock_run.side_effect = stalling_scan
+
+    response = await client.post("/api/scan", json={"limit_subreddits": 1})
+
+    events = [ev for ev in response.text.strip().split("\n\n") if ev.startswith("data:")]
+
+    heartbeat_index = next((i for i, ev in enumerate(events) if '"kind": "heartbeat"' in ev), -1)
+    done_index = next((i for i, ev in enumerate(events) if '"kind": "done"' in ev), -1)
+
+    assert heartbeat_index != -1, "No heartbeat detected"
+    assert done_index != -1, "Scan never finished"
+
+    assert heartbeat_index < done_index, (
+        f"Heartbeat (pos {heartbeat_index}) should have arrived "
+        f"before scan completion (pos {done_index})"
+    )
+
+    heartbeat_data = json.loads(events[heartbeat_index].replace("data: ", ""))
+    assert heartbeat_data["elapsed_s"] >= 5.0
+
+
+@patch("inkwell.routes.api_scan._run_scan_with_emit")
+@pytest.mark.asyncio
+async def test_rapid_restart(mock_run, client):
+    await client.post("/api/scan")
+    response = await client.post("/api/scan")
+    assert response.status_code == 200  # Should NOT be 409
+
+
+# ------------------------------------------------------------------
+# Tests for POST /api/scan/stop
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_stop_scan_not_running(client):
+    response = await client.post("/api/scan/stop")
+    assert response.json() == {"ok": False, "detail": "No scan is running."}
+
+
+@pytest.mark.asyncio
+async def test_stop_scan_success(client):
+    """Verify stop sets cancel flag and injects event into queue."""
+    mock_queue = Queue()
+    _scan_state["running"] = True
+    _scan_state["queue"] = mock_queue
+
+    response = await client.post("/api/scan/stop")
+
+    assert response.status_code == 200
+    assert response.json()["ok"] is True
+    assert _scan_state["cancel"] is True
+
+    # Verify the 'cancelled' event was pushed to wake up the SSE stream
+    injected_event = mock_queue.get()
+    assert injected_event["kind"] == "cancelled"

tests/test_routes/test_api_scan.py

@@ -0,0 +1,82 @@
+import pytest
+from unittest.mock import patch
+from httpx import AsyncClient, ASGITransport
+from inkwell.app import create_app
+import pytest_asyncio
+
+
+# ------------------------------------------------------------------
+# Fixtures
+# ------------------------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def client():
+    app = create_app()
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+# ------------------------------------------------------------------
+# Tests for POST /api/settings/test-llm
+# ------------------------------------------------------------------
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_settings.litellm.completion")
+async def test_llm_probe_success(mock_completion, client):
+    mock_completion.return_value = {"choices": [{"message": {"content": "pong"}}]}
+
+    payload = {"model": "gpt-4o"}
+    headers = {"X-LLM-Key": "sk-valid-key"}
+
+    response = await client.post("/api/settings/test-llm", json=payload, headers=headers)
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["ok"] is True
+    assert "gpt-4o responded" in data["detail"]
+
+    args, kwargs = mock_completion.call_args
+    assert kwargs["model"] == "gpt-4o"
+    assert kwargs["api_key"] == "sk-valid-key"
+    assert kwargs["max_tokens"] == 1
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_settings.litellm.completion")
+async def test_llm_probe_handles_auth_failure_and_redacts_key(mock_completion, client):
+    secret_key = "sk-very-secret-123"
+    mock_completion.side_effect = Exception(f"Invalid Request: Unauthorized for key {secret_key}")
+
+    payload = {"model": "claude-3-opus"}
+    headers = {"X-LLM-Key": secret_key}
+
+    response = await client.post("/api/settings/test-llm", json=payload, headers=headers)
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["ok"] is False
+    assert secret_key not in data["detail"]
+    assert "[redacted]" in data["detail"]
+
+
+@pytest.mark.asyncio
+async def test_llm_probe_fails_if_model_missing(client):
+    response = await client.post("/api/settings/test-llm", json={"model": "  "})
+
+    assert response.status_code == 200  # App logic returns ok: False instead of 422
+    assert response.json()["ok"] is False
+    assert "Model is required" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+@patch("inkwell.routes.api_settings.litellm.completion")
+async def test_llm_probe_works_without_header(mock_completion, client):
+    mock_completion.return_value = {"ok": True}
+
+    response = await client.post("/api/settings/test-llm", json={"model": "gpt-3.5-turbo"})
+
+    assert response.status_code == 200
+    # Ensure api_key was NOT passed to litellm if header was missing
+    _, kwargs = mock_completion.call_args
+    assert "api_key" not in kwargs