""" check_callback_attempts.py -- offline: the 3-checkbox re-contact tracker (crud/callback_attempts.py, 2026-09-29). No database, no network -- throwaway in-memory SQLite. uv run python scripts/check_callback_attempts.py Covers: * record_attempt() appends a UTC timestamp per call, caps at 3 (a 4th call is a no-op) * undo_last_attempt() pops the most recently checked box, is a no-op on an empty list * next_attempt_due() is None with 0 or 3 attempts recorded, and gives the right gap otherwise (48h after the 1st, 72h after the 2nd) * the weekend push: a due date that would land on Saturday or Sunday (PKT) rolls forward to Monday, same time -- checked by picking a 1st-attempt timestamp that lands the 48h-later due date ON a Saturday with NO push, to prove the push is what moves it * the three contexts (callback / no_show / proposal) are genuinely independent -- checking one does not touch the other two on the same lead * LeadOut.from_model() serializes the raw history with a "Z" suffix (matching every other backend-generated timestamp) and exposes the same due date computed above """ from __future__ import annotations import sys from datetime import datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from sqlalchemy import create_engine # noqa: E402 from sqlalchemy.orm import sessionmaker # noqa: E402 from sqlalchemy.pool import StaticPool # noqa: E402 import app.models # noqa: E402,F401 from app.crud import callback_attempts as crud # noqa: E402 from app.crud import leads as crud_leads # noqa: E402 from app.database import Base # noqa: E402 from app.schemas.lead import LeadOut # noqa: E402 FAILED = 0 def check(label: str, ok: bool, detail: str = "") -> None: global FAILED FAILED += 0 if ok else 1 print(f" [{'ok ' if ok else 'FAIL'}] {label}" + ("" if ok else f" {detail}")) def main() -> int: engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) db = sessionmaker(bind=engine)() lead = crud_leads.add_lead_dict(db, {"company_name": "Ancient Cherry Yachts"}) print("\nrecord_attempt: appends, caps at 3") check("starts empty", (lead.callback_attempts or []) == []) crud.record_attempt(db, lead, "callback") check("1 recorded", len(lead.callback_attempts) == 1) crud.record_attempt(db, lead, "callback") crud.record_attempt(db, lead, "callback") check("3 recorded", len(lead.callback_attempts) == 3) third = list(lead.callback_attempts) crud.record_attempt(db, lead, "callback") check("4th call is a no-op, still 3", lead.callback_attempts == third) print("\nundo_last_attempt: pops the most recent, no-op on empty") crud.undo_last_attempt(db, lead, "callback") check("back to 2 after undo", len(lead.callback_attempts) == 2) crud.undo_last_attempt(db, lead, "callback") crud.undo_last_attempt(db, lead, "callback") check("back to 0 after 2 more undos", lead.callback_attempts == []) crud.undo_last_attempt(db, lead, "callback") check("undo on empty is a no-op, no crash", lead.callback_attempts == []) print("\ncontexts are independent") lead2 = crud_leads.add_lead_dict(db, {"company_name": "Raspy Darkness Marine"}) crud.record_attempt(db, lead2, "callback") crud.record_attempt(db, lead2, "no_show") crud.record_attempt(db, lead2, "no_show") check("callback has 1", len(lead2.callback_attempts) == 1) check("no_show has 2, not shared with callback's count", len(lead2.no_show_attempts) == 2) check("proposal untouched", (lead2.proposal_attempts or []) == []) print("\nnext_attempt_due: gap math") check("0 attempts -> None (nothing to schedule yet)", crud.next_attempt_due([]) is None) check("3 attempts -> None (confirm point, not a reschedule point)", crud.next_attempt_due(["a", "b", "c"]) is None) # A Wednesday timestamp: 48h later is Friday -- no weekend push needed, isolates the gap math itself # from the push logic (checked separately below). wed = datetime(2026, 9, 30, 10, 0, 0) # 2026-09-30 is a Wednesday due_after_1 = crud.next_attempt_due([wed.isoformat()]) check( "after 1 attempt, due is +48h (no weekend in the way)", due_after_1 == (wed + timedelta(hours=48)).isoformat(), str(due_after_1), ) due_after_2 = crud.next_attempt_due([wed.isoformat(), (wed + timedelta(hours=48)).isoformat()]) check( "after 2 attempts, due is +72h from the 2nd (not the 1st)", due_after_2 == (wed + timedelta(hours=48 + 72)).isoformat(), str(due_after_2), ) print("\nweekend push: a due date landing on Sat/Sun (PKT) rolls to Monday same time") # Thursday 2026-10-01, 10:00 PKT +48h (raw, no push) = Saturday 2026-10-03, 10:00 PKT -- lands ON a # weekend with NO push, so this specific timestamp proves the push actually moved it, not that it # happened to land on a weekday anyway. Stored as naive UTC (matching _utcnow()), so PKT = UTC + 5h -- # pick the UTC instant whose PKT wall-clock is Thursday 10:00 (i.e. UTC 05:00). thu_utc = datetime(2026, 10, 1, 5, 0, 0) # PKT: Thu 2026-10-01 10:00 raw_due = thu_utc + timedelta(hours=48) # PKT: Sat 2026-10-03 10:00 -- the weekend case check( "(context) the raw +48h WOULD land on a Saturday in PKT, proving a push is actually needed here", (raw_due + timedelta(hours=5)).weekday() == 5, str((raw_due + timedelta(hours=5)).weekday()), ) pushed = crud.next_attempt_due([thu_utc.isoformat()]) pushed_dt = datetime.fromisoformat(pushed) pushed_pkt_weekday = (pushed_dt + timedelta(hours=5)).weekday() check("pushed date is no longer a weekend day in PKT", pushed_pkt_weekday < 5, str(pushed_pkt_weekday)) check( "pushed to the following Monday, same wall-clock time (not shifted)", pushed_dt == raw_due + timedelta(days=2) and pushed_dt.time() == raw_due.time(), str(pushed_dt), ) print("\nLeadOut.from_model: 'Z' suffix, due date exposed") lead3 = crud_leads.add_lead_dict(db, {"company_name": "Z Suffix Check"}) crud.record_attempt(db, lead3, "proposal") out = LeadOut.from_model(lead3) check("proposal_attempts has exactly 1 entry, 'Z'-suffixed", len(out.proposal_attempts) == 1 and out.proposal_attempts[0].endswith("Z")) check("proposal_next_attempt_due is set and 'Z'-suffixed", out.proposal_next_attempt_due.endswith("Z"), out.proposal_next_attempt_due) check("callback_attempts empty on this lead", out.callback_attempts == []) check("callback_next_attempt_due empty (0 attempts)", out.callback_next_attempt_due == "") print("\nFAILED: %d" % FAILED) return 1 if FAILED else 0 if __name__ == "__main__": sys.exit(main())