# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client """Tests for the per-session concurrency guard on POST /api/chat/sessions//messages. Chat is sequential by design: at most one in-flight research per session. A second send while a research is IN_PROGRESS must return 409 with the active research_id. """ import json from unittest.mock import patch def _create_session(client, query="Test"): resp = client.post( "/api/chat/sessions", json={"initial_query": query}, content_type="application/json", ) assert resp.status_code == 200, resp.data return json.loads(resp.data)["session_id"] class TestPerSessionConcurrencyGuard: """Per-session 409 guard tests.""" def test_second_send_while_in_progress_returns_409( self, authenticated_client ): """A second send while the first research is still IN_PROGRESS must return 409 with the active research_id.""" session_id = _create_session(authenticated_client) with patch("local_deep_research.chat.routes.start_research_process"): first = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "first", "trigger_research": True}, content_type="application/json", ) assert first.status_code == 200 first_rid = json.loads(first.data)["research_id"] assert first_rid # Second send: research_id from first is still IN_PROGRESS # because start_research_process was mocked (no thread ran), # so the row remains at status=in_progress. second = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "second", "trigger_research": True}, content_type="application/json", ) assert second.status_code == 409, second.data data = json.loads(second.data) assert data["success"] is False assert "in progress" in data["error"].lower() assert data.get("active_research_id") == first_rid def test_second_send_after_terminate_succeeds(self, authenticated_client): """Once the first research is terminated (status leaves IN_PROGRESS), a second send is accepted normally — i.e. the guard releases.""" session_id = _create_session(authenticated_client) with patch("local_deep_research.chat.routes.start_research_process"): first = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "first", "trigger_research": True}, content_type="application/json", ) assert first.status_code == 200 first_rid = json.loads(first.data)["research_id"] # Terminate via the existing endpoint — this is the real user flow: # click Stop, then retry. The endpoint flips status off IN_PROGRESS # without requiring the worker thread to have started. term_resp = authenticated_client.post(f"/api/terminate/{first_rid}") assert term_resp.status_code == 200, term_resp.data with patch("local_deep_research.chat.routes.start_research_process"): second = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "second", "trigger_research": True}, content_type="application/json", ) assert second.status_code == 200, second.data assert json.loads(second.data)["research_id"] != first_rid def test_guard_does_not_apply_when_trigger_research_false( self, authenticated_client ): """A trigger_research=False send must succeed even when a research is in flight — non-research messages don't conflict.""" session_id = _create_session(authenticated_client) with patch("local_deep_research.chat.routes.start_research_process"): first = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "first", "trigger_research": True}, content_type="application/json", ) assert first.status_code == 200 second = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "no-research", "trigger_research": False}, content_type="application/json", ) assert second.status_code == 200, second.data assert json.loads(second.data)["research_id"] is None def test_guard_isolates_per_session(self, authenticated_client): """An in-flight research on session A must not block sends on session B for the same user.""" session_a = _create_session(authenticated_client, query="topic A") session_b = _create_session(authenticated_client, query="topic B") with patch("local_deep_research.chat.routes.start_research_process"): ra = authenticated_client.post( f"/api/chat/sessions/{session_a}/messages", json={"content": "first on A", "trigger_research": True}, content_type="application/json", ) assert ra.status_code == 200 # B should be unaffected. rb = authenticated_client.post( f"/api/chat/sessions/{session_b}/messages", json={"content": "first on B", "trigger_research": True}, content_type="application/json", ) assert rb.status_code == 200, rb.data def test_blocked_send_does_not_create_orphan_user_message( self, authenticated_client ): """When the guard returns 409, the rejected user message must NOT be in chat_messages — otherwise we'd accumulate orphan rows.""" session_id = _create_session(authenticated_client) with patch("local_deep_research.chat.routes.start_research_process"): authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": "first", "trigger_research": True}, content_type="application/json", ) blocked = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={ "content": "should-be-rejected", "trigger_research": True, }, content_type="application/json", ) assert blocked.status_code == 409 # Read messages back; only the first user message should exist. list_resp = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages" ) assert list_resp.status_code == 200 msgs = json.loads(list_resp.data)["messages"] user_contents = [m["content"] for m in msgs if m["role"] == "user"] assert "should-be-rejected" not in user_contents