#!/usr/bin/env python3 """Create MySQL coupon via admin, apply on quote+pay, verify coupon_redemptions.""" from __future__ import annotations import hashlib import json import subprocess import sys import time import urllib.error import urllib.request from datetime import date, timedelta from pathlib import Path BASE = "http://127.0.0.1:4000" PROP = "cmrnmicpg3480b58ce68332ecd34b" LOG = Path("/Users/saleh/.cursor/projects/Users-saleh-Documents-SafarLibya/terminals/887154.txt") class Fail(Exception): pass def mysql(sql: str) -> str: r = subprocess.run( ["mysql", "-uroot", "-psafar_dev", "safar_libya", "-N", "-e", sql], capture_output=True, text=True, ) if r.returncode != 0: raise Fail(r.stderr.strip()) return r.stdout.strip() def api(method: str, path: str, token: str | None = None, data: dict | None = None): body = None if data is None else json.dumps(data).encode() headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(BASE + path, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read().decode() code = resp.status except urllib.error.HTTPError as e: raw = e.read().decode() code = e.code try: parsed = json.loads(raw) if raw else {} except json.JSONDecodeError: parsed = {"_raw": raw[:500]} return code, parsed def must(ok: bool, label: str, code: int, payload: dict): print(f"{label}: HTTP {code}") if not ok: raise Fail(f"{label}: {json.dumps(payload, ensure_ascii=False)[:700]}") def main(): log_start = len(LOG.read_text().splitlines()) if LOG.exists() else 0 ts = int(time.time()) coupon_code = f"CUT{ts % 100000}" email = f"coupon{ts}@example.com" phone = f"+21892{ts % 100000000:08d}" password = "CouponTest123!" known_otp = "654321" otp_hash = hashlib.sha256(known_otp.encode()).hexdigest() print("=== ADMIN CREATE COUPON ===") code, d = api("POST", "/api/auth/login", data={"email": "admin@safarlibya.com", "password": "19992000"}) must(code == 200, "admin-login", code, d) admin = d["accessToken"] code, d = api( "POST", "/api/admin/coupons", token=admin, data={ "code": coupon_code, "type": "PERCENT", "value": 10, "maxUses": 50, "active": True, "note": "mysql coupon e2e", }, ) must(code == 201, "create-coupon", code, d) coupon_id = (d.get("coupon") or {}).get("id") print(" coupon", coupon_code, "id", coupon_id, "_db", d.get("_db")) must(d.get("_db") == "mysql", "coupon-mysql", code, d) print("=== REGISTER + OTP CUSTOMER ===") code, d = api( "POST", "/api/auth/register", data={ "fullName": "Coupon User", "email": email, "phone": phone, "password": password, "role": "CUSTOMER", "locale": "ar", }, ) must(code == 201, "register", code, d) token = d["accessToken"] uid = d["user"]["id"] mysql( f"""UPDATE otp_challenges SET code_hash='{otp_hash}', expires_at=DATE_ADD(UTC_TIMESTAMP(), INTERVAL 10 MINUTE), attempts=0, consumed_at=NULL WHERE user_id='{uid}' AND channel='EMAIL' ORDER BY created_at DESC LIMIT 1""" ) code, d = api("POST", "/api/auth/verify-otp", token=token, data={"code": known_otp}) must(code == 200, "otp", code, d) code, d = api("POST", "/api/auth/login", data={"email": email, "password": password}) must(code == 200, "login", code, d) token = d["accessToken"] print("=== QUOTE WITH COUPON ===") check_in = (date.today() + timedelta(days=40)).isoformat() check_out = (date.today() + timedelta(days=43)).isoformat() # baseline without coupon code, base = api( "POST", "/api/bookings/quote", token=token, data={"propertyId": PROP, "checkIn": check_in, "checkOut": check_out, "guests": 2}, ) must(code in (200, 201), "quote-base", code, base) base_b = base.get("booking") or base base_total = float(base_b.get("totalTnd") or 0) base_disc = float(base_b.get("discountTnd") or 0) # with coupon — need different dates or cancel previous pending mysql( f"UPDATE bookings SET status='EXPIRED' WHERE customer_id='{uid}' AND status='PENDING_PAYMENT'" ) code, d = api( "POST", "/api/bookings/quote", token=token, data={ "propertyId": PROP, "checkIn": check_in, "checkOut": check_out, "guests": 2, "couponCode": coupon_code, }, ) must(code in (200, 201), "quote-coupon", code, d) b = d.get("booking") or d bid = b.get("id") or b.get("_id") disc = float(b.get("discountTnd") or 0) total = float(b.get("totalTnd") or 0) print(" base_totalTnd", base_total, "base_disc", base_disc) print(" coupon_totalTnd", total, "coupon_disc", disc, "code", b.get("couponCode")) must(disc > 0, "discount-applied", code, d) must(total < base_total, "total-reduced", code, {"base": base_total, "total": total, "disc": disc}) must((b.get("couponCode") or "").upper() == coupon_code.upper(), "coupon-stored", code, d) print("=== PAY (records redemption) ===") code, d = api("POST", f"/api/bookings/{bid}/pay", token=token, data={}) must(code == 200, "pay", code, d) print("=== VERIFY MySQL coupon_redemptions ===") row = mysql( f"SELECT id, coupon_id, booking_id, code, discount_tnd FROM coupon_redemptions WHERE booking_id='{bid}'" ) print(" redemption", row) must(bool(row), "redemption-row", 200, {"bid": bid}) used = mysql(f"SELECT used_count FROM coupons WHERE id='{coupon_id}'") print(" used_count", used) must(int(float(used)) >= 1, "used-count", 200, {"used": used}) print("=== LOG SCAN ===") lines = LOG.read_text().splitlines()[log_start:] bad = [ ln for ln in lines if ("MongooseError" in ln) or ("buffering timed out" in ln) or ("MongoNetworkError" in ln) ] if bad: raise Fail("mongoose in logs:\n" + "\n".join(bad[:15])) print("OK coupon path fully MySQL") if __name__ == "__main__": try: main() except Fail as e: print("FAIL:", e, file=sys.stderr) sys.exit(1)