"""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 shutil 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}") # Use a dedicated temp media dir so image counts are deterministic across runs. _SMOKE_MEDIA = os.path.join(tempfile.gettempdir(), "classifieds_smoke_media") if os.path.exists(_SMOKE_MEDIA): shutil.rmtree(_SMOKE_MEDIA) os.environ["MEDIA_ROOT"] = _SMOKE_MEDIA 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 _phase6(app): """Phase 6 foundation: audit log, settings, admin dashboard + user mgmt.""" import re from app.models.user import User from app.models.plan import Plan from app.models.listing import Listing from app.models.audit import AuditLog from app.models.enums import Role, UserStatus, TrustEventType from app.services import admin_users as usvc from app.services import admin_dashboard as dash from app.services.settings import get_setting, set_setting from app.models.report import Report from app.models.category import Category from app.models.enums import ListingStatus, ReportReason from app.services import listings as lsvc from app.services import reports as rsvc from app.services import moderation as msvc with app.app_context(): admin = User(email="admin@example.com", display_name="Admin", role=Role.admin, email_verified=True) admin.set_password("AdminPass123") admin.tier_id = Plan.query.filter_by(slug="free").first().id db.session.add(admin); db.session.commit() target = User.query.filter_by(email="t@example.com").first() # --- settings round-trip --- assert get_setting("flag_threshold", 5) == 5 set_setting("flag_threshold", 3) db.session.commit() assert get_setting("flag_threshold") == 3 print("settings round-trip: ok") # --- dashboard KPIs return sane values --- assert dash.active_listings_count() >= 0 assert dash.new_users_count(30) >= 2 assert dash.mrr_cents() >= 0 assert dash.revenue_30d_cents() >= 0 assert dash.flag_queue_depth() >= 0 print("dashboard KPI helpers: ok") # --- ban + audit log --- usvc.set_status(target, UserStatus.banned, actor=admin) db.session.commit() target_fresh = db.session.get(User, target.id) assert target_fresh.status == UserStatus.banned assert not target_fresh.is_active log = AuditLog.query.filter_by(target_type="user", target_id=target.id, action="user.banned").first() assert log is not None and log.actor_id == admin.id print("ban + audit log: ok") # --- tier override + audit log --- pro_plan = Plan.query.filter_by(slug="pro").first() usvc.set_tier(target, pro_plan, actor=admin) db.session.commit() assert db.session.get(User, target.id).tier_id == pro_plan.id assert AuditLog.query.filter_by(action="user.tier_override", target_id=target.id).count() == 1 print("tier override + audit log: ok") # --- trust adjust + audit log --- score_before = target.trust_score usvc.adjust_trust(target, TrustEventType.payment, 10, actor=admin) db.session.commit() assert db.session.get(User, target.id).trust_score == score_before + 10 assert AuditLog.query.filter_by(action="user.trust_adjust", target_id=target.id).count() == 1 print("trust adjust + audit log: ok") # --- reactivate so later route checks reflect a normal account --- usvc.set_status(target, UserStatus.active, actor=admin) db.session.commit() # --- keyword blocklist: auto-flag on create --- for_sale = Category.query.filter_by(slug="for-sale").first() set_setting("keyword_blocklist", ["viagra"]) db.session.commit() spammy = lsvc.create_listing(target, for_sale, title="Cheap meds", body="Buy real viagra online cheap", lang="en", price_cents=500, zip_code="92683", raw_attributes={"condition": "new"}) assert spammy.status == ListingStatus.flagged print("keyword blocklist auto-flags on create: ok") # --- keyword blocklist: auto-flag on update --- clean = lsvc.create_listing(target, for_sale, title="Clean Listing", body="A perfectly normal item for sale.", lang="en", price_cents=500, zip_code="92683", raw_attributes={"condition": "new"}) assert clean.status == ListingStatus.active lsvc.update_listing(clean, for_sale, title="Clean Listing", body="Now selling viagra too", lang="en", price_cents=500, zip_code="92683", raw_attributes={"condition": "new"}) assert clean.status == ListingStatus.flagged print("keyword blocklist auto-flags on update: ok") set_setting("keyword_blocklist", []) db.session.commit() # --- reports: self-report blocked --- report_target = lsvc.create_listing(target, for_sale, title="Reportable Item", body="Nothing wrong with this one.", lang="en", price_cents=1000, zip_code="92683", raw_attributes={"condition": "new"}) try: rsvc.create_report(report_target, target, ReportReason.spam) assert False, "expected self-report to be blocked" except rsvc.ReportError as e: assert "own listing" in str(e) print("self-report blocked: ok") # --- reports: duplicate report blocked --- buyer = User.query.filter_by(email="buyer@example.com").first() rsvc.create_report(report_target, buyer, ReportReason.spam, note="looks fake") try: rsvc.create_report(report_target, buyer, ReportReason.scam) assert False, "expected duplicate report to be blocked" except rsvc.ReportError as e: assert "already reported" in str(e) print("duplicate report blocked: ok") # --- reports: threshold auto-flags --- set_setting("flag_threshold", 2) db.session.commit() assert report_target.flag_count == 1 assert report_target.status == ListingStatus.active rsvc.create_report(report_target, admin, ReportReason.scam) assert report_target.flag_count == 2 assert report_target.status == ListingStatus.flagged print(f"report threshold auto-flag: ok (flag_count={report_target.flag_count})") # --- moderation actions + audit log --- msvc.approve(report_target, actor=admin) db.session.commit() assert report_target.status == ListingStatus.active and report_target.flag_count == 0 assert AuditLog.query.filter_by(action="listing.approved", target_id=report_target.id).count() == 1 print("moderation approve + audit log: ok") msvc.hide(spammy, actor=admin) db.session.commit() assert spammy.status == ListingStatus.flagged assert AuditLog.query.filter_by(action="listing.hidden", target_id=spammy.id).count() == 1 print("moderation hide + audit log: ok") msvc.remove(clean, actor=admin) db.session.commit() assert clean.status == ListingStatus.removed assert AuditLog.query.filter_by(action="listing.removed", target_id=clean.id).count() == 1 print("moderation remove + audit log: ok") # lower threshold to 1 so a single HTTP report flips status for the next check set_setting("flag_threshold", 1) http_listing = lsvc.create_listing(target, for_sale, title="HTTP Report Test", body="Totally fine listing.", lang="en", price_cents=750, zip_code="92683", raw_attributes={"condition": "new"}) http_listing_id = http_listing.id db.session.commit() # --- route checks --- c = app.test_client(); B = "https://localhost" def csrf(html): return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', html).group(1) def login(email, password): r = c.get("/auth/login", base_url=B) tok = csrf(r.get_data(as_text=True)) return c.post("/auth/login", base_url=B, data={"csrf_token": tok, "email": email, "password": password}, headers={"Referer": B + "/auth/login"}, follow_redirects=True) # non-admin gets 403 login("t@example.com", "NewPass456") for path in ("/admin", "/admin/listings", "/admin/reports", "/admin/settings", "/admin/categories", "/admin/plans", "/admin/transactions", "/admin/audit", "/admin/ads", "/admin/sponsors", "/admin/promoted-keywords", "/admin/analytics"): assert c.get(path, base_url=B).status_code == 403 print("non-admin /admin* -> 403: ok") c.get("/auth/logout", base_url=B) # admin gets 200 on all admin pages login("admin@example.com", "AdminPass123") with app.app_context(): target_id = User.query.filter_by(email="t@example.com").first().id cat_id = Category.query.filter_by(parent_id=None).first().id plan_id = Plan.query.filter_by(slug="free").first().id for path in ("/admin", "/admin/users", f"/admin/users/{target_id}", "/admin/listings", "/admin/reports", "/admin/settings", "/admin/categories", f"/admin/categories/{cat_id}/schema", "/admin/plans", f"/admin/plans/{plan_id}", "/admin/transactions", "/admin/audit", "/admin/ads", "/admin/ads/new", "/admin/sponsors", "/admin/sponsors/new", "/admin/promoted-keywords", "/admin/promoted-keywords/new", "/admin/analytics"): code = c.get(path, base_url=B).status_code assert code == 200, f"{path} -> {code}" print(f"{code} {path}") print("Phase 6 route renders: ok") # ban via route, confirm banned user's login is rejected r = c.get(f"/admin/users/{target_id}", base_url=B) tok = csrf(r.get_data(as_text=True)) c.post(f"/admin/users/{target_id}/ban", base_url=B, data={"csrf_token": tok}, headers={"Referer": B + f"/admin/users/{target_id}"}, follow_redirects=True) with app.app_context(): assert db.session.get(User, target_id).status == UserStatus.banned c.get("/auth/logout", base_url=B) r = login("t@example.com", "NewPass456") assert "suspended" in r.get_data(as_text=True).lower() print("banned user login rejected: ok") # buyer reports a listing via the real HTTP route c.get("/auth/logout", base_url=B) login("buyer@example.com", "BuyerPass123") r = c.get(f"/listings/{http_listing_id}", base_url=B) tok = csrf(r.get_data(as_text=True)) c.post(f"/listings/{http_listing_id}/report", base_url=B, data={"csrf_token": tok, "reason": "spam", "note": "test report"}, headers={"Referer": B + f"/listings/{http_listing_id}"}, follow_redirects=True) with app.app_context(): rep = Report.query.filter_by(listing_id=http_listing_id).first() assert rep is not None and rep.reason == ReportReason.spam report_id = rep.id assert db.session.get(Listing, http_listing_id).status == ListingStatus.flagged print("user-facing report route: ok") # admin sees it in the open reports queue, dismisses it, it disappears c.get("/auth/logout", base_url=B) login("admin@example.com", "AdminPass123") r = c.get("/admin/reports", base_url=B) assert "HTTP Report Test" in r.get_data(as_text=True) tok = csrf(r.get_data(as_text=True)) c.post(f"/admin/reports/{report_id}/dismiss", base_url=B, data={"csrf_token": tok}, headers={"Referer": B + "/admin/reports"}, follow_redirects=True) r2 = c.get("/admin/reports", base_url=B) assert "HTTP Report Test" not in r2.get_data(as_text=True) print("report dismiss hides from open queue: ok") # --- contact_density_threshold: configurable via settings --- from app.services.contact import contact_density, contact_revealed with app.app_context(): body = "Call me at 555-123-4567" assert contact_density(body) == 1 set_setting("contact_density_threshold", 1) db.session.commit() assert contact_density(body) >= get_setting("contact_density_threshold", 3) set_setting("contact_density_threshold", 3) db.session.commit() assert not (contact_density(body) >= get_setting("contact_density_threshold", 3)) print("contact_density_threshold configurable: ok") # --- new_user_trust_gate_days: blocks reveal even for trusted+verified --- with app.app_context(): buyer = User.query.filter_by(email="buyer@example.com").first() assert contact_revealed(buyer) set_setting("new_user_trust_gate_days", 9999) db.session.commit() assert not contact_revealed(buyer) set_setting("new_user_trust_gate_days", 0) db.session.commit() assert contact_revealed(buyer) print("new_user_trust_gate_days gates contact reveal: ok") # --- ads_enabled: short-circuits inject_ads() --- from app.blueprints.ads.routes import inject_ads with app.app_context(): set_setting("ads_enabled", False) db.session.commit() with app.test_request_context("/"): assert inject_ads() == {"ads": {}, "show_ads": False} with app.app_context(): set_setting("ads_enabled", True) db.session.commit() print("ads_enabled short-circuits inject_ads: ok") # --- registration_open: blocks new registrations when closed --- c.get("/auth/logout", base_url=B) with app.app_context(): set_setting("registration_open", False) db.session.commit() tok = csrf(c.get("/auth/register", base_url=B).get_data(as_text=True)) r = c.post("/auth/register", base_url=B, data={"csrf_token": tok, "display_name": "Blocked", "email": "blocked@example.com", "password": "BlockedPass123", "confirm": "BlockedPass123"}, headers={"Referer": B + "/auth/register"}, follow_redirects=True) assert "closed" in r.get_data(as_text=True).lower() with app.app_context(): assert User.query.filter_by(email="blocked@example.com").first() is None set_setting("registration_open", True) db.session.commit() print("registration_open blocks new registrations: ok") # --- maintenance_mode: 503 for non-admin, admin bypasses --- with app.app_context(): set_setting("maintenance_mode", True) db.session.commit() c.get("/auth/logout", base_url=B) assert c.get("/", base_url=B).status_code == 503 login("admin@example.com", "AdminPass123") assert c.get("/admin", base_url=B).status_code == 200 c.get("/auth/logout", base_url=B) with app.app_context(): set_setting("maintenance_mode", False) db.session.commit() assert c.get("/", base_url=B).status_code == 200 print("maintenance_mode blocks non-admins, admin bypasses: ok") # restore target to active for cleanliness (not strictly needed, smoke ends here) with app.app_context(): u = User.query.filter_by(email="t@example.com").first() u.status = UserStatus.active db.session.commit() 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) _phase6(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_root = (app.config.get("MEDIA_ROOT") or os.path.join(app.instance_path, "media")) media = os.path.join(media_root, 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()