test: Update tests for jobs endpoint

app/worker.py

@@ -0,0 +1,153 @@
+import asyncio
+from typing import Any, Dict, List
+from celery import Celery
+from app.core.config import settings
+from app.core.logging import logger
+from app.db.session import AsyncSessionLocal
+from app.repositories.job_repository import JobRepository
+
+# Define Celery app
+celery_app = Celery(
+    "alemno_worker",
+    broker=settings.CELERY_BROKER_URL,
+    backend=settings.CELERY_RESULT_BACKEND,
+)
+
+# Optional configurations
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    result_serializer="json",
+    timezone="UTC",
+    enable_utc=True,
+)
+
+
+def run_async(coro):
+    """Helper to run async functions synchronously in Celery worker context"""
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+@celery_app.task(name="tasks.process_transaction_job")
+def process_transaction_job(job_id: str, transactions: List[Dict[str, Any]]):
+    """
+    Background job to clean transactions, detect anomalies, calculate spend breakdown,
+    generate narrative summary, and persist results.
+    """
+    logger.info(f"Starting Celery task to process job_id: {job_id}")
+    return run_async(_process_job_async(job_id, transactions))
+
+
+async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]):
+    async with AsyncSessionLocal() as db:
+        repo = JobRepository(db)
+        job = await repo.get_by_id(job_id)
+        if not job:
+            logger.error(f"Job with ID {job_id} not found in database.")
+            return
+
+        try:
+            # Update status to processing
+            await repo.update(job, status="processing")
+            await db.commit()
+
+            # 1. Clean transactions
+            cleaned_transactions = []
+            total_amount = 0.0
+            categories = {}
+            for t in transactions:
+                # Clean description and category
+                desc = t["description"].strip()
+                cat = t["category"].strip().capitalize()
+                amount = float(t["amount"])
+
+                total_amount += amount
+                categories[cat] = categories.get(cat, 0.0) + amount
+
+                cleaned_transactions.append(
+                    {
+                        "transaction_id": t["transaction_id"],
+                        "date": t["date"],
+                        "amount": amount,
+                        "category": cat,
+                        "description": desc,
+                    }
+                )
+
+            # 2. Flag anomalies (Heuristic: > $5,000, or containing suspicious keywords)
+            anomalies = []
+            for t in cleaned_transactions:
+                is_anomaly = False
+                reasons = []
+
+                if t["amount"] > 5000.0:
+                    is_anomaly = True
+                    reasons.append("High value transaction (> $5,000)")
+
+                desc_lower = t["description"].lower()
+                if any(
+                    k in desc_lower
+                    for k in ["suspend", "hack", "error", "unknown", "fraud"]
+                ):
+                    is_anomaly = True
+                    reasons.append("Suspicious keyword found in description")
+
+                if is_anomaly:
+                    anomalies.append({**t, "reasons": reasons})
+
+            # 3. Spend breakdown (percentage & absolute)
+            spend_breakdown = {}
+            for cat, amount in categories.items():
+                pct = (amount / total_amount * 100) if total_amount > 0 else 0
+                spend_breakdown[cat] = {
+                    "total_spend": round(amount, 2),
+                    "percentage": round(pct, 2),
+                }
+
+            # 4. Generate LLM Narrative Summary (Mocked / simulated intelligence)
+            top_category = max(categories, key=categories.get) if categories else "None"
+            anomaly_pct = (
+                (len(anomalies) / len(transactions) * 100) if transactions else 0
+            )
+            narrative = (
+                f"Successfully parsed and processed {len(transactions)} transaction records. "
+                f"Total spend across all transactions is ${total_amount:,.2f}. "
+                f"We identified '{top_category}' as the largest spending category, amounting to "
+                f"${categories.get(top_category, 0):,.2f} ({spend_breakdown.get(top_category, {}).get('percentage', 0)}% of total). "
+                f"A scan for transactional anomalies flagged {len(anomalies)} records ({anomaly_pct:.1f}% of total). "
+                f"Most anomalies were triggered by high transaction limits or unrecognized transaction descriptions. "
+                f"Recommendation: Review flagged anomalies to prevent potential leakages."
+            )
+
+            # Build full structured output
+            results = {
+                "cleaned_transactions": cleaned_transactions,
+                "flagged_anomalies": anomalies,
+                "spend_breakdown": spend_breakdown,
+                "narrative_summary": narrative,
+            }
+
+            # Build high-level stats for status summary
+            summary = {
+                "total_rows": len(transactions),
+                "total_amount": round(total_amount, 2),
+                "anomalies_count": len(anomalies),
+                "top_category": top_category,
+            }
+
+            # Update job in DB
+            await repo.update(
+                job,
+                status="completed",
+                row_count=len(transactions),
+                summary=summary,
+                results=results,
+            )
+            await db.commit()
+            logger.info(f"Job {job_id} processing completed successfully.")
+
+        except Exception as e:
+            logger.exception(f"Error processing job {job_id}")
+            await repo.update(job, status="failed")
+            await db.commit()
+            raise e

docker-compose.yml

@@ -6,6 +6,26 @@ services:
     container_name: alemno_db
     ports:
       - "5432:5432"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: alemno_payments
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres -d alemno_payments"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  redis:
+    image: redis:7-alpine
+    container_name: alemno_redis
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+
   web:
     build: .
     container_name: alemno_web
@@ -13,8 +33,42 @@ services:
       - "3000:3000"
     volumes:
       - .:/app
+    environment:
+      - POSTGRES_HOST=db
+      - POSTGRES_PORT=5432
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+      - POSTGRES_DB=alemno_payments
+      - REDIS_HOST=redis
+      - REDIS_PORT=6379
+      - REDIS_DB=0
     depends_on:
       db:
         condition: service_healthy
+      redis:
+        condition: service_started
+
+  worker:
+    build: .
+    container_name: alemno_worker
+    volumes:
+      - .:/app
+    environment:
+      - POSTGRES_HOST=db
+      - POSTGRES_PORT=5432
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+      - POSTGRES_DB=alemno_payments
+      - REDIS_HOST=redis
+      - REDIS_PORT=6379
+      - REDIS_DB=0
+    command: celery -A app.worker.celery_app worker --loglevel=info
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_started
+
 volumes:
   postgres_data:
+  redis_data:

tests/api/test_jobs_api.py

@@ -2,8 +2,14 @@ import io
 import pytest
 from httpx import AsyncClient
 from app.worker import celery_app
+
+# Mark all tests as async
 pytestmark = pytest.mark.asyncio
+
+# Set celery to run tasks eagerly (synchronously) during API testing
 celery_app.conf.task_always_eager = True
+
+
 async def test_upload_csv_success(client: AsyncClient):
     csv_content = (
         "transaction_id,date,amount,category,description\n"
@@ -18,6 +24,8 @@ async def test_upload_csv_success(client: AsyncClient):
             "text/csv",
         )
     }
+
+    # 1. Post upload CSV
     response = await client.post("/api/v1/jobs/upload", files=files)
     assert response.status_code == 201
 
@@ -25,7 +33,11 @@ async def test_upload_csv_success(client: AsyncClient):
     assert "id" in data
     assert data["filename"] == "transactions.csv"
     assert data["status"] == "pending"  # Initial state returned by router immediately
+
     job_id = data["id"]
+
+    # Since Celery is in eager mode, the task should have completed immediately.
+    # 2. Get Job Status
     status_response = await client.get(f"/api/v1/jobs/{job_id}/status")
     assert status_response.status_code == 200
     status_data = status_response.json()
@@ -35,6 +47,8 @@ async def test_upload_csv_success(client: AsyncClient):
     assert status_data["summary"]["total_amount"] == 15120.50
     assert status_data["summary"]["anomalies_count"] == 1
     assert status_data["summary"]["top_category"] == "Electronics"
+
+    # 3. Get Job Results
     results_response = await client.get(f"/api/v1/jobs/{job_id}/results")
     assert results_response.status_code == 200
     results_data = results_response.json()
@@ -47,7 +61,10 @@ async def test_upload_csv_success(client: AsyncClient):
     assert results["flagged_anomalies"][0]["transaction_id"] == "TX1002"
     assert results["spend_breakdown"]["Groceries"]["total_spend"] == 120.50
     assert "Successfully parsed and processed" in results["narrative_summary"]
+
+
 async def test_upload_invalid_csv(client: AsyncClient):
+    # Invalid CSV structure (missing required column)
     csv_content = (
         "transaction_id,amount,category,description\n"
         "TX1001,120.50,Groceries,Weekly grocery shop\n"