Refactor architecture: Decouple infrastructure, fix RQ usage, and remove obsolete tests

Author: gabaoun

src/api/app.py

@@ -6,7 +6,7 @@
 from src.utils.logger import logger
 from src.db.session import SessionLocal, get_db
 from src.models.db import IngestionJob
-from src.worker import queue
+from src.infra.queue import get_queue
 import os
 from contextlib import asynccontextmanager
 from sqlalchemy.orm import Session
@@ -63,6 +63,7 @@ async def ingest_docs(db: Session = Depends(get_db)):
         db.refresh(job)
 
         # Enqueue the background task
+        queue = get_queue()
         queue.enqueue("src.jobs.ingestion.process_ingestion", str(job.id))
 
         logger.info(f"Ingestion job {job.id} enqueued.")

src/infra/queue.py

@@ -0,0 +1,12 @@
+import redis
+from rq import Queue
+from src.config.settings import settings
+
+# We don't initialize connection here directly to avoid side effects on import
+# instead we provide a way to get the queue when needed
+
+def get_redis_connection():
+    return redis.Redis(host=settings.redis_host, port=settings.redis_port)
+
+def get_queue(name="default"):
+    return Queue(name, connection=get_redis_connection())

src/worker.py

@@ -1,16 +1,16 @@
-import redis
-from rq import Worker, Queue, Connection
-from src.config.settings import settings
+from rq import Worker
+from src.infra.queue import get_queue, get_redis_connection
 from src.utils.logger import logger
 
-redis_conn = redis.Redis(host=settings.redis_host, port=settings.redis_port)
-queue = Queue("default", connection=redis_conn)
-
 def run_worker():
-    logger.info(f"Starting worker connected to Redis at {settings.redis_host}:{settings.redis_port}")
-    with Connection(redis_conn):
-        worker = Worker([queue])
-        worker.work()
+    redis_conn = get_redis_connection()
+    queue = get_queue()
+    
+    logger.info(f"Starting worker connected to Redis")
+    
+    # In recent RQ versions, connection is passed to the Worker constructor
+    worker = Worker([queue], connection=redis_conn)
+    worker.work()
 
 if __name__ == "__main__":
     run_worker()

tests/test_api.py

@@ -1,11 +1,32 @@
 import pytest
 from fastapi.testclient import TestClient
 from unittest.mock import MagicMock, AsyncMock
+import uuid
+
+# Mock dependencies before importing app
+import src.infra.queue
+import src.db.session
+
+# We mock the connection and session at the module level or within fixtures
 from src.api.app import app
-import src.api.app as api_app
 
 client = TestClient(app)
 
+@pytest.fixture
+def mock_db(mocker):
+    mock_session = MagicMock()
+    mocker.patch("src.db.session.SessionLocal", return_value=mock_session)
+    # Mock Depends(get_db)
+    app.dependency_overrides[src.db.session.get_db] = lambda: mock_session
+    yield mock_session
+    app.dependency_overrides.clear()
+
+@pytest.fixture
+def mock_queue(mocker):
+    mock_q = MagicMock()
+    mocker.patch("src.api.app.get_queue", return_value=mock_q)
+    return mock_q
+
 @pytest.fixture
 def mock_retrieval_wf(mocker):
     mock_wf = MagicMock()
@@ -23,7 +44,10 @@ def test_health_check():
     assert response.json()["status"] == "ok"
 
 def test_query_endpoint_uninitialized():
-    # Before initialization, it should return 503
+    # Set global retrieval_wf to None for this test
+    import src.api.app as api_app
+    api_app.retrieval_wf = None
+    
     response = client.post("/query", json={"query": "test"})
     assert response.status_code == 503
     assert "not initialized" in response.json()["detail"]
@@ -51,3 +75,28 @@ def test_query_endpoint_error(mock_retrieval_wf, mocker):
 
     assert response.status_code == 500
     assert "Retrieval failed" in response.json()["detail"]
+
+def test_ingest_endpoint(mock_db, mock_queue):
+    # Mock job creation
+    job_id = uuid.uuid4()
+    mock_db.add.side_effect = lambda job: setattr(job, 'id', job_id)
+    
+    response = client.post("/ingest")
+    
+    assert response.status_code == 202
+    assert response.json()["job_id"] == str(job_id)
+    mock_queue.enqueue.assert_called_once()
+
+def test_get_job_status(mock_db):
+    job_id = str(uuid.uuid4())
+    mock_job = MagicMock()
+    mock_job.id = job_id
+    mock_job.status = "COMPLETED"
+    
+    mock_db.query.return_value.filter.return_value.first.return_value = mock_job
+    
+    response = client.get(f"/jobs/{job_id}")
+    
+    assert response.status_code == 200
+    assert response.json()["status"] == "COMPLETED"
+    assert response.json()["id"] == job_id