From 40881b47735968318dcc34150b9f47088fa67d4c Mon Sep 17 00:00:00 2001 From: Aditya Gupta Date: Mon, 29 Jun 2026 17:52:08 +0530 Subject: [PATCH] test: Update tests for jobs endpoint --- app/worker.py | 153 +++++++++++++++++++++++++++++++++++++ docker-compose.yml | 54 +++++++++++++ tests/api/test_jobs_api.py | 17 +++++ 3 files changed, 224 insertions(+) create mode 100644 app/worker.py diff --git a/app/worker.py b/app/worker.py new file mode 100644 index 0000000..1df9a6f --- /dev/null +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml index 6345c7d..a7ad115 100644 --- a/docker-compose.yml +++ b/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: diff --git a/tests/api/test_jobs_api.py b/tests/api/test_jobs_api.py index 5cd45ef..e405ffb 100644 --- a/tests/api/test_jobs_api.py +++ b/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"