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