3d5c81fbd7
P3a: org carries plan/seats/active/created_at; create_user enforces seats + rejects inactive org; verify blocks login for inactive orgs; PATCH /api/admin/orgs (super_admin) updates plan/seats/active with audit. Fixed verify swallowing its AuthError. P3b: export/token issues a 5-min HMAC one-time CSV link; export accepts ?token=. P3c: setup requires accepted_terms (consent stored); Setup.vue consent checkbox. All 8 backend suites pass. Rebuilt dist.
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
"""Test: scenario-based chat start (scenario determines who opens)."""
|
|
import os, sys, tempfile, warnings
|
|
from pathlib import Path
|
|
|
|
warnings.filterwarnings("ignore")
|
|
BACKEND = str(Path(__file__).resolve().parents[1])
|
|
sys.path.insert(0, BACKEND)
|
|
|
|
from app.factory import create_app
|
|
from app.config import Config
|
|
|
|
td = tempfile.mkdtemp()
|
|
Config.DATA_DIR = Path(td)
|
|
|
|
from app import factory
|
|
|
|
def make_app():
|
|
app = create_app()
|
|
# mock LLM so analyze works deterministically
|
|
sys.path.insert(0, BACKEND + "/scripts")
|
|
from mock_llm import MockLLM
|
|
app.extensions["llm"] = MockLLM()
|
|
return app
|
|
|
|
app = make_app()
|
|
C = app.test_client()
|
|
|
|
def login(u, p):
|
|
return C.post("/api/auth/login", json={"username": u, "password": p}).get_json()["token"]
|
|
|
|
AT = login("admin", "1234")
|
|
AH = {"Authorization": f"Bearer {AT}"}
|
|
C.post("/api/auth/setup", headers=AH, json={"username": "admin", "email": "a@b.co", "password": "newpass", "accepted_terms": True}).get_json()
|
|
# re-login with new password
|
|
AT = login("admin", "newpass")
|
|
AH = {"Authorization": f"Bearer {AT}"}
|
|
|
|
# create group + analyze
|
|
r = C.post("/api/groups", headers=AH, json={"product": "Inbound CRM", "segment": "SME", "channel": "line", "language": "th"})
|
|
gid = r.get_json()["group"]["id"]
|
|
C.post(f"/api/groups/{gid}/analyze", headers=AH)
|
|
personas = C.get(f"/api/groups/{gid}/personas", headers=AH).get_json()["personas"]
|
|
# build a ready group accepted by trainees
|
|
C.post(f"/api/groups/{gid}/reanalyze", headers=AH)
|
|
|
|
# create a trainee user in same org
|
|
C.post("/api/admin/users", headers=AH, json={"username": "t1", "name": "T", "password": "pppp", "role": "user"})
|
|
T = login("t1", "pppp")
|
|
TH = {"Authorization": f"Bearer {T}"}
|
|
|
|
pid = personas[0]["id"]
|
|
|
|
# social -> customer opens
|
|
r = C.post(f"/api/chat/{gid}/personas/{pid}/chat/start", headers=TH, json={"scenario": "social"})
|
|
assert r.status_code == 200, r.get_json()
|
|
s = r.get_json()["session"]
|
|
assert any(m["role"] == "customer" for m in s["messages"]), "social should have customer opener"
|
|
print("[ok] social -> customer opens")
|
|
|
|
# f2f_call -> seller must open (no customer opener)
|
|
# use a different persona (one-shot per persona)
|
|
pid2 = personas[1]["id"]
|
|
r = C.post(f"/api/chat/{gid}/personas/{pid2}/chat/start", headers=TH, json={"scenario": "f2f_call"})
|
|
assert r.status_code == 200, r.get_json()
|
|
s2 = r.get_json()["session"]
|
|
assert not any(m["role"] == "customer" for m in s2["messages"]), "f2f should NOT have customer opener"
|
|
assert any(m["role"] == "system" for m in s2["messages"]), "f2f should have a scenario system note"
|
|
print("[ok] f2f_call -> seller must open (system note present)")
|
|
|
|
# recontact -> system preamble about time-lapse + customer opens
|
|
pid3 = personas[2]["id"]
|
|
r = C.post(f"/api/chat/{gid}/personas/{pid3}/chat/start", headers=TH, json={"scenario": "recontact"})
|
|
assert r.status_code == 200, r.get_json()
|
|
s3 = r.get_json()["session"]
|
|
assert any(m["role"] == "customer" for m in s3["messages"]), "recontact should open w/ customer"
|
|
assert any(m["role"] == "system" and "เดือน" in m["text"] for m in s3["messages"]), "recontact system note (months)"
|
|
print("[ok] recontact -> time-lapse system note + customer opens")
|
|
|
|
print("ALL SCENARIO TESTS PASSED")
|