VAKYA/Report_generation

1"""Tests for execution agent retries and email routing."""2 3from __future__ import annotations4 5import os6 7from server.database import ReportTrackingDB8from server.execution_agent import execute_report_job9 10 11def test_execute_success_sends_to_customer_email(tmp_path):12    os.environ["AGENT_EMAIL_MOCK"] = "1"13    db = ReportTrackingDB(str(tmp_path / "tracking.db"))14    db.seed_static_data()15    pairs = db.get_customer_lan_pairs()16    good = next(p for p in pairs if p["customer_code"] == "CUST-001")17 18    report_id = "job-success-001"19    db.create_live_track(20        report_id=report_id,21        batch_id="20260422101010999-10am",22        customer_id=int(good["customer_id"]),23        lan_id=int(good["lan_id"]),24        report_type="daily_morning_operational",25        report_format="pdf",26        scheduler_slot="10am",27    )28    result = execute_report_job(db, report_id)29    state = db.get_live_track(report_id)30 31    assert result["status"] == "success"32    assert result["email_recipient"] == "vakdevikankipati@gmail.com"33    assert state is not None34    assert state["status"] == "success"35    assert "success" in str(state["email_status"])36 37 38def test_execute_failure_retries_and_sends_fixed_email(tmp_path):39    os.environ["AGENT_EMAIL_MOCK"] = "1"40    db = ReportTrackingDB(str(tmp_path / "tracking.db"))41    db.seed_static_data()42    pairs = db.get_customer_lan_pairs()43    failing = next(p for p in pairs if p["lan_code"] == "LAN-005")44 45    report_id = "job-fail-001"46    db.create_live_track(47        report_id=report_id,48        batch_id="20260422101010999-10am",49        customer_id=int(failing["customer_id"]),50        lan_id=int(failing["lan_id"]),51        report_type="daily_morning_operational",52        report_format="pdf",53        scheduler_slot="10am",54    )55    result = execute_report_job(db, report_id)56    state = db.get_live_track(report_id)57 58    assert result["status"] == "failed"59    assert result["attempts"] == 560    assert result["email_recipient"] == "vakdevikankipati@gmail.com"61    assert state is not None62    assert state["status"] == "failed"63    assert state["retries_used"] == 564    assert "failure" in str(state["email_status"])65