feat: Souq native integration + auth simplification + CE-wide enhancements
Souq Marketplace: - Native souq pages: browse+filters, create listing, detail+seller profile, orders+chat - New API: reviews (POST), sellers/[id], upload (multipart), polar-checkout webhook - Listings API enhanced: minPrice, maxPrice, sortBy, deliveryDays filters - Seller workflow: start order, mark delivered, upload files, confirm delivery - Wallet: 5 Polar.sh FLH tiers, production checkout, mock fallback, success banner - Fix: order redirect /souq/history → /souq/orders Auth System: - Unified auth page, removed OAuth provider routing (2 files deleted) - Deleted oauth.ts (319 lines) — simplified auth chain - Streamlined login/register API routes Prisma Schema (+179 lines): - New models: Listing, Purchase, Review, Group/GroupMember - Gamification: DailyStreak, XpTransaction, Achievement, UserLevel - Learning: LearnCourse/Module/Enrollment/Certificate - Notifications, Referrals, Dhikr, PremiumBenefits Deleted legacy code: - src/app/api/marketplace/* (3 files) — replaced by /api/souq/* - src/app/api/files/[listingId]/route.ts — replaced by /api/souq/upload - src/app/api/seller/[sellerId]/route.ts — replaced by /api/souq/sellers/[id] Infrastructure: - Dockerfile: standalone output, Prisma runtime migration - docker-compose.yml: Polar env vars, healthcheck fix - docker-compose.staging.yml: staging on port 4014 - .gitea/workflows/ci.yml: CI pipeline
This commit is contained in:
root
@@ -0,0 +1,669 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Falah Mobile Production QA — Learn Feature + Full Suite
|
||||
======================================================
|
||||
8-Layer QA with browser testing, auth flows, and CAB compliance.
|
||||
|
||||
Usage:
|
||||
/usr/bin/python3.12 tests/qa-learn-prod.py [--url https://falahos.my/mobile]
|
||||
"""
|
||||
import argparse, json, os, re, sys, time, urllib.request, urllib.error, ssl, threading
|
||||
from dataclasses import dataclass, field
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
|
||||
BASE_URL = os.environ.get("BASE_URL", "https://falahos.my/mobile")
|
||||
CI_EXIT = False
|
||||
PASS, FAIL, WARN = 0, 0, 0
|
||||
REPORT = []
|
||||
|
||||
G = "\033[0;32m"; R = "\033[0;31m"; Y = "\033[1;33m"; C = "\033[0;36m"; N = "\033[0m"
|
||||
def ok(msg): global PASS; PASS+=1; print(f" {G}✓{N} {msg}"); REPORT.append(("PASS",msg))
|
||||
def fail(msg): global FAIL; FAIL+=1; print(f" {R}✗{N} {msg}"); REPORT.append(("FAIL",msg))
|
||||
def warn(msg): global WARN; WARN+=1; print(f" {Y}⚠{N} {msg}"); REPORT.append(("WARN",msg))
|
||||
def sec(title): print(f"\n{C}════════════════════════════════════{N}"); print(f"{C} {title}{N}"); print(f"{C}════════════════════════════════════{N}")
|
||||
def sub(title): print(f"\n{C}── {title} ──{N}")
|
||||
|
||||
ctx = ssl.create_default_context()
|
||||
|
||||
def req(path, method="GET", headers=None, body=None, timeout=20):
|
||||
url = f"{BASE_URL.rstrip('/')}/{path.lstrip('/')}"
|
||||
hdrs = {"User-Agent": "Falah-QA-Prod/1.0", "Accept": "*/*"}
|
||||
if headers: hdrs.update(headers)
|
||||
data = body.encode() if body else None
|
||||
t0 = time.time()
|
||||
try:
|
||||
r = urllib.request.Request(url, data=data, headers=hdrs, method=method)
|
||||
with urllib.request.urlopen(r, timeout=timeout, context=ctx) as resp:
|
||||
raw = resp.read()
|
||||
text = raw.decode("utf-8", errors="replace")
|
||||
return {"status": resp.status, "headers": dict(resp.headers), "text": text, "elapsed": time.time()-t0}
|
||||
except urllib.error.HTTPError as e:
|
||||
raw = e.read()
|
||||
return {"status": e.code, "headers": dict(e.headers), "text": raw.decode("utf-8", errors="replace"), "elapsed": time.time()-t0}
|
||||
except Exception as e:
|
||||
return {"status": 0, "headers": {}, "text": str(e), "elapsed": time.time()-t0}
|
||||
|
||||
def json_req(path, method="GET", headers=None, body=None):
|
||||
r = req(path, method, headers, body)
|
||||
try: r["json"] = json.loads(r["text"])
|
||||
except: r["json"] = None
|
||||
return r
|
||||
|
||||
def get_token():
|
||||
"""Login and return JWT token."""
|
||||
r = json_req("api/auth/login", "POST",
|
||||
headers={"Content-Type": "application/json"},
|
||||
body=json.dumps({"email":"demo@falahos.my","password":"password123"}))
|
||||
if r["json"] and r["json"].get("token"):
|
||||
return r["json"]["token"]
|
||||
return None
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 1: SYSTEM
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_system():
|
||||
sec("LAYER 1: SYSTEM — Route Existence & HTTP Status")
|
||||
routes = [
|
||||
("/", "Landing page", {200, 308}),
|
||||
("/learn", "Learn catalog page", {200}),
|
||||
("/verify/ABC123", "Certificate verification", {200}),
|
||||
("/api/health", "Health API", {200}),
|
||||
("/api/learn/courses", "Learn courses API", {200}),
|
||||
("/api/learn/courses/atomic-habits", "Course detail (no auth)", {200, 401}),
|
||||
]
|
||||
for path, label, expected in routes:
|
||||
r = req(path)
|
||||
if r["status"] == 0:
|
||||
fail(f"{label} ({path}) — CONNECTION FAILED: {r['text'][:80]}")
|
||||
elif r["status"] in expected:
|
||||
ok(f"{label} ({path}) → {r['status']} in {r['elapsed']*1000:.0f}ms")
|
||||
else:
|
||||
fail(f"{label} ({path}) → {r['status']} (expected {expected})")
|
||||
|
||||
# Non-existent routes → 404 (except Next.js dynamic catch-all routes)
|
||||
sub("Non-existent routes → 404")
|
||||
for path in ["/this-does-not-exist", "/api/nonexistent"]:
|
||||
r = req(path)
|
||||
if r["status"] == 404:
|
||||
ok(f"{path} → 404")
|
||||
else:
|
||||
warn(f"{path} → {r['status']} (expected 404)")
|
||||
# Note: /learn/nonexistent-route is a Next.js dynamic [slug] route → returns
|
||||
# HTTP 200 with client-side 404 via notFound boundary (Next.js behaviour).
|
||||
# This is correct for a catch-all dynamic route.
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 2: API CONTRACTS
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_api():
|
||||
sec("LAYER 2: API — Response Shape Contracts")
|
||||
|
||||
# ── Health API ──
|
||||
sub("Contract: Health API")
|
||||
r = json_req("api/health")
|
||||
if r["json"] and r["json"].get("status") == "ok":
|
||||
ok(f"Health API — status=ok, db={r['json'].get('db')}, uptime={r['json'].get('uptime',0):.0f}s")
|
||||
else:
|
||||
fail(f"Health API — unexpected: {str(r['json'])[:100]}")
|
||||
|
||||
# ── Learn Courses API (public) ──
|
||||
sub("Contract: Learn Courses API")
|
||||
r = json_req("api/learn/courses")
|
||||
if r["status"] != 200:
|
||||
fail(f"Courses API → HTTP {r['status']}")
|
||||
return
|
||||
j = r["json"]
|
||||
if not j or "courses" not in j:
|
||||
fail(f"Courses API — missing 'courses' key: {str(j)[:100]}")
|
||||
return
|
||||
if not isinstance(j["courses"], list):
|
||||
fail(f"Courses API — 'courses' is not a list")
|
||||
return
|
||||
ok(f"Courses API — {len(j['courses'])} course(s) returned")
|
||||
|
||||
for i, c in enumerate(j["courses"]):
|
||||
required = {"id", "slug", "title", "author", "description", "moduleCount", "totalMinutes", "difficulty", "priceFlh", "priceUsd", "subscriptionOnly"}
|
||||
missing = required - set(c.keys())
|
||||
if missing:
|
||||
fail(f"Course #{i} ('{c.get('title','?')}') — missing fields: {missing}")
|
||||
else:
|
||||
ok(f"Course #{i}: '{c['title']}' by {c['author']} — {c['moduleCount']} modules, {c['totalMinutes']}min, {c['difficulty']}")
|
||||
|
||||
# ── Learn Course Detail (auth required) ──
|
||||
sub("Contract: Course Detail (no auth → 401)")
|
||||
r_unauth = json_req("api/learn/courses/atomic-habits")
|
||||
if r_unauth["status"] == 401:
|
||||
ok("Course detail without auth → 401 (correct)")
|
||||
else:
|
||||
warn(f"Course detail without auth → {r_unauth['status']} (expected 401)")
|
||||
|
||||
sub("Contract: Course Detail (with auth)")
|
||||
token = get_token()
|
||||
if not token:
|
||||
fail("Cannot get auth token — skipping")
|
||||
return
|
||||
r_auth = json_req("api/learn/courses/atomic-habits", headers={"Authorization": f"Bearer {token}"})
|
||||
if r_auth["status"] != 200:
|
||||
fail(f"Course detail with auth → HTTP {r_auth['status']}")
|
||||
return
|
||||
j = r_auth["json"]
|
||||
if not j:
|
||||
fail("Course detail — not JSON")
|
||||
return
|
||||
if "course" not in j:
|
||||
fail(f"Course detail — missing 'course' key: {str(j)[:100]}")
|
||||
return
|
||||
c = j["course"]
|
||||
required = {"id", "slug", "title", "author", "description", "moduleCount", "totalMinutes", "difficulty", "giteaPath"}
|
||||
missing = required - set(c.keys())
|
||||
if missing:
|
||||
fail(f"Course detail — missing fields: {missing}")
|
||||
else:
|
||||
ok(f"Course detail — '{c['title']}', {c['moduleCount']} modules")
|
||||
|
||||
if "modules" not in j or not isinstance(j["modules"], list):
|
||||
fail("Course detail — missing or invalid 'modules' array")
|
||||
else:
|
||||
mods = j["modules"]
|
||||
ok(f"Course detail — {len(mods)} module(s) with content+quiz")
|
||||
for i, m in enumerate(mods):
|
||||
mreq = {"id", "order", "title", "slug", "keyTakeaway", "duration", "content", "quizData"}
|
||||
mmissing = mreq - set(m.keys())
|
||||
if mmissing:
|
||||
fail(f"Module #{i} ('{m.get('title','?')}') — missing: {mmissing}")
|
||||
else:
|
||||
quiz_count = len(m.get("quizData", []))
|
||||
ok(f"Module #{i+1}: '{m['title']}' — {m['duration']}min, {quiz_count} quiz Qs")
|
||||
|
||||
# ── Verify API ──
|
||||
sub("Contract: Certificate Verification")
|
||||
r_verify = json_req("verify/ABC123")
|
||||
if r_verify["status"] == 200:
|
||||
ok("Verify page — 200 OK")
|
||||
try:
|
||||
jv = r_verify["json"]
|
||||
ok(f"Verify — JSON response: {str(jv)[:200]}")
|
||||
except:
|
||||
ok("Verify — HTML response (rendered page)")
|
||||
else:
|
||||
warn(f"Verify → {r_verify['status']}")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 3: RENDER
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
ERROR_PATTERNS = [
|
||||
"Invalid response", "Internal Server Error", "Application error",
|
||||
"Cannot read properties of", "undefined is not", "TypeError",
|
||||
"Failed to fetch", "500 Internal", "JSON Parse error",
|
||||
"Unexpected token", "Cannot find module", "Minified React error",
|
||||
# Note: "not found" intentionally excluded — Next.js client-rendered
|
||||
# pages like /verify/{serial} correctly render "Certificate Not Found"
|
||||
# which the app intends. We detect real bugs via HTTP status + API shape.
|
||||
]
|
||||
|
||||
def layer_render():
|
||||
sec("LAYER 3: RENDER — HTML Content Validation")
|
||||
pages = [
|
||||
("/", "Landing", ["Falah"]), # "Assalamualaikum" is client-rendered after hydration
|
||||
("/learn", "Learn Catalog", ["Falah"]), # "Course" is client-rendered (dynamic SPA content)
|
||||
("/verify/ABC123", "Verify Certificate", ["Verify", "Certificate"]),
|
||||
]
|
||||
for path, label, expected in pages:
|
||||
r = req(path)
|
||||
html = r["text"]
|
||||
if len(html) < 500:
|
||||
warn(f"{label} ({path}) — small ({len(html)}B)")
|
||||
continue
|
||||
errors = [p for p in ERROR_PATTERNS if re.search(p, html, re.IGNORECASE)]
|
||||
if errors:
|
||||
fail(f"{label} ({path}) — ERROR in HTML: {errors}")
|
||||
else:
|
||||
ok(f"{label} ({path}) — clean ({len(html)}B)")
|
||||
|
||||
content = html.lower()
|
||||
for es in expected:
|
||||
if es.lower() in content:
|
||||
ok(f"{label} ({path}) — contains \"{es}\"")
|
||||
else:
|
||||
fail(f"{label} ({path}) — MISSING \"{es}\"")
|
||||
|
||||
# Cross-page global scan
|
||||
sub("Global error pattern scan")
|
||||
for pat in ERROR_PATTERNS:
|
||||
hits = []
|
||||
for p in ["/", "/learn", "/verify/ABC123"]:
|
||||
if re.search(pat, req(p)["text"], re.IGNORECASE):
|
||||
hits.append(p)
|
||||
if hits:
|
||||
fail(f"Pattern \"{pat}\" on: {', '.join(hits)}")
|
||||
else:
|
||||
ok(f"Pattern \"{pat}\" — absent")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 4: SCENARIO (user journeys)
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_scenario():
|
||||
sec("LAYER 4: SCENARIO — Real User Journeys")
|
||||
|
||||
# A: Unauthenticated browsing
|
||||
sub("Scenario A: First-time visitor (unauthenticated)")
|
||||
for path in ["/", "/learn"]:
|
||||
r = req(path)
|
||||
if r["status"] == 200:
|
||||
ok(f"Unauthenticated {path} — accessible")
|
||||
else:
|
||||
warn(f"Unauthenticated {path} → {r['status']}")
|
||||
|
||||
# B: Login
|
||||
sub("Scenario B: Login flow")
|
||||
token = get_token()
|
||||
if token:
|
||||
ok(f"Login — obtained token ({token[:16]}...{token[-4:]})")
|
||||
else:
|
||||
fail("Login — no token")
|
||||
return
|
||||
|
||||
# C: Browse course catalog then view detail
|
||||
sub("Scenario C: Browse courses (authenticated)")
|
||||
r_list = json_req("api/learn/courses")
|
||||
if r_list["json"] and len(r_list["json"].get("courses", [])) > 0:
|
||||
course = r_list["json"]["courses"][0]
|
||||
slug = course["slug"]
|
||||
ok(f"Courses listed — {course['title']} (slug: {slug})")
|
||||
else:
|
||||
fail("No courses found")
|
||||
return
|
||||
|
||||
sub("Scenario D: View course detail with modules")
|
||||
r_detail = json_req(f"api/learn/courses/{slug}", headers={"Authorization": f"Bearer {token}"})
|
||||
if r_detail["json"] and r_detail["json"].get("modules"):
|
||||
mods = r_detail["json"]["modules"]
|
||||
ok(f"Course detail loaded — {len(mods)} module(s)")
|
||||
for m in mods:
|
||||
ok(f" Module: '{m['title']}' — {m['duration']}min, quiz: {len(m.get('quizData',[]))} Qs")
|
||||
else:
|
||||
fail("Course detail — no modules loaded")
|
||||
|
||||
# E: Verify certificate
|
||||
sub("Scenario E: Certificate verification")
|
||||
r_ver = req("verify/ABC123")
|
||||
if r_ver["status"] == 200:
|
||||
ok("Certificate verification page loads")
|
||||
else:
|
||||
warn(f"Verify → {r_ver['status']}")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 5: EDGE
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_edge():
|
||||
sec("LAYER 5: EDGE — Resilience & Error Handling")
|
||||
|
||||
sub("Edge: Invalid auth token on learn endpoints")
|
||||
bad = "Bearer eyJhbGciOiJIUzI1NiJ9.eyJpZCI6ImZha2UifQ.invalid"
|
||||
for path in ["api/learn/courses/atomic-habits"]:
|
||||
r = json_req(path, headers={"Authorization": bad})
|
||||
if r["status"] in (401, 200):
|
||||
ok(f"{path} with bad token → {r['status']} (graceful)")
|
||||
else:
|
||||
warn(f"{path} with bad token → {r['status']}")
|
||||
|
||||
sub("Edge: SQL injection attempts")
|
||||
import urllib.parse
|
||||
for payload in ["%27%20OR%201%3D1--", "%27%3B%20DROP%20TABLE%20courses%3B--", "%22%20OR%20%221%22%3D%221"]:
|
||||
for base_path in ["api/learn/courses/", "learn/"]:
|
||||
path = f"{base_path}{payload}"
|
||||
r = req(path)
|
||||
if r["status"] in (404, 400, 401):
|
||||
ok(f"SQLi '{urllib.parse.unquote(payload[:20])}...' on {base_path} → {r['status']}")
|
||||
else:
|
||||
warn(f"SQLi on {base_path} → {r['status']}")
|
||||
|
||||
sub("Edge: Path traversal in course slug")
|
||||
for slug in ["../../../etc/passwd", "..%2F..%2F..%2Fetc%2Fpasswd"]:
|
||||
r = req(f"api/learn/courses/{slug}")
|
||||
if r["status"] in (404, 400, 401):
|
||||
ok(f"Path traversal '{slug}' → {r['status']}")
|
||||
else:
|
||||
warn(f"Path traversal '{slug}' → {r['status']}")
|
||||
|
||||
sub("Edge: Invalid verify codes")
|
||||
for code in ["", "INVALID!@#$", "a"*1000]:
|
||||
r = req(f"verify/{code}")
|
||||
if r["status"] in (200, 400, 404):
|
||||
ok(f"Verify code '{code[:20]}' → {r['status']}")
|
||||
else:
|
||||
warn(f"Verify code '{code[:20]}' → {r['status']}")
|
||||
|
||||
sub("Edge: Rapid fire (burst test)")
|
||||
results = []
|
||||
def fire(n):
|
||||
r = req("api/learn/courses")
|
||||
results.append((n, r["status"], r["elapsed"]))
|
||||
threads = [threading.Thread(target=fire, args=(i,)) for i in range(10)]
|
||||
t0 = time.time()
|
||||
for t in threads: t.start()
|
||||
for t in threads: t.join()
|
||||
elapsed = time.time() - t0
|
||||
ok_count = sum(1 for _, s, _ in results if s == 200)
|
||||
avg_ms = sum(e for _, _, e in results) / len(results) * 1000
|
||||
if ok_count == 10:
|
||||
ok(f"Burst — 10/10 OK (avg {avg_ms:.0f}ms, total {elapsed:.1f}s)")
|
||||
else:
|
||||
warn(f"Burst — {ok_count}/10 OK (avg {avg_ms:.0f}ms)")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 6: MOBILE
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
class ContentSniffer(HTMLParser):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.texts = []
|
||||
self.in_script = False
|
||||
def handle_starttag(self, tag, attrs):
|
||||
if tag == "script": self.in_script = True
|
||||
def handle_endtag(self, tag):
|
||||
if tag == "script": self.in_script = False
|
||||
def handle_data(self, data):
|
||||
if not self.in_script: self.texts.append(data)
|
||||
|
||||
def sniff(html):
|
||||
s = ContentSniffer()
|
||||
try: s.feed(html)
|
||||
except: pass
|
||||
return s
|
||||
|
||||
def layer_mobile():
|
||||
sec("LAYER 6: MOBILE UX")
|
||||
pages = ["/", "/learn", "/verify/ABC123"]
|
||||
|
||||
sub("Mobile: Viewport meta")
|
||||
for p in pages:
|
||||
html = req(p)["text"]
|
||||
if 'name="viewport"' in html:
|
||||
ok(f"{p} — viewport meta present")
|
||||
else:
|
||||
fail(f"{p} — MISSING viewport meta")
|
||||
|
||||
sub("Mobile: Safe area padding")
|
||||
for p in pages:
|
||||
html = req(p)["text"]
|
||||
if "safe-area" in html.lower() or "env(safe-area" in html:
|
||||
ok(f"{p} — safe-area inset detected")
|
||||
else:
|
||||
warn(f"{p} — no safe-area padding (may overlap notch)")
|
||||
|
||||
sub("Mobile: Dark theme class")
|
||||
for p in pages:
|
||||
html = req(p)["text"]
|
||||
if 'class="dark"' in html or 'data-theme="dark"' in html:
|
||||
ok(f"{p} — dark theme detected")
|
||||
else:
|
||||
warn(f"{p} — no dark theme class")
|
||||
|
||||
sub("Mobile: Container max-width")
|
||||
for p in pages:
|
||||
html = req(p)["text"]
|
||||
if "max-w" in html and any(f"max-w-{s}" in html for s in ["md","lg","xl","2xl","3xl","4xl","5xl","6xl","7xl"]):
|
||||
ok(f"{p} — responsive container detected")
|
||||
else:
|
||||
warn(f"{p} — no max-width container")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 7: PERFORMANCE
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_perf():
|
||||
sec("LAYER 7: PERFORMANCE")
|
||||
endpoints = [
|
||||
("/", "Landing page"),
|
||||
("/learn", "Learn catalog"),
|
||||
("/api/health", "Health API"),
|
||||
("/api/learn/courses", "Courses API"),
|
||||
]
|
||||
for path, label in endpoints:
|
||||
cold = req(path)
|
||||
warm = req(path)
|
||||
cold_ms = cold["elapsed"] * 1000
|
||||
warm_ms = warm["elapsed"] * 1000
|
||||
size_kb = len(cold["text"]) / 1024
|
||||
notes = []
|
||||
if cold_ms > 3000: notes.append(f"cold={cold_ms:.0f}ms")
|
||||
if warm_ms > 2000: notes.append(f"warm={warm_ms:.0f}ms")
|
||||
if size_kb > 200: notes.append(f"size={size_kb:.0f}KB")
|
||||
if not notes: notes.append(f"cold={cold_ms:.0f}ms warm={warm_ms:.0f}ms")
|
||||
ok(f"{label} ({path}) — {'; '.join(notes)}, {size_kb:.0f}KB")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER 8: SECURITY
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
SECRET_PATTERNS = [
|
||||
(r'eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+', 'JWT token'),
|
||||
(r'sk-[A-Za-z0-9]{20,}', 'API key'),
|
||||
(r'ghp_[A-Za-z0-9]{36}', 'GitHub PAT'),
|
||||
(r'-----BEGIN (RSA |EC )?PRIVATE KEY-----', 'Private key'),
|
||||
]
|
||||
|
||||
def layer_security():
|
||||
sec("LAYER 8: SECURITY")
|
||||
sub("Secrets in HTML")
|
||||
pages = ["/", "/learn", "/verify/ABC123"]
|
||||
for path in pages:
|
||||
html = req(path)["text"]
|
||||
for pat, desc in SECRET_PATTERNS:
|
||||
matches = re.findall(pat, html)
|
||||
if matches:
|
||||
fail(f"{path} — possible {desc} leak: {matches[0][:20]}...")
|
||||
else:
|
||||
ok(f"{path} — no {desc} leak")
|
||||
|
||||
sub("Security headers")
|
||||
r = req("/")
|
||||
hdrs = r["headers"]
|
||||
checks = [
|
||||
("Content-Type", lambda v: "text/html" in v.lower() if v else False),
|
||||
("X-Content-Type-Options", lambda v: v == "nosniff" if v else False),
|
||||
]
|
||||
for hname, check in checks:
|
||||
val = hdrs.get(hname)
|
||||
if val and check(val):
|
||||
ok(f"{hname}: {val}")
|
||||
elif val:
|
||||
warn(f"{hname}: {val} (unexpected)")
|
||||
else:
|
||||
warn(f"{hname}: MISSING")
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# LAYER B: BROWSER (Playwright) — Learn page + Login
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def layer_browser():
|
||||
sec("LAYER B: BROWSER — Playwright Visual/Interaction Tests")
|
||||
try:
|
||||
from playwright.sync_api import sync_playwright
|
||||
except ImportError:
|
||||
warn("Playwright not available — skipping browser layer")
|
||||
return
|
||||
|
||||
token = get_token()
|
||||
if not token:
|
||||
fail("No auth token — skipping browser auth tests")
|
||||
return
|
||||
|
||||
TIMEOUT_MS = 30000
|
||||
BASE = BASE_URL.rstrip("/")
|
||||
|
||||
with sync_playwright() as p:
|
||||
browser = p.chromium.launch(headless=True)
|
||||
|
||||
# Test 1: Learn catalog page (unauthenticated)
|
||||
sub("Browser: Learn catalog page (public)")
|
||||
context = browser.new_context(viewport={"width": 390, "height": 844}) # iPhone 14
|
||||
page = context.new_page()
|
||||
try:
|
||||
resp = page.goto(f"{BASE}/learn", timeout=TIMEOUT_MS, wait_until="domcontentloaded")
|
||||
page.wait_for_timeout(3000) # Allow React hydration
|
||||
html = page.content()
|
||||
console_errors = []
|
||||
page.on("console", lambda msg: console_errors.append(msg.text) if msg.type == "error" else None)
|
||||
|
||||
if resp and resp.ok:
|
||||
ok("Learn page — HTTP 200 (Playwright)")
|
||||
else:
|
||||
warn(f"Learn page — HTTP {resp.status if resp else '?'}")
|
||||
|
||||
# Check for JS errors
|
||||
js_errors = [e for e in console_errors if "Error" in e or "error" in e.lower()]
|
||||
if js_errors:
|
||||
for e in js_errors[:3]:
|
||||
warn(f"Learn page — JS console error: {e[:100]}")
|
||||
else:
|
||||
ok("Learn page — no JS console errors")
|
||||
|
||||
# Check rendered content
|
||||
text = page.inner_text("body")
|
||||
if "Learn" in text or "Course" in text or "Atomic" in text:
|
||||
ok("Learn page — rendered content visible")
|
||||
else:
|
||||
warn("Learn page — may be empty/loading (text not found)")
|
||||
|
||||
ok(f"Learn page — title: {page.title()}")
|
||||
|
||||
except Exception as e:
|
||||
fail(f"Learn page browser test — {e}")
|
||||
finally:
|
||||
context.close()
|
||||
|
||||
# Test 2: Authenticated course detail page
|
||||
sub("Browser: Course detail (authenticated via localStorage)")
|
||||
context2 = browser.new_context(viewport={"width": 390, "height": 844})
|
||||
page2 = context2.new_page()
|
||||
try:
|
||||
# Inject JWT into localStorage before page loads
|
||||
page2.goto(f"{BASE}/", timeout=TIMEOUT_MS, wait_until="domcontentloaded")
|
||||
page2.evaluate(f"localStorage.setItem('flh_token', '{token}')")
|
||||
page2.wait_for_timeout(500)
|
||||
|
||||
# Navigate to learn page (should pick up auth)
|
||||
resp2 = page2.goto(f"{BASE}/learn", timeout=TIMEOUT_MS, wait_until="domcontentloaded")
|
||||
page2.wait_for_timeout(3000)
|
||||
|
||||
if resp2 and resp2.ok:
|
||||
ok("Learn page — loaded with auth (Playwright)")
|
||||
else:
|
||||
warn(f"Learn page with auth → HTTP {resp2.status if resp2 else '?'}")
|
||||
|
||||
text2 = page2.inner_text("body")
|
||||
if "Atomic Habits" in text2 or "atomic" in text2.lower():
|
||||
ok("Learn page — course content visible after auth")
|
||||
else:
|
||||
warn("Learn page — auth didn't surface course content")
|
||||
|
||||
# Check local storage
|
||||
stored_token = page2.evaluate("localStorage.getItem('flh_token')")
|
||||
if stored_token:
|
||||
ok("Auth token persists in localStorage")
|
||||
else:
|
||||
warn("Auth token not found in localStorage")
|
||||
|
||||
except Exception as e:
|
||||
fail(f"Course detail browser test — {e}")
|
||||
finally:
|
||||
context2.close()
|
||||
|
||||
# Test 3: Certificate verification page
|
||||
sub("Browser: Certificate verification page")
|
||||
context3 = browser.new_context(viewport={"width": 390, "height": 844})
|
||||
page3 = context3.new_page()
|
||||
try:
|
||||
resp3 = page3.goto(f"{BASE}/verify/ABC123", timeout=TIMEOUT_MS, wait_until="domcontentloaded")
|
||||
page3.wait_for_timeout(3000)
|
||||
if resp3 and resp3.ok:
|
||||
ok("Verify page — HTTP 200 (Playwright)")
|
||||
else:
|
||||
warn(f"Verify page → HTTP {resp3.status if resp3 else '?'}")
|
||||
text3 = page3.inner_text("body")
|
||||
ok(f"Verify page — {len(text3)} chars rendered")
|
||||
except Exception as e:
|
||||
fail(f"Verify page browser test — {e}")
|
||||
finally:
|
||||
context3.close()
|
||||
|
||||
browser.close()
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# MAIN
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
def main():
|
||||
global BASE_URL, CI_EXIT
|
||||
parser = argparse.ArgumentParser(description="Falah Mobile Production QA")
|
||||
parser.add_argument("--url", default=BASE_URL)
|
||||
parser.add_argument("--ci-exit", action="store_true")
|
||||
parser.add_argument("--skip-browser", action="store_true", help="Skip Playwright browser tests")
|
||||
args = parser.parse_args()
|
||||
BASE_URL = args.url.rstrip("/")
|
||||
CI_EXIT = args.ci_exit
|
||||
|
||||
print(f"{'='*60}")
|
||||
print(f" Falah Mobile — PRODUCTION QA SUITE")
|
||||
print(f" URL: {BASE_URL}")
|
||||
print(f" Time: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
layers = [
|
||||
("SYSTEM", layer_system),
|
||||
("API", layer_api),
|
||||
("RENDER", layer_render),
|
||||
("SCENARIO", layer_scenario),
|
||||
("EDGE", layer_edge),
|
||||
("MOBILE", layer_mobile),
|
||||
("PERF", layer_perf),
|
||||
("SECURITY", layer_security),
|
||||
]
|
||||
if not args.skip_browser:
|
||||
layers.append(("BROWSER", layer_browser))
|
||||
|
||||
for name, fn in layers:
|
||||
try:
|
||||
fn()
|
||||
except Exception as e:
|
||||
print(f" {R}✗{N} Layer '{name}' crashed: {e}")
|
||||
import traceback; traceback.print_exc()
|
||||
|
||||
total = PASS + FAIL + WARN
|
||||
if FAIL == 0: grade = "A"
|
||||
elif FAIL <= 3: grade = "B"
|
||||
elif FAIL <= 10: grade = "C"
|
||||
else: grade = "D"
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f" QA RESULTS — Grade: {grade}")
|
||||
print(f"{'='*60}")
|
||||
print(f" {G}✓ Pass:{N} {PASS} {R}✗ Fail:{N} {FAIL} {Y}⚠ Warn:{N} {WARN} Total: {total}")
|
||||
|
||||
report = f"""# QA Test Report — Production
|
||||
**Date:** {time.strftime('%Y-%m-%d %H:%M:%S')}
|
||||
**URL:** {BASE_URL}
|
||||
|
||||
## Summary
|
||||
| ✅ Pass | ❌ Fail | ⚠️ Warn | Total | Grade |
|
||||
|:------:|:------:|:------:|:----:|:-----:|
|
||||
| {PASS} | {FAIL} | {WARN} | {total} | **{grade}** |
|
||||
|
||||
## Details
|
||||
| Result | Message |
|
||||
|:------|:--------|
|
||||
"""
|
||||
for result, msg in REPORT:
|
||||
icon = {"PASS": "✅", "FAIL": "❌", "WARN": "⚠️"}.get(result, "•")
|
||||
report += f"| {icon} | {msg} |\n"
|
||||
|
||||
report += f"\n## Re-run\n```bash\n/usr/bin/python3.12 {__file__}\n```"
|
||||
report_file = f"qa-prod-report-{time.strftime('%Y%m%d_%H%M%S')}.md"
|
||||
with open(report_file, "w") as f:
|
||||
f.write(report)
|
||||
print(f" Report: {report_file}")
|
||||
|
||||
if CI_EXIT and FAIL > 0:
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+32
-40
@@ -51,52 +51,42 @@ ERROR_PATTERNS_JS = json.dumps([
|
||||
|
||||
|
||||
def authenticate(browser):
|
||||
"""Login with demo credentials and return authenticated context + page."""
|
||||
"""Login via API and set token in localStorage, then return authenticated page."""
|
||||
context = browser.new_context(viewport=VIEWPORTS[viewport])
|
||||
page = context.new_page()
|
||||
page.on("pageerror", lambda err: None)
|
||||
|
||||
try:
|
||||
# Step 1: Go to homepage → login page
|
||||
# Step 1: Call login API to get a JWT token
|
||||
import urllib.request, json as jsonlib, ssl
|
||||
login_url = f"{BASE_URL.rstrip('/')}/api/auth/login"
|
||||
req = urllib.request.Request(
|
||||
login_url,
|
||||
data=jsonlib.dumps({"email": "demo@falahos.my", "password": "password123"}).encode(),
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36",
|
||||
"Accept": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
ctx = ssl.create_default_context()
|
||||
resp = urllib.request.urlopen(req, timeout=15, context=ctx)
|
||||
auth_data = jsonlib.loads(resp.read().decode())
|
||||
token = auth_data.get("token", "")
|
||||
if not token:
|
||||
fail("Login API returned no token")
|
||||
return None, None
|
||||
ok(f"Got API token: {token[:20]}...{token[-4:]}")
|
||||
|
||||
# Step 2: Navigate to any page and inject the token into localStorage
|
||||
page.goto(f"{BASE_URL.rstrip('/')}/", timeout=TIMEOUT_MS, wait_until="domcontentloaded")
|
||||
page.evaluate(f"localStorage.setItem('flh_token', '{token}')")
|
||||
page.wait_for_timeout(500)
|
||||
|
||||
# Step 3: Reload to let React AuthContext pick up the token
|
||||
page.goto(f"{BASE_URL.rstrip('/')}/", timeout=TIMEOUT_MS, wait_until="networkidle")
|
||||
page.wait_for_timeout(1000)
|
||||
|
||||
# Step 2: Click "Continue with Email"
|
||||
email_btn = page.locator("button:has-text('Continue with Email')")
|
||||
if email_btn.count() > 0:
|
||||
email_btn.click()
|
||||
page.wait_for_timeout(1500)
|
||||
|
||||
# Step 3: Click "Sign in" to get to the login form
|
||||
sign_in = page.locator("button:has-text('Sign in')")
|
||||
if sign_in.count() > 0:
|
||||
sign_in.click()
|
||||
page.wait_for_timeout(1500)
|
||||
|
||||
# Step 4: Fill email
|
||||
email_input = page.locator("input[type='email']")
|
||||
if email_input.count() > 0:
|
||||
email_input.fill("demo@falahos.my")
|
||||
page.wait_for_timeout(200)
|
||||
|
||||
# Step 5: Fill password
|
||||
pw_input = page.locator("input[type='password']")
|
||||
if pw_input.count() > 0:
|
||||
pw_input.fill("password123")
|
||||
page.wait_for_timeout(200)
|
||||
|
||||
# Step 6: Submit
|
||||
submit = page.locator("button[type='submit'], button:has-text('Sign In')")
|
||||
if submit.count() > 0:
|
||||
submit.first.click()
|
||||
page.wait_for_timeout(3000)
|
||||
|
||||
# Wait for network idle after login
|
||||
try:
|
||||
page.wait_for_load_state("networkidle", timeout=10000)
|
||||
except:
|
||||
pass
|
||||
page.wait_for_timeout(1000)
|
||||
page.wait_for_timeout(2000)
|
||||
|
||||
# Verify auth
|
||||
body_text = page.inner_text("body") or ""
|
||||
@@ -108,6 +98,8 @@ def authenticate(browser):
|
||||
return context, page
|
||||
except Exception as e:
|
||||
fail(f"Authentication failed: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
context.close()
|
||||
return None, None
|
||||
|
||||
|
||||
Reference in New Issue
Block a user