62 lines
4.2 KiB
Python
62 lines
4.2 KiB
Python
"""Тест флоу Autopilot $99 (shkor.com/start): KYC → оплата → деплой. Портал и Didit — фейки.
|
|
python3 sites/shkor-com/test_autopilot.py
|
|
"""
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlsplit, parse_qs
|
|
S={"calls":[], "paid":False}
|
|
class H(BaseHTTPRequestHandler):
|
|
def r(s,o):
|
|
b=json.dumps(o).encode(); s.send_response(200); s.send_header("Content-Type","application/json"); s.end_headers(); s.wfile.write(b)
|
|
def do_POST(s):
|
|
d=json.loads(s.rfile.read(int(s.headers["Content-Length"])))
|
|
p=urlsplit(s.path).path; S["calls"].append((p,d))
|
|
if p=="/api/onboarding/start": return s.r({"pending_id":"pid123"})
|
|
if p=="/api/onboarding/save": return s.r({"ok":True})
|
|
if p=="/api/onboarding/checkout": return s.r({"url":"https://checkout.stripe.test/x"})
|
|
if p=="/__pay": S["paid"]=True; return s.r({})
|
|
if p=="/__calls": return s.r(S["calls"])
|
|
def do_GET(s):
|
|
if S["paid"]: return s.r({"status":"ready","tenant_slug":"t1","portal_url":"https://t1.cil.by","agent":{"name":"a","status":"running"},"panel":{"url":"https://a.ui.shkor.com/","user":"a","password":"p"}})
|
|
return s.r({"status":"pending"})
|
|
def log_message(s,*a): pass
|
|
|
|
|
|
import os, sys, sqlite3, importlib.util, threading
|
|
threading.Thread(target=ThreadingHTTPServer(("127.0.0.1",18899),H).serve_forever,daemon=True).start()
|
|
import tempfile
|
|
TMP=tempfile.mkdtemp()
|
|
os.environ.update(SHKOR_DB=TMP+"/t.db", PORTAL_URL="http://127.0.0.1:18899", TG_ENV_FILE="/nonexistent")
|
|
spec=importlib.util.spec_from_file_location("api",os.path.join(os.path.dirname(os.path.abspath(__file__)),"api.py")); api=importlib.util.module_from_spec(spec); spec.loader.exec_module(api)
|
|
st={"s":"Not Started"}
|
|
api.didit_session=lambda wf,v,cb: {"session_id":"sid1","url":"https://verify.didit.me/x","status":"Not Started"}
|
|
api.didit_status=lambda sid: st["s"]
|
|
ok=0; bad=0
|
|
def eq(a,b,m):
|
|
global ok,bad
|
|
if a==b: ok+=1
|
|
else: bad+=1; print("FAIL",m,a,"!=",b)
|
|
code,r=api.ap_start({"name":"","email":"x@y.com","consent":True},"1.1.1.1"); eq(code,400,"name required")
|
|
code,r=api.ap_start({"name":"A B","email":"bad","consent":True},"1.1.1.1"); eq(code,400,"email")
|
|
code,r=api.ap_start({"name":"A B","email":"a@b.com"},"1.1.1.1"); eq(code,400,"consent")
|
|
code,r=api.ap_start({"name":"Ann Bee","email":"A@B.com","company":"Acme Co","consent":True},"1.1.1.1"); eq(code,200,"start"); t=r["token"]
|
|
eq(r["kyc_url"],"https://verify.didit.me/x","kyc url")
|
|
code,r=api.ap_status(t); eq(r["stage"],"kyc","stage kyc")
|
|
code,r=api.ap_checkout({"t":t}); eq(code,403,"checkout blocked before kyc")
|
|
st["s"]="Declined"; code,r=api.ap_status(t); eq(r["stage"],"declined","declined"); eq(r["kyc_url"],"","no url when final")
|
|
# new order approved
|
|
st["s"]="Not Started"; code,r=api.ap_start({"name":"Ann Bee","email":"a@b.com","company":"Acme Co","consent":True},"2.2.2.2"); t=r["token"]
|
|
st["s"]="Approved"; code,r=api.ap_status(t); eq(r["stage"],"pay","pay stage")
|
|
code,r=api.ap_checkout({"t":t}); eq(code,200,"checkout ok"); eq(r["url"],"https://checkout.stripe.test/x","stripe url")
|
|
code,r=api.ap_checkout({"t":t}); eq(code,200,"checkout again reuses pending")
|
|
code,r=api.ap_status(t); eq(r["stage"],"pay","still pay before payment")
|
|
import urllib.request
|
|
def call(p): return json.loads(urllib.request.urlopen(urllib.request.Request("http://127.0.0.1:18899"+p,data=b"{}",headers={"Content-Type":"application/json"})).read())
|
|
calls=call("/__calls"); starts=[c for c in calls if c[0]=="/api/onboarding/start"]; eq(len(starts),1,"one pending created")
|
|
eq(starts[0][1],{"email":"a@b.com","plan":"autopilot"},"plan autopilot")
|
|
co=[c for c in calls if c[0]=="/api/onboarding/checkout"][0][1]; eq(co["return_url"],f"https://shkor.com/start/status/?t={t}","return url")
|
|
sv=[c for c in calls if c[0]=="/api/onboarding/save"][0][1]; eq(sv["agent_name"].startswith("acme-co-"),True,"agent name")
|
|
call("/__pay"); code,r=api.ap_status(t); eq(r["stage"],"ready","ready"); eq(r["panel"]["user"],"a","panel creds")
|
|
c=sqlite3.connect(TMP+"/t.db"); eq(c.execute("select paid_notified from ap_orders where token=?",(t,)).fetchone()[0],1,"paid flag")
|
|
code,r=api.ap_status("nope-nope-nope"); eq(code,404,"404")
|
|
print(f"{ok} ok, {bad} fail"); sys.exit(1 if bad else 0)
|