"""Phase 1 smoke test. Runs against SQLite in-memory; no MySQL/Redis needed. python -m tests.test_smoke (or) pytest -q Production runs over HTTPS (PREFERRED_URL_SCHEME=https), so Flask-WTF enforces the secure-referrer CSRF check. The client helper below sends an https base_url and a matching Referer, exactly as a real browser would. """ import os import re import io import logging import tempfile os.environ.setdefault("SECRET_KEY", "test-secret") # File-based SQLite: in-memory sqlite:// gives each connection its own empty DB # under Flask-SQLAlchemy, so use a temp file that all connections share. _SMOKE_DB = os.path.join(tempfile.gettempdir(), "classifieds_smoke.db") if os.path.exists(_SMOKE_DB): os.remove(_SMOKE_DB) os.environ.setdefault("DATABASE_URL", f"sqlite:///{_SMOKE_DB}") os.environ.setdefault("REDIS_URL", "memory://") # limiter in-memory os.environ.setdefault("FLASK_CONFIG", "dev") from app import create_app # noqa: E402 from app.extensions import db # noqa: E402 from app.models.plan import Plan # noqa: E402 from app.models.user import User # noqa: E402 from app.utils.text import normalize # noqa: E402 from seed import seed_plans, seed_categories, seed_zip_sample # noqa: E402 BASE = "https://localhost" def _csrf(html): return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', html).group(1) def _phase3(app): """Phase 3: messaging, favorites, contact masking.""" from app.models.user import User from app.models.plan import Plan from app.models.category import Category from app.models.listing import Listing from app.models.messaging import Conversation, Message from app.models.enums import TrustTier from app.services import messaging as msvc from app.services import favorites as fsvc from app.services.contact import mask_body, contact_revealed, contact_density with app.app_context(): # create a second user (buyer) buyer = User(email="buyer@example.com", display_name="Buyer", email_verified=True) buyer.set_password("BuyerPass123") buyer.tier_id = Plan.query.filter_by(slug="free").first().id buyer.trust_tier = TrustTier.trusted # can see contact info db.session.add(buyer); db.session.commit() seller = User.query.filter_by(email="t@example.com").first() fs = Category.query.filter_by(slug="for-sale").first() listing = Listing.query.filter_by(user_id=seller.id).first() # --- conversation create --- conv, created = msvc.get_or_create_conversation(listing, buyer) assert created and conv.buyer_id == buyer.id assert conv.seller_id == seller.id print("conversation created: ok") # idempotent: get same conv again conv2, created2 = msvc.get_or_create_conversation(listing, buyer) assert conv2.id == conv.id and not created2 print("conversation idempotent: ok") # seller can't message own listing try: msvc.get_or_create_conversation(listing, seller) assert False, "should have raised" except msvc.MessagingError: pass print("self-message blocked: ok") # --- send message --- msg1 = msvc.send_message(conv, buyer, "Hi, is this still available?") assert msg1.sender_id == buyer.id and msg1.read_at is None assert conv.last_message_at is not None print("send message: ok") # seller replies msg2 = msvc.send_message(conv, seller, "Yes, come pick it up tomorrow.") assert len(conv.messages) == 2 print("reply: ok") # --- unread count --- # buyer: 1 unread (seller's reply) # seller: 1 unread (buyer's opening message) unread_buyer = conv.unread_count(buyer) unread_seller = conv.unread_count(seller) assert unread_buyer == 1, unread_buyer assert unread_seller == 1, unread_seller print(f"unread count: ok (buyer={unread_buyer}, seller={unread_seller})") # --- mark read --- msvc.mark_conversation_read(conv, buyer) assert conv.unread_count(buyer) == 0 print("mark read: ok") # --- total unread --- total = msvc.total_unread(seller) assert total >= 1 print(f"total unread: ok (seller sees {total})") # --- inbox --- page = msvc.inbox(buyer) assert any(c.id == conv.id for c in page.items) print("inbox: ok") # --- contact masking --- spammy = "Call me at 703-555-0192 or email me@test.com to buy!" assert contact_density(spammy) == 2 masked = mask_body(spammy, reveal=False) assert "703" not in masked and "me@test.com" not in masked revealed = mask_body(spammy, reveal=True) assert "703-555-0192" in revealed print("contact masking: ok") # trusted user can reveal; new-trust cannot assert contact_revealed(buyer) # trust_tier=trusted + email_verified seller.trust_tier = TrustTier.new db.session.commit() assert not contact_revealed(seller) print("contact reveal gating by trust tier: ok") # --- favorites --- now_fav = fsvc.toggle_favorite(buyer.id, listing.id) assert now_fav assert fsvc.is_favorited(buyer.id, listing.id) page = fsvc.user_favorites(buyer.id) assert any(f.listing_id == listing.id for f in page.items) print("favorite add: ok") now_fav = fsvc.toggle_favorite(buyer.id, listing.id) assert not now_fav assert not fsvc.is_favorited(buyer.id, listing.id) print("favorite remove: ok") # --- empty body rejected --- try: msvc.send_message(conv, buyer, " ") assert False except msvc.MessagingError: pass print("empty message rejected: ok") # --- route render checks --- import re c = app.test_client(); B = "https://localhost" def csrf(h): return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', h).group(1) # login as buyer r = c.get("/auth/login", base_url=B) tok = csrf(r.get_data(as_text=True)) c.post("/auth/login", base_url=B, data={"csrf_token": tok, "email": "buyer@example.com", "password": "BuyerPass123"}, headers={"Referer": B + "/auth/login"}, follow_redirects=True) with app.app_context(): conv_id = Conversation.query.filter_by( buyer_id=User.query.filter_by(email="buyer@example.com").first().id ).first().id listing_id = Listing.query.first().id for path, expect in [ ("/messages", 200), (f"/messages/{conv_id}", 200), ("/my/favorites", 200), (f"/listings/{listing_id}/contact", 200), ]: code = c.get(path, base_url=B).status_code assert code == expect, f"{path} -> {code}" print(f"{code} {path}") print("Phase 3 route renders: ok") def _phase4(app): """Phase 4: payments — models, boost activation, expiry, tier sync, webhook handler (mocked), route renders. No live Stripe calls; tests the logic layer with mock data. """ from datetime import datetime, timedelta from app.models.user import User from app.models.plan import Plan from app.models.listing import Listing from app.models.payments import Subscription, Transaction, Boost from app.models.enums import Role from app.services import billing as bsvc with app.app_context(): user = User.query.filter_by(email="t@example.com").first() listing = Listing.query.filter_by(user_id=user.id).first() pro_plan = Plan.query.filter_by(slug="pro").first() # --- boost activation (no Stripe call) --- boost = bsvc.activate_boost( user_id=user.id, listing_id=listing.id, boost_type="featured", stripe_payment_intent_id="pi_test_001", amount_cents=499, ) assert boost.is_active assert boost.type == "featured" listing_fresh = db.session.get(Listing, listing.id) assert listing_fresh.is_featured print("boost activation: ok") # --- idempotency: same payment intent → no duplicate --- boost2 = bsvc.activate_boost( user_id=user.id, listing_id=listing.id, boost_type="featured", stripe_payment_intent_id="pi_test_001", amount_cents=499, ) assert Boost.query.filter_by(listing_id=listing.id, type="featured").count() == 1 print("boost idempotency: ok") # --- second boost type (bump) --- bump = bsvc.activate_boost( user_id=user.id, listing_id=listing.id, boost_type="bump", stripe_payment_intent_id="pi_test_002", amount_cents=199, ) assert bump.type == "bump" assert listing_fresh.bump_at is not None print("bump boost: ok") # --- boost expiry sweep --- boost.expires_at = datetime.utcnow() - timedelta(hours=1) db.session.commit() n = bsvc.expire_boosts() assert n >= 1 listing_after = db.session.get(Listing, listing.id) assert not listing_after.is_featured print(f"boost expiry sweep: ok (cleared {n})") # --- sync_subscription: simulate webhook updating local DB --- # Build a fake Stripe subscription object fake_stripe_sub = { "id": "sub_test_001", "customer": "cus_test_001", "status": "active", "cancel_at_period_end": False, "current_period_end": int( (datetime.utcnow() + timedelta(days=30)).timestamp()), "items": {"data": [{"price": {"id": pro_plan.stripe_price_id or "price_test_pro"}}]}, } # temporarily set stripe_price_id so sync works original_price_id = pro_plan.stripe_price_id pro_plan.stripe_price_id = "price_test_pro" fake_stripe_sub["items"]["data"][0]["price"]["id"] = "price_test_pro" db.session.commit() sub = bsvc.sync_subscription(user.id, fake_stripe_sub) assert sub.stripe_sub_id == "sub_test_001" assert sub.status == "active" user_fresh = db.session.get(User, user.id) assert user_fresh.role == Role.subscriber assert user_fresh.tier.slug == "pro" print("sync_subscription → user upgraded to pro subscriber: ok") # --- downgrade to free --- bsvc.downgrade_to_free(user.id) user_fresh = db.session.get(User, user.id) assert user_fresh.role == Role.free assert user_fresh.tier.slug == "free" sub_fresh = Subscription.query.filter_by(user_id=user.id).first() assert sub_fresh.status == "canceled" print("downgrade_to_free: ok") # restore pro_plan.stripe_price_id = original_price_id db.session.commit() # --- Transaction record exists --- txns = Transaction.query.filter_by(user_id=user.id).all() assert len(txns) >= 2 # featured + bump boosts print(f"transactions recorded: ok ({len(txns)} found)") # --- route render checks --- import re c = app.test_client(); B = "https://localhost" def csrf(h): return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', h).group(1) # login r = c.get("/auth/login", base_url=B) tok = csrf(r.get_data(as_text=True)) c.post("/auth/login", base_url=B, data={"csrf_token": tok, "email": "t@example.com", "password": "NewPass456"}, headers={"Referer": B + "/auth/login"}, follow_redirects=True) with app.app_context(): lid = Listing.query.filter_by( user_id=User.query.filter_by(email="t@example.com").first().id ).first().id for path, expect in [ ("/pricing", 200), ("/my/billing", 200), (f"/listings/{lid}/boost", 200), ]: code = c.get(path, base_url=B).status_code assert code == expect, f"{path} -> {code}" print(f"{code} {path}") print("Phase 4 route renders: ok") def _phase5(app): """Phase 5: ads, sponsors, promoted search.""" from datetime import datetime, timedelta from app.models.ads import Ad, Sponsor, PromotedKeyword from app.models.listing import Listing from app.services.ads import (get_ad, record_impression, record_click, promoted_listings, active_sponsors, expire_promoted_keywords) with app.app_context(): # --- create test ad --- now = datetime.utcnow() ad = Ad( advertiser_name="Pho 99 Restaurant", slot="sidebar", target_url="https://pho99.example.com", alt_text="Best pho in town", lang="vi", starts_at=now - timedelta(hours=1), ends_at=now + timedelta(days=30), is_active=True, ) db.session.add(ad); db.session.commit() print(f"ad created: id={ad.id} slot={ad.slot}") # untargeted ad for fallback ad2 = Ad( advertiser_name="Generic Ads Inc", slot="header", target_url="https://generic.example.com", starts_at=now - timedelta(hours=1), ends_at=now + timedelta(days=30), is_active=True, ) db.session.add(ad2); db.session.commit() # --- get_ad: targeted hit --- found = get_ad("sidebar", lang="vi") assert found and found.id == ad.id print("get_ad targeted: ok") # --- get_ad: slot with no match → None --- miss = get_ad("footer", lang="vi", state="CA") assert miss is None print("get_ad miss: ok") # --- get_ad: untargeted fallback --- h = get_ad("header") assert h and h.id == ad2.id print("get_ad untargeted fallback: ok") # --- impression tracking --- before = ad.impressions record_impression(ad.id) db.session.expire(ad) assert ad.impressions == before + 1 print("impression tracking: ok") # --- click tracking --- before_c = ad.clicks record_click(ad.id) db.session.expire(ad) assert ad.clicks == before_c + 1 print("click tracking: ok") # --- CTR --- assert ad.ctr > 0 print(f"CTR: {ad.ctr}%: ok") # --- sponsored ad suppressed for subscriber --- # (tested via plan limit — ad_free=True for basic+) from app.models.plan import Plan basic = Plan.query.filter_by(slug="basic").first() assert basic.limit("ad_free") is True print("ad_free plan limit: ok") # --- sponsor directory --- sp = Sponsor( name="Little Saigon Weekly", url="https://lsweekly.example.com", tagline="Your community newspaper", tier="directory", starts_at=now - timedelta(hours=1), ends_at=now + timedelta(days=60), is_active=True, ) db.session.add(sp); db.session.commit() sponsors = active_sponsors(tier="directory") assert any(s.id == sp.id for s in sponsors) print("sponsor directory: ok") # --- category sponsor --- from app.models.category import Category jobs_cat = Category.query.filter_by(slug="jobs").first() cat_sp = Sponsor( name="VN Jobs Network", url="https://vnjobs.example.com", tier="category", category_id=jobs_cat.id, starts_at=now - timedelta(hours=1), ends_at=now + timedelta(days=60), is_active=True, ) db.session.add(cat_sp); db.session.commit() cat_sponsors = active_sponsors(tier="category", category_id=jobs_cat.id) assert any(s.id == cat_sp.id for s in cat_sponsors) print("category sponsor: ok") # --- promoted search --- listing = Listing.query.first() pk = PromotedKeyword( keyword="pho", listing_id=listing.id, priority=10, expires_at=now + timedelta(days=7), ) db.session.add(pk); db.session.commit() results = promoted_listings("pho") assert listing in results print("promoted search (keyword match): ok") # accent-insensitive: "phở" should match keyword "pho" results2 = promoted_listings("phở") assert listing in results2 print("promoted search (accent-insensitive): ok") # no match for unrelated keyword results3 = promoted_listings("sofa") assert listing not in results3 print("promoted search (no match): ok") # --- expire promoted keywords --- pk.expires_at = now - timedelta(hours=1) db.session.commit() n = expire_promoted_keywords() assert n == 1 print(f"expire promoted keywords: ok (removed {n})") # --- route render checks --- import re c = app.test_client(); B = "https://localhost" for path, expect in [ ("/sponsors", 200), ("/ads/1/click", 302), # redirects to target_url ]: code = c.get(path, base_url=B).status_code assert code == expect, f"{path} -> {code}" print(f"{code} {path}") print("Phase 5 route renders: ok") def run(): app = create_app() with app.app_context(): db.create_all() seed_plans() seed_categories() seed_zip_sample() plans = [(p.slug, p.limit("active_listings"), p.limit("ad_free")) for p in Plan.query.order_by(Plan.sort_order)] assert plans == [("free", 3, False), ("basic", 15, True), ("pro", 50, True), ("business", None, True)], plans print("plans seeded:", plans) assert normalize("Phở Bò Đặc Biệt") == "pho bo dac biet" assert normalize("Ñandú Jalapeño") == "nandu jalapeno" print("accent normalizer ok") c = app.test_client() buf = io.StringIO() h = logging.StreamHandler(buf); h.setLevel(logging.INFO) app.logger.addHandler(h) def post(path, data, ref=None): return c.post(path, base_url=BASE, data=data, follow_redirects=True, headers={"Referer": ref or f"{BASE}{path}"}) def get(path): return c.get(path, base_url=BASE, follow_redirects=True) # register tok = _csrf(get("/auth/register").get_data(as_text=True)) r = post("/auth/register", {"csrf_token": tok, "display_name": "Test User", "email": "t@example.com", "password": "StrongPass123", "confirm": "StrongPass123"}) assert r.status_code == 200 and "verify" in r.get_data(as_text=True).lower() print("register + verify email: ok") # verify email link link = re.search(r"/auth/verify/(\S+)", buf.getvalue()).group(1) r = get("/auth/verify/" + link) assert "verified" in r.get_data(as_text=True).lower() with app.app_context(): u = User.query.filter_by(email="t@example.com").first() assert u.email_verified and u.trust_score == 5 and u.tier.slug == "free" print(f"verify: ok (trust={u.trust_score}, tier={u.tier.slug})") # login tok = _csrf(get("/auth/login").get_data(as_text=True)) r = post("/auth/login", {"csrf_token": tok, "email": "t@example.com", "password": "StrongPass123"}) assert "Test User" in r.get_data(as_text=True) print("login: ok") # language switch assert get("/lang/vi").status_code in (302, 200) print("lang switch: ok") # bad password rejected get("/auth/logout") tok = _csrf(get("/auth/login").get_data(as_text=True)) r = post("/auth/login", {"csrf_token": tok, "email": "t@example.com", "password": "wrong"}) assert "invalid" in r.get_data(as_text=True).lower() print("bad login rejected: ok") # duplicate email rejected tok = _csrf(get("/auth/register").get_data(as_text=True)) r = post("/auth/register", {"csrf_token": tok, "display_name": "Dupe", "email": "t@example.com", "password": "StrongPass123", "confirm": "StrongPass123"}) assert "already exists" in r.get_data(as_text=True).lower() print("duplicate email rejected: ok") # password reset round-trip tok = _csrf(get("/auth/reset").get_data(as_text=True)) buf2 = io.StringIO(); h2 = logging.StreamHandler(buf2); h2.setLevel(logging.INFO) app.logger.addHandler(h2) post("/auth/reset", {"csrf_token": tok, "email": "t@example.com"}) rlink = re.search(r"/auth/reset/(\S+)", buf2.getvalue()).group(1) tok = _csrf(get("/auth/reset/" + rlink).get_data(as_text=True)) r = post("/auth/reset/" + rlink, {"csrf_token": tok, "password": "NewPass456", "confirm": "NewPass456"}) assert "updated" in r.get_data(as_text=True).lower() tok = _csrf(get("/auth/login").get_data(as_text=True)) r = post("/auth/login", {"csrf_token": tok, "email": "t@example.com", "password": "NewPass456"}) assert "Test User" in r.get_data(as_text=True) print("password reset round-trip: ok") # misc assert get("/").status_code == 200 assert get("/healthz").get_json() == {"status": "ok"} assert get("/nope").status_code == 404 print("home / healthz / 404: ok") _phase2(app) _phase3(app) _phase4(app) _phase5(app) print("\nALL SMOKE CHECKS PASSED") def _phase2(app): """Phase 2: listings core — geocode, tier limits, accent search, radius, image pipeline, expiry sweep.""" import io from datetime import datetime, timedelta from werkzeug.datastructures import FileStorage from PIL import Image from app.models.user import User from app.models.category import Category from app.models.listing import Listing from app.models.enums import ListingStatus from app.services import listings as lsvc from app.services.geo import geocode_zip, haversine_mi from app.services.images import process_upload with app.app_context(): user = User.query.filter_by(email="t@example.com").first() for_sale = Category.query.filter_by(slug="for-sale").first() assert for_sale and for_sale.hot_fields(), "category schema missing" # geocode sanity g = geocode_zip("92683") assert g and g[2] == "Westminster" and g[3] == "CA" print(f"geocode 92683: ok ({g[2]}, {g[3]})") # create three (free cap = 3) l1 = lsvc.create_listing(user, for_sale, title="Phở Bò Đặc Biệt", body="Authentic beef noodle soup gear for sale.", lang="vi", price_cents=2500, zip_code="92683", raw_attributes={"condition": "good", "brand": "Acme"}) l2 = lsvc.create_listing(user, for_sale, title="Used Sofa", body="Comfortable three-seat sofa, gently used.", lang="en", price_cents=15000, zip_code="77036", raw_attributes={"condition": "good"}) l3 = lsvc.create_listing(user, for_sale, title="iPhone 14", body="Unlocked phone in great shape, no scratches.", lang="en", price_cents=40000, zip_code="92840", raw_attributes={"condition": "like_new"}) assert l1.lat and l1.attr_condition == "good" and l1.title_norm == "pho bo dac biet" print("create + geocode + hot-column + title_norm: ok") # tier limit: 4th create blocked try: lsvc.create_listing(user, for_sale, title="Fourth", body="blocked one two", lang="en", price_cents=100, zip_code="92683", raw_attributes={"condition": "fair"}) assert False, "expected tier limit error" except lsvc.ListingError as e: assert "limit" in str(e).lower() print("tier active-listing limit enforced: ok") # attribute validation (unit-level, independent of tier cap) from app.services.field_schema import validate_attributes _cleaned, _errs = validate_attributes(for_sale, {"condition": "banana"}) assert _errs.get("condition") == "invalid option" _cleaned, _errs = validate_attributes(for_sale, {"brand": "OK"}) assert _errs.get("condition") == "required" # required select missing _cleaned, _errs = validate_attributes(for_sale, {"condition": "new"}) assert not _errs and _cleaned["condition"] == "new" print("attribute schema validation: ok") # accent-insensitive search: 'pho' finds the diacritic title found = lsvc.browse_query(q="pho").all() assert l1 in found and l2 not in found print("accent-insensitive search: ok") # price filter cheap = lsvc.browse_query(max_price=5000).all() assert l1 in cheap and l3 not in cheap print("price filter: ok") # radius: near 92683 within 30 mi -> Westminster + Garden Grove, not Houston base = lsvc.browse_query() lat, lng = geocode_zip("92683")[0], geocode_zip("92683")[1] near = lsvc.search_with_radius(base, lat, lng, 30) ids = {l.id for l, d in near} assert l1.id in ids and l3.id in ids and l2.id not in ids d_gg = dict((l.id, d) for l, d in near)[l3.id] assert d_gg < 15, d_gg print(f"radius filter: ok (Garden Grove {d_gg} mi from Westminster)") # image pipeline: synthesize a PNG, process it buf = io.BytesIO() Image.new("RGB", (1200, 900), (80, 120, 200)).save(buf, "PNG") buf.seek(0) fs = FileStorage(stream=buf, filename="x.png", content_type="image/png") img = process_upload(fs, l1.id, sort_order=0) db.session.add(img); db.session.commit() assert img.path.endswith(".jpg") and img.thumb_path and img.width <= 1600 import os media = os.path.join(app.instance_path, "media", str(l1.id)) assert os.path.isdir(media) and len(os.listdir(media)) == 2 print("image pipeline (re-encode + thumbnail + EXIF strip): ok") # expiry sweep: backdate l2, sweep, confirm status transition l2.expires_at = datetime.utcnow() - timedelta(hours=1) db.session.commit() active_status_before = Listing.query.filter_by( user_id=user.id, status=ListingStatus.active).count() n = lsvc.expire_due_listings() active_status_after = Listing.query.filter_by( user_id=user.id, status=ListingStatus.active).count() assert n == 1, n assert active_status_after == active_status_before - 1 assert db.session.get(Listing, l2.id).status == ListingStatus.expired print(f"expiry sweep: ok (status active {active_status_before}->{active_status_after})") def test_smoke(): # pytest entry run() if __name__ == "__main__": run()