404 lines
16 KiB
Python
404 lines
16 KiB
Python
"""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 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)
|
|
|
|
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()
|