import asyncio import os import pytest from unittest.mock import AsyncMock, MagicMock from agents.base_agent import AgentContext # Keep tests hermetic: text detectors ship disabled unless a test opts in. # Env vars override the dev .env file, so pin them OFF explicitly. for _k, _v in { "TEXTDET_ENABLE_FASTDETECTGPT": "false", "TEXTDET_ENABLE_BINOCULARS": "false", "TEXTDET_ENABLE_WATERMARK": "false", "TEXTDET_ENABLE_PROVENANCE": "false", }.items(): os.environ[_k] = _v @pytest.fixture(scope="session") def event_loop(): loop = asyncio.new_event_loop() yield loop loop.close() @pytest.fixture def mock_neo4j(): neo4j = MagicMock() neo4j.run.return_value = [] neo4j.run_write.return_value = [] return neo4j @pytest.fixture def mock_mongo(): mongo = MagicMock() mongo.upsert_job = AsyncMock() mongo.get_job = AsyncMock(return_value={"job_id": "test_job", "status": "running"}) mongo.save_result = AsyncMock() return mongo @pytest.fixture def mock_llm(): llm = MagicMock() llm.complete = AsyncMock(return_value="Test summary response.") llm.complete_json = AsyncMock(return_value={"overall_score": 7.0}) llm.complete_list = AsyncMock( return_value=[ { "text": "Test claim", "type": "significant", "confidence": 0.7, "novelty": 0.8, } ] ) return llm @pytest.fixture def mock_embedder(): embedder = MagicMock() embedder.embed.return_value = [0.1] * 768 embedder.embed_batch.return_value = [[0.1] * 768] return embedder @pytest.fixture def mock_graph_memory(mock_neo4j, mock_embedder): from memory.graph_memory import GraphMemory gm = GraphMemory(neo4j=mock_neo4j, embedder=mock_embedder) return gm @pytest.fixture def agent_ctx( mock_neo4j, mock_mongo, mock_llm, mock_embedder, mock_graph_memory, tmp_path ): # Create a minimal PDF-like file for testing pdf_file = tmp_path / "test.pdf" pdf_file.write_bytes(b"%PDF-1.4 test") return AgentContext( job_id="test_job", doc_id="test_doc", pdf_path=str(pdf_file), neo4j=mock_neo4j, mongo=mock_mongo, llm=mock_llm, embedder=mock_embedder, graph_memory=mock_graph_memory, )