98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
import sys
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# Add backend to path
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from sqlmodel import Session, select
|
|
from database import engine
|
|
from models import User, Vertical, UserVerticalPreference, Show, Venue, Notification, NotificationType, PreferenceTier
|
|
from services.notification_service import NotificationService
|
|
|
|
def verify_notifications():
|
|
print(f"DEBUG: Using database URL: {engine.url}")
|
|
with Session(engine) as session:
|
|
print("Setting up test data...")
|
|
|
|
# 1. Get or create a user
|
|
user = session.exec(select(User).where(User.email == "test_notify@example.com")).first()
|
|
if not user:
|
|
user = User(email="test_notify@example.com", hashed_password="hashed_password", is_active=True)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
print(f"User ID: {user.id}")
|
|
|
|
# 2. Get a vertical (Phish or Goose)
|
|
vertical = session.exec(select(Vertical).where(Vertical.slug == "phish")).first()
|
|
if not vertical:
|
|
print("Phish vertical not found, creating dummy...")
|
|
vertical = Vertical(slug="phish", name="Phish")
|
|
session.add(vertical)
|
|
session.commit()
|
|
session.refresh(vertical)
|
|
print(f"Vertical ID: {vertical.id}")
|
|
|
|
# 3. Create/Update preference
|
|
pref = session.exec(
|
|
select(UserVerticalPreference)
|
|
.where(UserVerticalPreference.user_id == user.id)
|
|
.where(UserVerticalPreference.vertical_id == vertical.id)
|
|
).first()
|
|
|
|
if not pref:
|
|
pref = UserVerticalPreference(
|
|
user_id=user.id,
|
|
vertical_id=vertical.id,
|
|
tier=PreferenceTier.HEADLINER,
|
|
notify_on_show=True
|
|
)
|
|
session.add(pref)
|
|
else:
|
|
pref.notify_on_show = True
|
|
session.add(pref)
|
|
session.commit()
|
|
print("User preference set to notify_on_show=True")
|
|
|
|
# 4. Create a Venue
|
|
venue = session.exec(select(Venue).where(Venue.name == "Test Venue")).first()
|
|
if not venue:
|
|
venue = Venue(name="Test Venue", city="Test City", country="USA")
|
|
session.add(venue)
|
|
session.commit()
|
|
session.refresh(venue)
|
|
|
|
# 5. Create a Show using Service logic (simulate API call)
|
|
# We invoke NotificationService manually on a new show object
|
|
print("Creating new show...")
|
|
new_show = Show(
|
|
date=datetime.now(),
|
|
slug=f"phish-test-{int(datetime.now().timestamp())}",
|
|
vertical_id=vertical.id,
|
|
venue_id=venue.id
|
|
)
|
|
session.add(new_show)
|
|
session.commit()
|
|
session.refresh(new_show)
|
|
|
|
service = NotificationService(session)
|
|
service.check_show_alert(new_show)
|
|
|
|
# 6. Verify Notification
|
|
print("Checking for notification...")
|
|
notes = service.get_user_notifications(user.id)
|
|
found = False
|
|
for n in notes:
|
|
if n.type == NotificationType.SHOW_ALERT and n.link == f"/{vertical.slug}/shows/{new_show.slug}":
|
|
print(f"✅ Notification found: {n.title} - {n.message}")
|
|
found = True
|
|
break
|
|
|
|
if not found:
|
|
print("❌ Notification NOT found!")
|
|
|
|
# Clean up test data if needed, but keeping for debug might be fine
|
|
|
|
if __name__ == "__main__":
|
|
verify_notifications()
|