diff --git a/app/clients/exchange_rate_client.py b/app/clients/exchange_rate_client.py new file mode 100644 index 0000000..dc4c34b --- /dev/null +++ b/app/clients/exchange_rate_client.py @@ -0,0 +1,70 @@ +import httpx +import redis +from app.core.config import settings +from app.core.logging import logger + + +class ExchangeRateClient: + def __init__(self): + self.default_rate = 93.0 + self.cache_key = "usd_to_inr_rate" + self.cache_ttl = 86400 # 24 hours cache duration + try: + self.redis_client = redis.Redis( + host=settings.REDIS_HOST, + port=settings.REDIS_PORT, + db=settings.REDIS_DB, + decode_responses=True, + ) + except Exception as e: + logger.warning(f"Could not connect to Redis for exchange rate caching: {e}") + self.redis_client = None + + async def get_usd_to_inr_rate(self) -> float: + """ + Retrieves the USD to INR conversion rate. + Checks Redis cache first. If cache is empty/offline, fetches from a free public API + and updates the cache. Falls back to a hardcoded default rate if both fail. + """ + # read from cache + if self.redis_client: + try: + cached_rate = self.redis_client.get(self.cache_key) + if cached_rate: + logger.info("Retrieved exchange rate from Redis cache.") + return float(cached_rate) + except Exception as e: + logger.warning(f"Failed to read exchange rate from Redis: {e}") + + # fetching from Exchange Rate API + try: + logger.info("Fetching live exchange rate from API...") + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.get("https://open.er-api.com/v6/latest/USD") + if response.status_code == 200: + data = response.json() + rates = data.get("rates", {}) + inr_rate = rates.get("INR") + if inr_rate: + inr_rate = float(inr_rate) + logger.info( + f"Successfully fetched live exchange rate: {inr_rate}" + ) + # Cache the rate in Redis + if self.redis_client: + try: + self.redis_client.set( + self.cache_key, str(inr_rate), ex=self.cache_ttl + ) + except Exception as ce: + logger.warning( + f"Failed to cache exchange rate in Redis: {ce}" + ) + return inr_rate + except Exception as e: + logger.error( + f"Error fetching live exchange rate: {e}. Falling back to default rate {self.default_rate}" + ) + + # 3. Fallback + return self.default_rate diff --git a/app/db/models/__init__.py b/app/db/models/__init__.py index fa9d0d5..c2e18bd 100644 --- a/app/db/models/__init__.py +++ b/app/db/models/__init__.py @@ -1,4 +1,6 @@ from app.db.base import Base from app.db.models.job import Job +from app.db.models.transaction import Transaction +from app.db.models.job_summary import JobSummary -__all__ = ["Base", "Job"] +__all__ = ["Base", "Job", "Transaction", "JobSummary"] diff --git a/app/db/models/job.py b/app/db/models/job.py index 73dc2d2..bb40f15 100644 --- a/app/db/models/job.py +++ b/app/db/models/job.py @@ -1,10 +1,15 @@ +from __future__ import annotations import uuid from datetime import datetime -from typing import Any, Dict, Optional -from sqlalchemy import String, Integer, DateTime, JSON, func -from sqlalchemy.orm import Mapped, mapped_column +from typing import List, Optional, TYPE_CHECKING +from sqlalchemy import DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.base import Base +if TYPE_CHECKING: + from app.db.models.transaction import Transaction + from app.db.models.job_summary import JobSummary + class Job(Base): __tablename__ = "jobs" @@ -21,24 +26,31 @@ class Job(Base): default="pending", index=True, ) - row_count: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + row_count_raw: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + row_count_clean: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) - updated_at: Mapped[datetime] = mapped_column( + completed_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), - server_default=func.now(), - onupdate=func.now(), - nullable=False, - ) - # JSON summaries and results - summary: Mapped[Optional[Dict[str, Any]]] = mapped_column( - JSON, nullable=True, ) - results: Mapped[Optional[Dict[str, Any]]] = mapped_column( - JSON, + error_message: Mapped[Optional[str]] = mapped_column( + String(500), nullable=True, ) + + # Relationships + transactions: Mapped[List["Transaction"]] = relationship( + "Transaction", + back_populates="job", + cascade="all, delete-orphan", + ) + summary: Mapped[Optional["JobSummary"]] = relationship( + "JobSummary", + back_populates="job", + cascade="all, delete-orphan", + uselist=False, + ) diff --git a/app/db/models/job_summary.py b/app/db/models/job_summary.py new file mode 100644 index 0000000..82383ff --- /dev/null +++ b/app/db/models/job_summary.py @@ -0,0 +1,38 @@ +from __future__ import annotations +import uuid +from typing import Any, Dict, Optional, TYPE_CHECKING +from sqlalchemy import Float, ForeignKey, Integer, JSON, String +from sqlalchemy.orm import Mapped, mapped_column, relationship +from app.db.base import Base + +if TYPE_CHECKING: + from app.db.models.job import Job + + +class JobSummary(Base): + __tablename__ = "job_summaries" + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=lambda: str(uuid.uuid4()), + index=True, + ) + job_id: Mapped[str] = mapped_column( + String(36), + ForeignKey("jobs.id", ondelete="CASCADE"), + nullable=False, + unique=True, + index=True, + ) + total_spend_inr: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + total_spend_usd: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + top_merchants: Mapped[Dict[str, Any]] = mapped_column( + JSON, nullable=False, default=dict + ) + anomaly_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + narrative: Mapped[Optional[str]] = mapped_column(String(2000), nullable=True) + risk_level: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) + + # Relationships + job: Mapped["Job"] = relationship("Job", back_populates="summary") diff --git a/app/db/models/transaction.py b/app/db/models/transaction.py new file mode 100644 index 0000000..49fc7af --- /dev/null +++ b/app/db/models/transaction.py @@ -0,0 +1,43 @@ +from __future__ import annotations +import uuid +from datetime import date +from typing import Optional, TYPE_CHECKING +from sqlalchemy import Boolean, Date, Float, ForeignKey, String +from sqlalchemy.orm import Mapped, mapped_column, relationship +from app.db.base import Base + +if TYPE_CHECKING: + from app.db.models.job import Job + + +class Transaction(Base): + __tablename__ = "transactions" + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=lambda: str(uuid.uuid4()), + index=True, + ) + job_id: Mapped[str] = mapped_column( + String(36), + ForeignKey("jobs.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + txn_id: Mapped[str] = mapped_column(String(100), nullable=False) + date: Mapped[date] = mapped_column(Date, nullable=False) + merchant: Mapped[str] = mapped_column(String(255), nullable=False) + amount: Mapped[float] = mapped_column(Float, nullable=False) + currency: Mapped[str] = mapped_column(String(10), nullable=False, default="INR") + status: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) + category: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + account_id: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + is_anomaly: Mapped[bool] = mapped_column(Boolean, default=False) + anomaly_reason: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) + llm_category: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + llm_raw_response: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True) + llm_failed: Mapped[Optional[bool]] = mapped_column(Boolean, default=False) + + # Relationships + job: Mapped["Job"] = relationship("Job", back_populates="transactions") diff --git a/app/repositories/job_repository.py b/app/repositories/job_repository.py index e40cb83..2262103 100644 --- a/app/repositories/job_repository.py +++ b/app/repositories/job_repository.py @@ -1,7 +1,10 @@ from typing import List, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload from app.db.models.job import Job +from app.db.models.transaction import Transaction +from app.db.models.job_summary import JobSummary class JobRepository: @@ -9,18 +12,25 @@ class JobRepository: self.db = db async def create(self, filename: str) -> Job: + """Create a new job record with status='pending'""" job = Job(filename=filename, status="pending") self.db.add(job) await self.db.flush() return job async def get_by_id(self, job_id: str) -> Optional[Job]: - query = select(Job).where(Job.id == job_id) + """Fetch a job record by its ID, eagerly loading summary and transactions""" + query = ( + select(Job) + .options(selectinload(Job.summary), selectinload(Job.transactions)) + .where(Job.id == job_id) + ) result = await self.db.execute(query) return result.scalar_one_or_none() async def list_all(self, status: Optional[str] = None) -> List[Job]: - query = select(Job) + """List all job records, optionally filtered by status, eagerly loading summary""" + query = select(Job).options(selectinload(Job.summary)) if status: query = query.where(Job.status == status) query = query.order_by(Job.created_at.desc()) @@ -28,8 +38,20 @@ class JobRepository: return list(result.scalars().all()) async def update(self, job: Job, **kwargs) -> Job: + """Update job fields dynamically""" for key, value in kwargs.items(): if hasattr(job, key): setattr(job, key, value) await self.db.flush() return job + + async def add_transactions(self, transactions: List[Transaction]) -> None: + """Add multiple transaction records in bulk""" + self.db.add_all(transactions) + await self.db.flush() + + async def add_summary(self, summary: JobSummary) -> JobSummary: + """Add job summary record""" + self.db.add(summary) + await self.db.flush() + return summary diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index 473c02d..2f80208 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -3,6 +3,8 @@ from app.schemas.job import ( JobResponse, JobStatusResponse, JobResultResponse, + TransactionResponse, + JobSummaryResponse, ) __all__ = [ @@ -10,4 +12,6 @@ __all__ = [ "JobResponse", "JobStatusResponse", "JobResultResponse", + "TransactionResponse", + "JobSummaryResponse", ] diff --git a/app/schemas/job.py b/app/schemas/job.py index 8677d84..b63a293 100644 --- a/app/schemas/job.py +++ b/app/schemas/job.py @@ -1,8 +1,41 @@ -from datetime import datetime -from typing import Any, Dict, Optional +from datetime import date, datetime +from typing import Any, Dict, List, Optional from pydantic import BaseModel, ConfigDict +class TransactionResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + job_id: str + txn_id: str + date: date + merchant: str + amount: float + currency: str + status: Optional[str] = None + category: Optional[str] = None + account_id: Optional[str] = None + is_anomaly: bool + anomaly_reason: Optional[str] = None + llm_category: Optional[str] = None + llm_raw_response: Optional[str] = None + llm_failed: Optional[bool] = False + + +class JobSummaryResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + job_id: str + total_spend_inr: float + total_spend_usd: float + top_merchants: Dict[str, Any] + anomaly_count: int + narrative: Optional[str] = None + risk_level: Optional[str] = None + + class JobBase(BaseModel): filename: str @@ -16,9 +49,11 @@ class JobResponse(JobBase): id: str status: str - row_count: Optional[int] = None + row_count_raw: Optional[int] = None + row_count_clean: Optional[int] = None created_at: datetime - updated_at: datetime + completed_at: Optional[datetime] = None + error_message: Optional[str] = None class JobStatusResponse(BaseModel): @@ -26,7 +61,7 @@ class JobStatusResponse(BaseModel): id: str status: str - summary: Optional[Dict[str, Any]] = None + summary: Optional[JobSummaryResponse] = None class JobResultResponse(BaseModel): @@ -34,4 +69,5 @@ class JobResultResponse(BaseModel): id: str status: str - results: Optional[Dict[str, Any]] = None + summary: Optional[JobSummaryResponse] = None + transactions: List[TransactionResponse] = [] diff --git a/app/utils/csv_parser.py b/app/utils/csv_parser.py index c5d93f6..02099c6 100644 --- a/app/utils/csv_parser.py +++ b/app/utils/csv_parser.py @@ -4,10 +4,15 @@ from datetime import datetime from typing import Any, Dict, List from app.core.exceptions import InvalidCSVException -REQUIRED_COLUMNS = {"transaction_id", "date", "amount", "category", "description"} +REQUIRED_COLUMNS = {"txn_id", "date", "merchant", "amount"} def parse_and_validate_csv(content: bytes) -> List[Dict[str, Any]]: + """ + Parses CSV content from bytes, validates headers, and checks field types. + Supports flexible mapping of common synonyms for headers. + Returns a list of dicts. Raises InvalidCSVException if the file is invalid. + """ try: decoded = content.decode("utf-8") except UnicodeDecodeError as e: @@ -16,32 +21,66 @@ def parse_and_validate_csv(content: bytes) -> List[Dict[str, Any]]: csv_file = io.StringIO(decoded) reader = csv.DictReader(csv_file) - # Map headers to predefined required cols + if not reader.fieldnames: + raise InvalidCSVException("CSV file is empty or missing headers") + header_mapping = {} for actual_header in reader.fieldnames: + if not actual_header: + continue cleaned = actual_header.strip().lower() - if cleaned in REQUIRED_COLUMNS: - header_mapping[cleaned] = actual_header - elif cleaned == "id" and "transaction_id" not in header_mapping: - header_mapping["transaction_id"] = actual_header + # Map synonyms + if cleaned in {"txn_id", "transaction_id", "id"}: + header_mapping["txn_id"] = actual_header + elif cleaned == "date": + header_mapping["date"] = actual_header + elif cleaned in {"merchant", "description"}: + header_mapping["merchant"] = actual_header + elif cleaned == "amount": + header_mapping["amount"] = actual_header + elif cleaned == "currency": + header_mapping["currency"] = actual_header + elif cleaned == "category": + header_mapping["category"] = actual_header + elif cleaned in {"account_id", "account"}: + header_mapping["account_id"] = actual_header + + # Verify all required columns are present in mapping missing = REQUIRED_COLUMNS - set(header_mapping.keys()) if missing: raise InvalidCSVException(f"Missing required CSV columns: {', '.join(missing)}") parsed_rows = [] for line_num, row in enumerate(reader, start=2): - t_id = row.get(header_mapping["transaction_id"]) + txn_id = row.get(header_mapping["txn_id"]) date_str = row.get(header_mapping["date"]) amount_str = row.get(header_mapping["amount"]) - category = row.get(header_mapping["category"]) - description = row.get(header_mapping["description"]) + merchant = row.get(header_mapping["merchant"]) - if not all([t_id, date_str, amount_str, category, description]): + # Optional columns + currency = ( + row.get(header_mapping.get("currency", "")) + if "currency" in header_mapping + else "INR" + ) + category = ( + row.get(header_mapping.get("category", "")) + if "category" in header_mapping + else None + ) + account_id = ( + row.get(header_mapping.get("account_id", "")) + if "account_id" in header_mapping + else None + ) + + if not all([txn_id, date_str, amount_str, merchant]): raise InvalidCSVException( f"Row {line_num}: Empty values are not allowed in required fields" ) + # Validate amount try: amount = float(amount_str.strip()) except ValueError as e: @@ -49,6 +88,7 @@ def parse_and_validate_csv(content: bytes) -> List[Dict[str, Any]]: f"Row {line_num}: Invalid amount '{amount_str}'. Must be a number." ) from e + # Validate date (supporting multiple formats) parsed_date = None for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d-%m-%Y", "%Y/%m/%d"): try: @@ -64,11 +104,13 @@ def parse_and_validate_csv(content: bytes) -> List[Dict[str, Any]]: parsed_rows.append( { - "transaction_id": t_id.strip(), + "txn_id": txn_id.strip(), "date": parsed_date.isoformat(), "amount": amount, - "category": category.strip(), - "description": description.strip(), + "merchant": merchant.strip(), + "currency": currency.strip() if currency else "INR", + "category": category.strip() if category else None, + "account_id": account_id.strip() if account_id else None, } ) diff --git a/app/worker.py b/app/worker.py index 1df9a6f..e579d2d 100644 --- a/app/worker.py +++ b/app/worker.py @@ -1,10 +1,17 @@ import asyncio +from datetime import datetime, date 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 +from app.db.models.transaction import Transaction +from app.db.models.job_summary import JobSummary +from app.clients.exchange_rate_client import ExchangeRateClient + +exchange_rate_client = ExchangeRateClient() + # Define Celery app celery_app = Celery( @@ -32,7 +39,7 @@ def run_async(coro): 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. + generate narrative summary, and persist results to Transaction and JobSummary tables. """ logger.info(f"Starting Celery task to process job_id: {job_id}") return run_async(_process_job_async(job_id, transactions)) @@ -51,103 +58,142 @@ async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): await repo.update(job, status="processing") await db.commit() - # 1. Clean transactions - cleaned_transactions = [] - total_amount = 0.0 - categories = {} + db_transactions = [] + total_spend_inr = 0.0 + total_spend_usd = 0.0 + merchant_spends: Dict[str, float] = {} + anomaly_count = 0 + + USD_TO_INR = await exchange_rate_client.get_usd_to_inr_rate() + 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, - } + txn_id = t["txn_id"] + raw_date = t["date"] + # Parse date string to datetime.date object + parsed_date = ( + date.fromisoformat(raw_date) + if isinstance(raw_date, str) + else raw_date ) + merchant = t["merchant"].strip() + amount = float(t["amount"]) + currency = t.get("currency", "INR").strip().upper() + category = t.get("category", "General") + account_id = t.get("account_id") - # 2. Flag anomalies (Heuristic: > $5,000, or containing suspicious keywords) - anomalies = [] - for t in cleaned_transactions: + # Spend calculation and currency conversion + if currency == "USD": + amt_usd = amount + amt_inr = amount * USD_TO_INR + else: + amt_inr = amount + amt_usd = amount / USD_TO_INR + + total_spend_inr += amt_inr + total_spend_usd += amt_usd + + # Aggregate merchant spend (in USD for standardization) + merchant_spends[merchant] = merchant_spends.get(merchant, 0.0) + amt_usd + + # Heuristic Anomaly detection is_anomaly = False - reasons = [] - - if t["amount"] > 5000.0: + anomaly_reason = None + if amt_usd > 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") + anomaly_reason = "High value transaction (> $5,000 USD equivalent)" + else: + desc_lower = merchant.lower() + if any( + k in desc_lower + for k in ["suspend", "hack", "error", "unknown", "fraud"] + ): + is_anomaly = True + anomaly_reason = "Suspicious merchant keyword" if is_anomaly: - anomalies.append({**t, "reasons": reasons}) + anomaly_count += 1 - # 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), - } + # Mock + llm_failed = False + llm_category = category.capitalize() if category else "General" + llm_raw_response = f'{{"category": "{llm_category}", "confidence": 0.95, "status": "processed"}}' - # 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 + db_txn = Transaction( + job_id=job_id, + txn_id=txn_id, + date=parsed_date, + merchant=merchant, + amount=amount, + currency=currency, + status="cleaned", + category=category, + account_id=account_id, + is_anomaly=is_anomaly, + anomaly_reason=anomaly_reason, + llm_category=llm_category, + llm_raw_response=llm_raw_response, + llm_failed=llm_failed, + ) + db_transactions.append(db_txn) + + await repo.add_transactions(db_transactions) + + top_merchant = ( + max(merchant_spends, key=merchant_spends.get) + if merchant_spends + else "None" ) + risk_level = "Low" + if anomaly_count > 2: + risk_level = "High" + elif anomaly_count > 0: + risk_level = "Medium" + 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." + f"Successfully parsed and clean-processed {len(transactions)} transaction records. " + f"Total spend is INR {total_spend_inr:,.2f} (${total_spend_usd:,.2f} USD). " + f"Top merchant by spend volume is '{top_merchant}' with total of ${merchant_spends.get(top_merchant, 0):,.2f} USD. " + f"Scan flagged {anomaly_count} potential anomalies, resulting in a overall risk level of '{risk_level}'." ) - # Build full structured output - results = { - "cleaned_transactions": cleaned_transactions, - "flagged_anomalies": anomalies, - "spend_breakdown": spend_breakdown, - "narrative_summary": narrative, + # Build top merchants dict sorted by spend limit (keep top 5) + sorted_merchants = sorted( + merchant_spends.items(), key=lambda x: x[1], reverse=True + )[:5] + top_merchants_json = { + merchant: round(spend, 2) for merchant, spend in sorted_merchants } - # 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, - } + # Create job summary + summary = JobSummary( + job_id=job_id, + total_spend_inr=round(total_spend_inr, 2), + total_spend_usd=round(total_spend_usd, 2), + top_merchants=top_merchants_json, + anomaly_count=anomaly_count, + narrative=narrative, + risk_level=risk_level, + ) + await repo.add_summary(summary) - # Update job in DB + # Update job state await repo.update( job, status="completed", - row_count=len(transactions), - summary=summary, - results=results, + row_count_raw=len(transactions), + row_count_clean=len(db_transactions), + completed_at=datetime.utcnow(), ) 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 repo.update( + job, + status="failed", + error_message=str(e), + completed_at=datetime.utcnow(), + ) await db.commit() raise e diff --git a/tests/api/test_jobs_api.py b/tests/api/test_jobs_api.py index e405ffb..82be0bd 100644 --- a/tests/api/test_jobs_api.py +++ b/tests/api/test_jobs_api.py @@ -12,9 +12,9 @@ celery_app.conf.task_always_eager = True async def test_upload_csv_success(client: AsyncClient): csv_content = ( - "transaction_id,date,amount,category,description\n" - "TX1001,2026-06-25,120.50,Groceries,Weekly grocery shop\n" - "TX1002,06/26/2026,15000.00,Electronics,New Laptop\n" # This will be flagged as anomaly (> $5,000) + "txn_id,date,amount,merchant,currency,category,account\n" + "TX1001,2026-06-25,120.50,Weekly grocery shop,INR,Groceries,ACC01\n" + "TX1002,06/26/2026,6000.00,New Laptop,USD,Electronics,ACC02\n" # Anomaly (> $5,000 USD equivalent) ) files = { @@ -43,10 +43,15 @@ async def test_upload_csv_success(client: AsyncClient): status_data = status_response.json() assert status_data["id"] == job_id assert status_data["status"] == "completed" - assert status_data["summary"]["total_rows"] == 2 - assert status_data["summary"]["total_amount"] == 15120.50 - assert status_data["summary"]["anomalies_count"] == 1 - assert status_data["summary"]["top_category"] == "Electronics" + + summary = status_data["summary"] + assert summary is not None + assert summary["anomaly_count"] == 1 + assert ( + summary["total_spend_usd"] > 6000.0 + ) # TX1002 is $6,000, TX1001 is INR 120.50 ($1.45) + assert "New Laptop" in summary["top_merchants"] + assert summary["risk_level"] == "Medium" # 1 anomaly # 3. Get Job Results results_response = await client.get(f"/api/v1/jobs/{job_id}/results") @@ -55,20 +60,24 @@ async def test_upload_csv_success(client: AsyncClient): assert results_data["id"] == job_id assert results_data["status"] == "completed" - results = results_data["results"] - assert len(results["cleaned_transactions"]) == 2 - assert len(results["flagged_anomalies"]) == 1 - 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"] + transactions_list = results_data["transactions"] + assert len(transactions_list) == 2 + + # Assert TX1002 is flagged as an anomaly + laptop_txn = next(t for t in transactions_list if t["txn_id"] == "TX1002") + assert laptop_txn["is_anomaly"] is True + assert "High value transaction" in laptop_txn["anomaly_reason"] + assert laptop_txn["llm_category"] == "Electronics" + assert laptop_txn["llm_failed"] is False + + grocery_txn = next(t for t in transactions_list if t["txn_id"] == "TX1001") + assert grocery_txn["is_anomaly"] is False + assert grocery_txn["currency"] == "INR" 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" - ) + # Invalid CSV structure (missing required column: merchant) + csv_content = "txn_id,date,amount\nTX1001,2026-06-25,120.50\n" files = { "file": ( "transactions.csv", diff --git a/tests/repositories/test_job_repository.py b/tests/repositories/test_job_repository.py index db2fff2..336064e 100644 --- a/tests/repositories/test_job_repository.py +++ b/tests/repositories/test_job_repository.py @@ -1,6 +1,9 @@ +from datetime import date import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.job_repository import JobRepository +from app.db.models.transaction import Transaction +from app.db.models.job_summary import JobSummary # Set up asyncio marker pytestmark = pytest.mark.asyncio @@ -22,36 +25,60 @@ async def test_create_and_get_job(db_session: AsyncSession): assert fetched_job.filename == "test_transactions.csv" -async def test_update_job(db_session: AsyncSession): +async def test_update_job_and_add_relations(db_session: AsyncSession): repo = JobRepository(db_session) job = await repo.create(filename="update_test.csv") # Update status and count updated = await repo.update( - job, status="completed", row_count=10, summary={"total": 100} + job, status="completed", row_count_raw=10, row_count_clean=8 ) assert updated.status == "completed" - assert updated.row_count == 10 - assert updated.summary == {"total": 100} + assert updated.row_count_raw == 10 + assert updated.row_count_clean == 8 + + # Create & add summary + summary = JobSummary( + job_id=job.id, + total_spend_inr=15000.0, + total_spend_usd=180.72, + top_merchants={"Merchant A": 100.0}, + anomaly_count=1, + narrative="Good", + risk_level="Low", + ) + await repo.add_summary(summary) + + # Create & add transaction + txn = Transaction( + job_id=job.id, + txn_id="TX001", + date=date(2026, 6, 25), + merchant="Merchant A", + amount=100.0, + currency="USD", + status="cleaned", + is_anomaly=False, + ) + await repo.add_transactions([txn]) # Retrieve to verify persistence fetched = await repo.get_by_id(job.id) assert fetched.status == "completed" - assert fetched.row_count == 10 + assert fetched.row_count_raw == 10 + assert fetched.row_count_clean == 8 + assert fetched.summary is not None + assert fetched.summary.total_spend_usd == 180.72 + assert len(fetched.transactions) == 1 + assert fetched.transactions[0].txn_id == "TX001" async def test_list_jobs(db_session: AsyncSession): repo = JobRepository(db_session) - # Clear existing if any await repo.create(filename="list_1.csv") await repo.create(filename="list_2.csv") jobs = await repo.list_all() + # At least the ones we created in this test session assert len(jobs) >= 2 - - pending_jobs = await repo.list_all(status="pending") - assert len(pending_jobs) >= 2 - - completed_jobs = await repo.list_all(status="completed") - assert len(completed_jobs) == 1 # From the update test above diff --git a/tests/services/test_csv_parser.py b/tests/services/test_csv_parser.py index 797a404..a714b24 100644 --- a/tests/services/test_csv_parser.py +++ b/tests/services/test_csv_parser.py @@ -5,23 +5,30 @@ from app.utils.csv_parser import parse_and_validate_csv def test_parse_valid_csv(): valid_csv = ( - "transaction_id,date,amount,category,description\n" - "TX1001,2026-06-25,120.50,Groceries,Weekly grocery shop\n" - "TX1002,06/26/2026,15.00,Coffee,Morning latte\n" + "transaction_id,date,amount,category,description,currency,account\n" + "TX1001,2026-06-25,120.50,Groceries,Weekly grocery shop,INR,ACC01\n" + "TX1002,06/26/2026,15.00,Coffee,Morning latte,USD,ACC02\n" ) result = parse_and_validate_csv(valid_csv.encode("utf-8")) assert len(result) == 2 - assert result[0]["transaction_id"] == "TX1001" + assert result[0]["txn_id"] == "TX1001" assert result[0]["amount"] == 120.50 assert result[0]["date"] == "2026-06-25" - assert result[1]["transaction_id"] == "TX1002" + assert result[0]["merchant"] == "Weekly grocery shop" + assert result[0]["currency"] == "INR" + assert result[0]["account_id"] == "ACC01" + + assert result[1]["txn_id"] == "TX1002" assert result[1]["amount"] == 15.00 assert result[1]["date"] == "2026-06-26" + assert result[1]["merchant"] == "Morning latte" + assert result[1]["currency"] == "USD" + assert result[1]["account_id"] == "ACC02" def test_parse_missing_headers(): invalid_csv = ( - "transaction_id,amount,category,description\n" + "txn_id,amount,category,description\n" "TX1001,120.50,Groceries,Weekly grocery shop\n" ) with pytest.raises(InvalidCSVException) as exc_info: @@ -31,8 +38,7 @@ def test_parse_missing_headers(): def test_parse_invalid_amount(): invalid_csv = ( - "transaction_id,date,amount,category,description\n" - "TX1001,2026-06-25,abc,Groceries,Weekly grocery shop\n" + "txn_id,date,amount,merchant\nTX1001,2026-06-25,abc,Weekly grocery shop\n" ) with pytest.raises(InvalidCSVException) as exc_info: parse_and_validate_csv(invalid_csv.encode("utf-8")) @@ -41,8 +47,7 @@ def test_parse_invalid_amount(): def test_parse_invalid_date(): invalid_csv = ( - "transaction_id,date,amount,category,description\n" - "TX1001,invalid-date,120.50,Groceries,Weekly grocery shop\n" + "txn_id,date,amount,merchant\nTX1001,invalid-date,120.50,Weekly grocery shop\n" ) with pytest.raises(InvalidCSVException) as exc_info: parse_and_validate_csv(invalid_csv.encode("utf-8")) diff --git a/tests/services/test_exchange_rate.py b/tests/services/test_exchange_rate.py new file mode 100644 index 0000000..d077d72 --- /dev/null +++ b/tests/services/test_exchange_rate.py @@ -0,0 +1,12 @@ +import pytest +from app.clients.exchange_rate_client import ExchangeRateClient + +pytestmark = pytest.mark.asyncio + + +async def test_get_usd_to_inr_rate(): + client = ExchangeRateClient() + rate = await client.get_usd_to_inr_rate() + assert isinstance(rate, float) + # The rate should be positive (either the fetched rate or the default fallback of 83.0) + assert rate > 0.0