""" check_notification_template.py -- offline: notification rows carry the lead's category and who added / last touched it (2026-09-28, user: "lead name category or added by and last activity by show ho"). uv run python scripts/check_notification_template.py No database, no network -- throwaway in-memory SQLite. Covers crud.list_for_user(): * category / added_by_name / last_activity_by_name resolved correctly for a lead with both set * blank (not an error) when a lead has no creator/toucher recorded (pre-login data) * a Problem Desk mention (no lead_id) leaves all three blank, company_name too """ from __future__ import annotations import sys 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 leads as crud_leads # noqa: E402 from app.crud import notifications as crud # noqa: E402 from app.database import Base # noqa: E402 from app.models import Notification, SalesProblem, User, _utcnow # 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)() ali = User(email="ali@elipsestudio.com", name="Ali", is_active=True) bilal = User(email="bilal@elipsestudio.com", name="Bilal", is_active=True) recipient = User(email="me@elipsestudio.com", name="Me", is_active=True) db.add_all([ali, bilal, recipient]) db.commit() lead = crud_leads.add_lead_dict(db, {"company_name": "Acme Yachts", "industry_tag": "⛵ Superyachts & Marine"}) lead.created_by_user_id = ali.id lead.last_touched_by_user_id = bilal.id db.commit() old_lead = crud_leads.add_lead_dict(db, {"company_name": "Legacy Co"}) # no creator/toucher recorded db.commit() problem = SalesProblem(title="Client wants a refund", priority="High Priority", status="Open", created_at=_utcnow()) db.add(problem) db.commit() db.add_all([ Notification(user_id=recipient.id, actor_user_id=ali.id, kind="mention", lead_id=lead.id, excerpt="check this", created_at=_utcnow()), Notification(user_id=recipient.id, actor_user_id=bilal.id, kind="mention", lead_id=old_lead.id, excerpt="hi", created_at=_utcnow()), Notification(user_id=recipient.id, actor_user_id=ali.id, kind="mention", problem_id=problem.id, excerpt="look", created_at=_utcnow()), ]) db.commit() rows = {r["n"].lead_id or f"problem-{r['n'].problem_id}": r for r in crud.list_for_user(db, recipient.id)} print("\nLead with a recorded creator/toucher") r = rows[lead.id] check("category resolved", r["category"] == "⛵ Superyachts & Marine", r["category"]) check("added_by_name resolved", r["added_by_name"] == "Ali", r["added_by_name"]) check("last_activity_by_name resolved", r["last_activity_by_name"] == "Bilal", r["last_activity_by_name"]) print("\nLead with no creator/toucher recorded (pre-login data)") r2 = rows[old_lead.id] check("added_by_name blank, not an error", r2["added_by_name"] == "", repr(r2["added_by_name"])) check("last_activity_by_name blank", r2["last_activity_by_name"] == "", repr(r2["last_activity_by_name"])) print("\nProblem Desk mention (no lead at all)") r3 = rows[f"problem-{problem.id}"] check("company_name blank", r3["company_name"] == "") check("category blank", r3["category"] == "") check("added_by_name blank", r3["added_by_name"] == "") print("\nFAILED: %d" % FAILED) return 1 if FAILED else 0 if __name__ == "__main__": sys.exit(main())