06/15 Phase 4 + 5 codes
This commit is contained in:
@@ -186,6 +186,298 @@ def _phase3(app):
|
||||
print("Phase 3 route renders: ok")
|
||||
|
||||
|
||||
|
||||
def _phase4(app):
|
||||
"""Phase 4: payments — models, boost activation, expiry, tier sync,
|
||||
webhook handler (mocked), route renders.
|
||||
No live Stripe calls; tests the logic layer with mock data.
|
||||
"""
|
||||
from datetime import datetime, timedelta
|
||||
from app.models.user import User
|
||||
from app.models.plan import Plan
|
||||
from app.models.listing import Listing
|
||||
from app.models.payments import Subscription, Transaction, Boost
|
||||
from app.models.enums import Role
|
||||
from app.services import billing as bsvc
|
||||
|
||||
with app.app_context():
|
||||
user = User.query.filter_by(email="t@example.com").first()
|
||||
listing = Listing.query.filter_by(user_id=user.id).first()
|
||||
pro_plan = Plan.query.filter_by(slug="pro").first()
|
||||
|
||||
# --- boost activation (no Stripe call) ---
|
||||
boost = bsvc.activate_boost(
|
||||
user_id=user.id,
|
||||
listing_id=listing.id,
|
||||
boost_type="featured",
|
||||
stripe_payment_intent_id="pi_test_001",
|
||||
amount_cents=499,
|
||||
)
|
||||
assert boost.is_active
|
||||
assert boost.type == "featured"
|
||||
listing_fresh = db.session.get(Listing, listing.id)
|
||||
assert listing_fresh.is_featured
|
||||
print("boost activation: ok")
|
||||
|
||||
# --- idempotency: same payment intent → no duplicate ---
|
||||
boost2 = bsvc.activate_boost(
|
||||
user_id=user.id, listing_id=listing.id,
|
||||
boost_type="featured",
|
||||
stripe_payment_intent_id="pi_test_001",
|
||||
amount_cents=499,
|
||||
)
|
||||
assert Boost.query.filter_by(listing_id=listing.id,
|
||||
type="featured").count() == 1
|
||||
print("boost idempotency: ok")
|
||||
|
||||
# --- second boost type (bump) ---
|
||||
bump = bsvc.activate_boost(
|
||||
user_id=user.id, listing_id=listing.id,
|
||||
boost_type="bump",
|
||||
stripe_payment_intent_id="pi_test_002",
|
||||
amount_cents=199,
|
||||
)
|
||||
assert bump.type == "bump"
|
||||
assert listing_fresh.bump_at is not None
|
||||
print("bump boost: ok")
|
||||
|
||||
# --- boost expiry sweep ---
|
||||
boost.expires_at = datetime.utcnow() - timedelta(hours=1)
|
||||
db.session.commit()
|
||||
n = bsvc.expire_boosts()
|
||||
assert n >= 1
|
||||
listing_after = db.session.get(Listing, listing.id)
|
||||
assert not listing_after.is_featured
|
||||
print(f"boost expiry sweep: ok (cleared {n})")
|
||||
|
||||
# --- sync_subscription: simulate webhook updating local DB ---
|
||||
# Build a fake Stripe subscription object
|
||||
fake_stripe_sub = {
|
||||
"id": "sub_test_001",
|
||||
"customer": "cus_test_001",
|
||||
"status": "active",
|
||||
"cancel_at_period_end": False,
|
||||
"current_period_end": int(
|
||||
(datetime.utcnow() + timedelta(days=30)).timestamp()),
|
||||
"items": {"data": [{"price": {"id": pro_plan.stripe_price_id or "price_test_pro"}}]},
|
||||
}
|
||||
# temporarily set stripe_price_id so sync works
|
||||
original_price_id = pro_plan.stripe_price_id
|
||||
pro_plan.stripe_price_id = "price_test_pro"
|
||||
fake_stripe_sub["items"]["data"][0]["price"]["id"] = "price_test_pro"
|
||||
db.session.commit()
|
||||
|
||||
sub = bsvc.sync_subscription(user.id, fake_stripe_sub)
|
||||
assert sub.stripe_sub_id == "sub_test_001"
|
||||
assert sub.status == "active"
|
||||
user_fresh = db.session.get(User, user.id)
|
||||
assert user_fresh.role == Role.subscriber
|
||||
assert user_fresh.tier.slug == "pro"
|
||||
print("sync_subscription → user upgraded to pro subscriber: ok")
|
||||
|
||||
# --- downgrade to free ---
|
||||
bsvc.downgrade_to_free(user.id)
|
||||
user_fresh = db.session.get(User, user.id)
|
||||
assert user_fresh.role == Role.free
|
||||
assert user_fresh.tier.slug == "free"
|
||||
sub_fresh = Subscription.query.filter_by(user_id=user.id).first()
|
||||
assert sub_fresh.status == "canceled"
|
||||
print("downgrade_to_free: ok")
|
||||
|
||||
# restore
|
||||
pro_plan.stripe_price_id = original_price_id
|
||||
db.session.commit()
|
||||
|
||||
# --- Transaction record exists ---
|
||||
txns = Transaction.query.filter_by(user_id=user.id).all()
|
||||
assert len(txns) >= 2 # featured + bump boosts
|
||||
print(f"transactions recorded: ok ({len(txns)} found)")
|
||||
|
||||
# --- route render checks ---
|
||||
import re
|
||||
c = app.test_client(); B = "https://localhost"
|
||||
def csrf(h): return re.search(r'name="csrf_token"[^>]*value="([^"]+)"', h).group(1)
|
||||
|
||||
# login
|
||||
r = c.get("/auth/login", base_url=B)
|
||||
tok = csrf(r.get_data(as_text=True))
|
||||
c.post("/auth/login", base_url=B,
|
||||
data={"csrf_token": tok, "email": "t@example.com",
|
||||
"password": "NewPass456"},
|
||||
headers={"Referer": B + "/auth/login"}, follow_redirects=True)
|
||||
|
||||
with app.app_context():
|
||||
lid = Listing.query.filter_by(
|
||||
user_id=User.query.filter_by(email="t@example.com").first().id
|
||||
).first().id
|
||||
|
||||
for path, expect in [
|
||||
("/pricing", 200),
|
||||
("/my/billing", 200),
|
||||
(f"/listings/{lid}/boost", 200),
|
||||
]:
|
||||
code = c.get(path, base_url=B).status_code
|
||||
assert code == expect, f"{path} -> {code}"
|
||||
print(f"{code} {path}")
|
||||
|
||||
print("Phase 4 route renders: ok")
|
||||
|
||||
|
||||
|
||||
def _phase5(app):
|
||||
"""Phase 5: ads, sponsors, promoted search."""
|
||||
from datetime import datetime, timedelta
|
||||
from app.models.ads import Ad, Sponsor, PromotedKeyword
|
||||
from app.models.listing import Listing
|
||||
from app.services.ads import (get_ad, record_impression, record_click,
|
||||
promoted_listings, active_sponsors,
|
||||
expire_promoted_keywords)
|
||||
|
||||
with app.app_context():
|
||||
# --- create test ad ---
|
||||
now = datetime.utcnow()
|
||||
ad = Ad(
|
||||
advertiser_name="Pho 99 Restaurant",
|
||||
slot="sidebar",
|
||||
target_url="https://pho99.example.com",
|
||||
alt_text="Best pho in town",
|
||||
lang="vi",
|
||||
starts_at=now - timedelta(hours=1),
|
||||
ends_at=now + timedelta(days=30),
|
||||
is_active=True,
|
||||
)
|
||||
db.session.add(ad); db.session.commit()
|
||||
print(f"ad created: id={ad.id} slot={ad.slot}")
|
||||
|
||||
# untargeted ad for fallback
|
||||
ad2 = Ad(
|
||||
advertiser_name="Generic Ads Inc",
|
||||
slot="header",
|
||||
target_url="https://generic.example.com",
|
||||
starts_at=now - timedelta(hours=1),
|
||||
ends_at=now + timedelta(days=30),
|
||||
is_active=True,
|
||||
)
|
||||
db.session.add(ad2); db.session.commit()
|
||||
|
||||
# --- get_ad: targeted hit ---
|
||||
found = get_ad("sidebar", lang="vi")
|
||||
assert found and found.id == ad.id
|
||||
print("get_ad targeted: ok")
|
||||
|
||||
# --- get_ad: slot with no match → None ---
|
||||
miss = get_ad("footer", lang="vi", state="CA")
|
||||
assert miss is None
|
||||
print("get_ad miss: ok")
|
||||
|
||||
# --- get_ad: untargeted fallback ---
|
||||
h = get_ad("header")
|
||||
assert h and h.id == ad2.id
|
||||
print("get_ad untargeted fallback: ok")
|
||||
|
||||
# --- impression tracking ---
|
||||
before = ad.impressions
|
||||
record_impression(ad.id)
|
||||
db.session.expire(ad)
|
||||
assert ad.impressions == before + 1
|
||||
print("impression tracking: ok")
|
||||
|
||||
# --- click tracking ---
|
||||
before_c = ad.clicks
|
||||
record_click(ad.id)
|
||||
db.session.expire(ad)
|
||||
assert ad.clicks == before_c + 1
|
||||
print("click tracking: ok")
|
||||
|
||||
# --- CTR ---
|
||||
assert ad.ctr > 0
|
||||
print(f"CTR: {ad.ctr}%: ok")
|
||||
|
||||
# --- sponsored ad suppressed for subscriber ---
|
||||
# (tested via plan limit — ad_free=True for basic+)
|
||||
from app.models.plan import Plan
|
||||
basic = Plan.query.filter_by(slug="basic").first()
|
||||
assert basic.limit("ad_free") is True
|
||||
print("ad_free plan limit: ok")
|
||||
|
||||
# --- sponsor directory ---
|
||||
sp = Sponsor(
|
||||
name="Little Saigon Weekly",
|
||||
url="https://lsweekly.example.com",
|
||||
tagline="Your community newspaper",
|
||||
tier="directory",
|
||||
starts_at=now - timedelta(hours=1),
|
||||
ends_at=now + timedelta(days=60),
|
||||
is_active=True,
|
||||
)
|
||||
db.session.add(sp); db.session.commit()
|
||||
sponsors = active_sponsors(tier="directory")
|
||||
assert any(s.id == sp.id for s in sponsors)
|
||||
print("sponsor directory: ok")
|
||||
|
||||
# --- category sponsor ---
|
||||
from app.models.category import Category
|
||||
jobs_cat = Category.query.filter_by(slug="jobs").first()
|
||||
cat_sp = Sponsor(
|
||||
name="VN Jobs Network",
|
||||
url="https://vnjobs.example.com",
|
||||
tier="category",
|
||||
category_id=jobs_cat.id,
|
||||
starts_at=now - timedelta(hours=1),
|
||||
ends_at=now + timedelta(days=60),
|
||||
is_active=True,
|
||||
)
|
||||
db.session.add(cat_sp); db.session.commit()
|
||||
cat_sponsors = active_sponsors(tier="category",
|
||||
category_id=jobs_cat.id)
|
||||
assert any(s.id == cat_sp.id for s in cat_sponsors)
|
||||
print("category sponsor: ok")
|
||||
|
||||
# --- promoted search ---
|
||||
listing = Listing.query.first()
|
||||
pk = PromotedKeyword(
|
||||
keyword="pho",
|
||||
listing_id=listing.id,
|
||||
priority=10,
|
||||
expires_at=now + timedelta(days=7),
|
||||
)
|
||||
db.session.add(pk); db.session.commit()
|
||||
|
||||
results = promoted_listings("pho")
|
||||
assert listing in results
|
||||
print("promoted search (keyword match): ok")
|
||||
|
||||
# accent-insensitive: "phở" should match keyword "pho"
|
||||
results2 = promoted_listings("phở")
|
||||
assert listing in results2
|
||||
print("promoted search (accent-insensitive): ok")
|
||||
|
||||
# no match for unrelated keyword
|
||||
results3 = promoted_listings("sofa")
|
||||
assert listing not in results3
|
||||
print("promoted search (no match): ok")
|
||||
|
||||
# --- expire promoted keywords ---
|
||||
pk.expires_at = now - timedelta(hours=1)
|
||||
db.session.commit()
|
||||
n = expire_promoted_keywords()
|
||||
assert n == 1
|
||||
print(f"expire promoted keywords: ok (removed {n})")
|
||||
|
||||
# --- route render checks ---
|
||||
import re
|
||||
c = app.test_client(); B = "https://localhost"
|
||||
for path, expect in [
|
||||
("/sponsors", 200),
|
||||
("/ads/1/click", 302), # redirects to target_url
|
||||
]:
|
||||
code = c.get(path, base_url=B).status_code
|
||||
assert code == expect, f"{path} -> {code}"
|
||||
print(f"{code} {path}")
|
||||
|
||||
print("Phase 5 route renders: ok")
|
||||
|
||||
|
||||
def run():
|
||||
app = create_app()
|
||||
with app.app_context():
|
||||
@@ -283,6 +575,8 @@ def run():
|
||||
|
||||
_phase2(app)
|
||||
_phase3(app)
|
||||
_phase4(app)
|
||||
_phase5(app)
|
||||
|
||||
print("\nALL SMOKE CHECKS PASSED")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user