classifieds/tests/test_smoke.py

"""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()