VAKYA/Report_generation

1"""Tests for customer/lan mapping and live report tracking."""2 3from __future__ import annotations4 5from server.database import ReportTrackingDB6 7 8def test_seed_mapping_and_live_tracking(tmp_path):9    db_file = tmp_path / "tracking.db"10    db = ReportTrackingDB(str(db_file))11    db.seed_static_data()12 13    pairs = db.get_customer_lan_pairs()14    assert len(pairs) == 515    first = pairs[0]16 17    report_id = "rep-10am-001"18    db.create_live_track(19        report_id=report_id,20        batch_id="20260422101010999-10am",21        customer_id=int(first["customer_id"]),22        lan_id=int(first["lan_id"]),23        report_type="daily_sales_summary",24        report_format="pdf",25        scheduler_slot="10am",26    )27 28    db.update_live_track_status(29        report_id,30        status="in_progress",31        retries_used=1,32        report_generated=False,33    )34    db.update_live_track_status(35        report_id,36        status="retrying",37        retries_used=2,38        error_code="lan_unreachable",39        error_message="LAN unreachable",40        report_generated=False,41    )42    db.update_live_track_status(43        report_id,44        status="failed",45        retries_used=5,46        error_code="lan_permanent_failure",47        error_message="Permanent failure after retries",48        report_generated=False,49        finished=True,50    )51    db.update_email_status(report_id, sent=True, status="failure_sent")52 53    track = db.get_live_track(report_id)54    all_tracks = db.list_live_tracks()55 56    assert track is not None57    assert track["status"] == "failed"58    assert track["retries_used"] == 559    assert track["last_error_code"] == "lan_permanent_failure"60    assert track["email_sent"] == 161    assert track["email_status"] == "failure_sent"62    assert track["finished_at_utc"] is not None63 64    assert len(all_tracks) == 165    assert all_tracks[0]["customer_code"].startswith("CUST-")66 67 68def test_email_routing_rules(tmp_path):69    db_file = tmp_path / "tracking.db"70    db = ReportTrackingDB(str(db_file))71    db.seed_static_data()72    pairs = db.get_customer_lan_pairs()73 74    good = next(p for p in pairs if p["customer_code"] == "CUST-001")75    blocked = next(p for p in pairs if p["customer_code"] == "CUST-005")76 77    ok_report = "rep-ok-001"78    db.create_live_track(79        report_id=ok_report,80        batch_id="20260422101010999-10am",81        customer_id=int(good["customer_id"]),82        lan_id=int(good["lan_id"]),83        report_type="daily_morning_operational",84        report_format="pdf",85        scheduler_slot="10am",86    )87 88    blocked_report = "rep-block-001"89    db.create_live_track(90        report_id=blocked_report,91        batch_id="20260422101010999-10am",92        customer_id=int(blocked["customer_id"]),93        lan_id=int(blocked["lan_id"]),94        report_type="daily_morning_operational",95        report_format="pdf",96        scheduler_slot="10am",97    )98 99    # Success path -> mapped customer email.100    assert db.resolve_recipient_email(ok_report, is_failure=False) == "vakdevikankipati@gmail.com"101 102    # Failure path -> fixed escalation email only.103    assert db.resolve_recipient_email(ok_report, is_failure=True) == "vakdevikankipati@gmail.com"104 105    # Blocked customer email -> reroute to fixed email.106    assert db.resolve_recipient_email(blocked_report, is_failure=False) == "vakdevikankipati@gmail.com"107