local-deep-research/tests/api_tests/test_research_creation.py

#!/usr/bin/env python3
"""
Test research creation endpoint specifically
"""

import json
import time

import pytest


@pytest.mark.requires_llm
class TestResearchCreation:
    """Test research creation functionality.

    All tests in this class POST to /api/start_research with real
    Ollama / searxng / model dependencies (PUNCHLIST Tier 4: FLAKY_SLEEP).
    The class-level @pytest.mark.requires_llm gate makes them auto-
    skip in environments without a live LLM stack, so CI doesn't
    flake on the timing-sensitive sleep(0.5) and model-availability
    assertions. Manual runs can invoke them with -m requires_llm.
    """

    def test_research_creation_endpoint(self, authenticated_client):
        """Test the research creation endpoint."""
        print("\nTesting /api/start_research endpoint...")

        research_data = {
            "query": "Test research query from Python",
            "mode": "quick",
            "model": "gpt-3.5-turbo",
            "search_engines": ["searxng"],
            "local_context": 2000,
            "web_context": 2000,
            "temperature": 0.7,
        }

        response = authenticated_client.post(
            "/api/start_research",
            json=research_data,
            content_type="application/json",
        )

        print(f"Response status: {response.status_code}")

        assert response.status_code == 200
        data = json.loads(response.data)

        if data.get("status") == "success":
            assert "research_id" in data
            print(f"\n✅ SUCCESS! Research ID: {data.get('research_id')}")
            return data["research_id"]
        # Alternative response format
        assert "research_id" in data
        print(f"\n✅ SUCCESS! Research ID: {data['research_id']}")
        return data["research_id"]

    def test_research_creation_with_minimal_params(self, authenticated_client):
        """Test research creation with minimal parameters."""
        research_data = {
            "query": "Minimal test query",
            "mode": "quick",
            "model": "llama2",
            "model_provider": "OLLAMA",
        }

        print(f"\n[DEBUG] Sending minimal research request: {research_data}")
        response = authenticated_client.post(
            "/api/start_research",
            json=research_data,
            content_type="application/json",
        )

        print(f"[DEBUG] Response status: {response.status_code}")
        if response.status_code != 200:
            print(f"[DEBUG] Response data: {response.data.decode()[:500]}")

        assert response.status_code == 200
        data = json.loads(response.data)
        assert "research_id" in data or (
            data.get("status") == "success" and "research_id" in data
        )

    def test_research_creation_validation(self, authenticated_client):
        """Test research creation validation."""
        # Missing query
        response = authenticated_client.post(
            "/api/start_research",
            json={"mode": "quick"},
            content_type="application/json",
        )

        assert response.status_code == 400
        data = json.loads(response.data)
        assert "error" in data or "message" in data

    def test_research_status_check(self, authenticated_client):
        """Test checking research status."""
        # First create a research
        research_data = {
            "query": "Status check test",
            "mode": "quick",
            "model": "llama2",
            "model_provider": "OLLAMA",
        }

        response = authenticated_client.post(
            "/api/start_research",
            json=research_data,
            content_type="application/json",
        )

        assert response.status_code == 200
        data = json.loads(response.data)

        # Extract research_id
        if "research_id" in data:
            research_id = data["research_id"]
        else:
            research_id = data.get("data", {}).get("research_id")

        assert research_id is not None

        # Give it a moment to start
        time.sleep(0.5)  # allow: unmarked-sleep

        # Check status
        response = authenticated_client.get(
            f"/api/research/{research_id}/status"
        )
        assert response.status_code == 200

        status_data = json.loads(response.data)
        assert "status" in status_data or "research_status" in status_data

    def test_research_termination(self, authenticated_client):
        """Test terminating a research."""
        # First create a research
        research_data = {
            "query": "Termination test",
            "mode": "quick",
            "model": "llama2",
            "model_provider": "OLLAMA",
        }

        response = authenticated_client.post(
            "/api/start_research",
            json=research_data,
            content_type="application/json",
        )

        assert response.status_code == 200
        data = json.loads(response.data)
        research_id = data.get("research_id") or data.get("data", {}).get(
            "research_id"
        )
        assert research_id is not None

        # Give it a moment to start
        time.sleep(0.5)  # allow: unmarked-sleep

        # Terminate it
        response = authenticated_client.post(f"/api/terminate/{research_id}")
        assert response.status_code in [200, 404]  # 404 if already finished

    def test_research_modes(self, authenticated_client):
        """Test different research modes."""
        modes = ["quick", "normal", "comprehensive"]

        for mode in modes:
            research_data = {
                "query": f"Test {mode} mode",
                "mode": mode,
                "model": "llama2",
                "model_provider": "OLLAMA",
            }

            response = authenticated_client.post(
                "/api/start_research",
                json=research_data,
                content_type="application/json",
            )

            assert response.status_code == 200
            data = json.loads(response.data)
            assert "research_id" in data or (
                data.get("status") == "success" and "research_id" in data
            )
            print(f"✅ {mode} mode research created successfully")

            # Terminate to clean up
            research_id = data.get("research_id") or data.get("data", {}).get(
                "research_id"
            )
            if research_id:
                authenticated_client.post(f"/api/terminate/{research_id}")

    def test_research_with_custom_model(self, authenticated_client):
        """Test research with custom model settings."""
        research_data = {
            "query": "Custom model test",
            "mode": "quick",
            "model_provider": "OLLAMA",
            "model": "mistral",
            "temperature": 0.5,
            "max_tokens": 1000,
        }

        response = authenticated_client.post(
            "/api/start_research",
            json=research_data,
            content_type="application/json",
        )

        assert response.status_code == 200
        data = json.loads(response.data)
        assert "research_id" in data or (
            data.get("status") == "success" and "research_id" in data
        )