06/15 Phase 1 + 2 codes
This commit is contained in:
@@ -0,0 +1,253 @@
|
||||
"""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 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)
|
||||
|
||||
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 Listing.query.get(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()
|
||||
Reference in New Issue
Block a user