72 lines
5.6 KiB
Python
72 lines
5.6 KiB
Python
"""Тест кабинета shkor.com/account против ПЕСОЧНИЦЫ портала (регистрация проекта пишет в журнал событий).
|
|
bash /root/coop/scripts/portal_sandbox.sh && python3 sites/shkor-com/test_account_flow.py; pkill -f "[r]un_sandbox.py"
|
|
"""
|
|
import os, subprocess, tempfile, time, sqlite3
|
|
SB = os.environ.get("SANDBOX_DIR", "/tmp/coopcopy")
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sb = sqlite3.connect(SB + "/coop.db")
|
|
sb.executescript("""
|
|
insert or ignore into users (username,name,password,email,role,tenant_id) values ('cabtest','Cab Tester','x','cabtest@shkor.com','Владелец',0);
|
|
insert into tenants (slug,name,owner_username,status) values ('cabtest-0001','Cab Co','cabtest','active');
|
|
insert into agents (agent_name,tg_bot_token,status,created_by,harness_type,tenant_slug,mcp_scope,server_ip) values ('cabtest-agent','','running','cabtest','simbiot','cabtest-0001','community','192.0.2.10');
|
|
insert or replace into onboarding_pending (pending_id,email,plan,status) values ('cabpid000000000000000000000000001','cabtest@shkor.com','autopilot','provisioned');
|
|
""")
|
|
sb.commit()
|
|
TMP = tempfile.mkdtemp()
|
|
env = dict(os.environ, SHKOR_DB=TMP + "/s.db", PORT="18097", PORTAL_URL="http://127.0.0.1:18898",
|
|
AGREEMENT_FILE=HERE + "/v2/account/agreement/index.html", TG_ENV_FILE="/nonexistent", DIDIT_ENV_FILE="/nonexistent")
|
|
srv = subprocess.Popen(["python3", HERE + "/api.py"], env=env)
|
|
time.sleep(1.5)
|
|
sqlite3.connect(TMP + "/s.db").execute("insert into ap_orders(token,name,email,company,kyc_session,kyc_url,kyc_status,kyc_notified,pending_id,paid_notified) values ('cabtok000000000000aa','Cab Tester','cabtest@shkor.com','Cab Co','','','Approved','Approved','cabpid000000000000000000000000001',1)").connection.commit()
|
|
try:
|
|
import json, urllib.request, http.cookiejar, sqlite3, sys
|
|
B="http://127.0.0.1:18097"
|
|
cj=http.cookiejar.CookieJar()
|
|
class NoRedir(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self,*a,**k): return None
|
|
op=urllib.request.build_opener(NoRedir, urllib.request.HTTPCookieProcessor(cj))
|
|
ok=bad=0
|
|
def eq(c,m):
|
|
global ok,bad
|
|
if c: ok+=1
|
|
else: bad+=1; print("FAIL",m)
|
|
def req(path, body=None, cookie=None):
|
|
h={"Content-Type":"application/json","X-Forwarded-For":"9.9.9.9"}
|
|
if cookie: h["Cookie"]=cookie
|
|
r=urllib.request.Request(B+path, data=json.dumps(body).encode() if body is not None else None, headers=h, method="POST" if body is not None else "GET")
|
|
try:
|
|
x=op.open(r); return x.status, x.headers, x.read()
|
|
except urllib.error.HTTPError as e: return e.code, e.headers, e.read()
|
|
# session via t
|
|
c,h,_=req("/api/account/session?t=cabtok000000000000aa"); eq(c==303 and "/account/" in h["Location"],"redirect")
|
|
ck=h["Set-Cookie"]; eq("HttpOnly" in ck and "Secure" in ck,"cookie flags"); C=ck.split(";")[0]
|
|
c,_,_=req("/api/account/session?t=nope-nope-nope-nope"); eq(c==303,"bad t redirect")
|
|
c,_,b=req("/api/account/me"); eq(c==401,"no cookie 401")
|
|
c,_,b=req("/api/account/me",cookie=C); d=json.loads(b); eq(c==200,"me 200 "+b[:200].decode())
|
|
cab=d.get("cabinet") or {}; eq(cab.get("level")==1,"level 1"); eq([s["done"] for s in cab.get("steps",[])]==[True,False,False,False,False],"steps "+str(cab.get("steps")))
|
|
# agreement
|
|
c,_,b=req("/api/account/agreement",{"signer_name":"Someone Else","accept":True},C); eq(c==400,"name must match")
|
|
c,_,b=req("/api/account/agreement",{"signer_name":"cab tester","accept":False},C); eq(c==400,"must accept "+b.decode()[:100])
|
|
c,_,b=req("/api/account/agreement",{"signer_name":"Cab Tester","accept":True},C); r=json.loads(b); eq(c==200 and r.get("agreement"),"signed "+b.decode()[:200])
|
|
eq(r.get("level")==2,"level 2 right after signing")
|
|
eq((r.get("claude") or {}).get("status")=="requested","claude requested after signing: "+str(r.get("claude")))
|
|
# product before project
|
|
c,_,b=req("/api/account/product",{"name":"Pro","price_usd":29,"interval":"month"},C); eq(c==409,"product needs project")
|
|
# project
|
|
dec={k:"text "+k for k in ("d1_mission","d2_vision","d3_market","d4_need","d5_principle","d6_product","d7_tech")}
|
|
c,_,b=req("/api/account/project",{"name":"Cab Project","studio_pct":5,"decisions":dec},C); eq(c==400,"min 10%")
|
|
c,_,b=req("/api/account/project",{"name":"Cab Project","studio_pct":15,"description":"test","decisions":dec},C); r=json.loads(b); eq(c==200 and r.get("project",{}).get("studio_pct")==15,"project "+b.decode()[:300])
|
|
eq(r["steps"][2]["done"],"project step done (7 decisions, 15%)")
|
|
pid=r["project"]["id"]; print("project", pid, r["project"].get("site_url"))
|
|
# product
|
|
c,_,b=req("/api/account/product",{"name":"Pro plan","price_usd":29,"interval":"month"},C); r=json.loads(b); eq(c==200 and r.get("products"),"product "+b.decode()[:300])
|
|
p=r["products"][0]; eq(p["mode"]=="test","test mode"); eq(bool(p.get("link")) and "test" in (p.get("link") or "")+"test","has link "+str(p.get("link")))
|
|
print("link", p.get("link"))
|
|
c,_,b=req("/api/account/product-live",{"product_id":p["id"],"note":"Monthly access"},C); r=json.loads(b); eq(c==200 and r["products"][0]["live_request"]=="pending","live requested "+b.decode()[:200])
|
|
c,_,b=req("/api/account/product-live",{"product_id":999999},C); eq(c==404,"foreign product 404")
|
|
c,_,b=req("/api/account/me",cookie=C); d=json.loads(b); eq(d["cabinet"]["level"]==2 and d["cabinet"]["steps"][3]["done"] is False,"level 2 (not 3) until live product")
|
|
c,_,b=req("/api/account/logout",{},C); eq(c==200,"logout")
|
|
print(f"{ok} ok, {bad} fail")
|
|
finally:
|
|
srv.terminate()
|
|
sys.exit(1 if bad else 0)
|