Spaces:
Running
Running
| """ | |
| 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()) | |