kunthawat/sales-trainer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1ad21cc8f185bf8282e098c8cf835ed37e17180d
sales-trainer/backend/tests/test_group_redaction.py
Macky 3c22d88bcd feat: demo SaaS + training flow security hardening (8/8 review gate passed) 
- Demo accounts: super_admin-only provisioning into isolated DEMO_ORG_ID tenant,
  30-day UTC trial on first login, revocable, one-time credential delivery via
  optional SES/webhook (never persisted). Adds boto3 dependency.
- Analytics/report/export/privacy: shared bounded scan budget across users/groups/
  sessions, tenant-consistent session/user/group joins, scalar-only CSV export
  (no nested persisted-value stringification).
- Ownership/tenant isolation: canonical owner-tenant predicate for list/read/chat;
  client sees is_owned only, never owner_user_id.
- Lifecycle/races: status transition validation, analyzing is an in-progress gate
  (no duplicate reanalysis), structured-ready publication, stale-variant revalidation.
- Auth/setup/consent/JWT/OAuth/config: fail-closed consent, bounded JWT lifetime,
  provider-subject atomic OAuth identity, repeated-secret rejection, strict Persona
  trait validation.
- Chat/session/privacy: pre-seller opener redaction, corrupt-session recovery,
  role-aware completed-chat dashboard routing.
- Frontend: Training→product→personas→practice flow, demo/role/demo guards,
  is_owned-based ownership display, 320×568 and 500×768 responsive E2E.
- 8 independent exact-five-key review scopes passed; backend 509, frontend 26,
  production build 1775 modules, isolated E2E 15.
2026-08-25 06:39:06 +07:00

568 lines
19 KiB
Python
Raw Blame History

"""Regression tests for group response redaction."""
from __future__ import annotations
import pytest
from app.services.groups import GroupStore
from scripts.mock_llm import MockLLM
def _headers(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def _setup_admin(user_store, login):
user_store.complete_setup(
"admin",
"admin@example.com",
"admin-ready-password",
accepted_terms=True,
accepted_terms_at="2026-08-13T00:00:00Z",
)
return login("admin", "admin-ready-password")["token"]
def _ready_group(client, token):
client.application.extensions["llm"] = MockLLM()
headers = _headers(token)
created = client.post(
"/api/groups", json={"product": "CRM", "segment": "SME"}, headers=headers
)
assert created.status_code == 201, created.get_json()
gid = created.get_json()["group"]["id"]
analyzed = client.post(f"/api/groups/{gid}/analyze", headers=headers)
assert analyzed.status_code == 200, analyzed.get_json()
return gid, analyzed.get_json()
def test_cross_tenant_group_ids_are_indistinguishable_from_missing_groups(
client, user_store, login
):
user_store.create_org("Organization B", org_id="org-b")
group_store = client.application.extensions["group_store"]
group = group_store.create(
org_id="org-b", creator_id="org-b-user", title="Shared tenant group"
)
private_group = group_store.create(
org_id="org-b",
creator_id="org-b-user",
owner_user_id="org-b-user",
title="Private tenant group",
)
admin = user_store.create_user(
org_id="org-default",
username="cross-tenant-group-reader",
password="cross-tenant-group-reader-password",
name="Cross Tenant Group Reader",
role="admin",
must_setup=False,
)
token = login(admin["username"], "cross-tenant-group-reader-password")["token"]
response = client.get(
f"/api/groups/{group['id']}", headers=_headers(token)
)
assert response.status_code == 404, response.get_json()
assert response.get_json() == {"error": "group not found"}
private_response = client.get(
f"/api/groups/{private_group['id']}", headers=_headers(token)
)
assert private_response.status_code == 404, private_response.get_json()
assert private_response.get_json() == {"error": "group not found"}
def test_admin_analyze_get_and_list_use_same_redaction(client, user_store, login):
_setup_admin(user_store, login)
admin = user_store.create_user(
org_id="org-default",
username="review-admin",
password="review-admin-password",
name="Review Admin",
role="admin",
email="review-admin@example.com",
must_setup=False,
)
admin_token = login(admin["username"], "review-admin-password")["token"]
gid, analyzed = _ready_group(client, admin_token)
headers = _headers(admin_token)
analyze_group = analyzed["group"]
get_group = client.get(f"/api/groups/{gid}", headers=headers).get_json()["group"]
listed = client.get(f"/api/groups/{gid}/personas", headers=headers).get_json()["personas"]
assert analyze_group.get("sales_kit") is None
assert analyze_group.get("report") is None
assert "creator_id" not in analyze_group
assert "owner_user_id" not in analyze_group
assert get_group.get("sales_kit") is None
assert get_group.get("report") is None
assert "creator_id" not in get_group
assert "owner_user_id" not in get_group
for persona in (analyze_group["personas"][0], get_group["personas"][0], listed[0]):
assert "pains" not in persona
assert "tolerance" not in persona
assert "negotiation_levers" not in persona
def test_admin_response_does_not_expose_future_secret_fields(
client, user_store, login
):
super_token = _setup_admin(user_store, login)
gid, analyzed = _ready_group(client, super_token)
source_id = analyzed["personas"][0]["id"]
stored = user_store # keep fixture dependency explicit for isolated data
group_store = client.application.extensions["group_store"]
group = group_store.get(gid)
group["personas"][0]["future_hidden_signal"] = {"secret": "must-not-leak"}
group_store.groups.replace(gid, group)
admin = stored.create_user(
org_id="org-default",
username="future-field-admin",
password="future-field-admin-password",
name="Future Field Admin",
role="admin",
email="future-field-admin@example.com",
must_setup=False,
)
token = login(admin["username"], "future-field-admin-password")["token"]
responses = [
client.get(f"/api/groups/{gid}", headers=_headers(token)).get_json()["group"]["personas"][0],
client.get(
f"/api/groups/{gid}/personas", headers=_headers(token)
).get_json()["personas"][0],
client.get(
f"/api/groups/{gid}/personas/{source_id}", headers=_headers(token)
).get_json()["persona"],
]
assert all("future_hidden_signal" not in persona for persona in responses)
def test_group_metadata_storage_error_is_not_returned_to_client(
client, user_store, login, monkeypatch
):
super_token = _setup_admin(user_store, login)
group_store = client.application.extensions["group_store"]
group = group_store.create(
org_id="org-default", creator_id="admin", title="Storage error group"
)
def fail_update(*args, **kwargs):
raise ValueError("/private/storage/path: internal detail")
monkeypatch.setattr(group_store, "update", fail_update)
response = client.patch(
f"/api/groups/{group['id']}/metadata",
json={"title": "New title"},
headers=_headers(super_token),
)
assert response.status_code == 400, response.get_json()
assert response.get_json() == {"error": "group update failed"}
assert "/private/storage/path" not in response.get_data(as_text=True)
def test_tenant_admin_group_view_hides_uploaded_source_payload(
client, user_store, login
):
_setup_admin(user_store, login)
group_store = client.application.extensions["group_store"]
group = group_store.create(
org_id="org-default", creator_id="admin", title="Source protected group"
)
group_store.update(
group["id"],
input={
"product": "CRM",
"segment": "SME",
"description": "safe summary",
"channel": "social",
"language": "th",
"files": ["internal.txt"],
"file_text": "provider-key=/private/path/source-secret",
},
)
admin = user_store.create_user(
org_id="org-default",
username="source-protected-admin",
password="source-protected-password",
name="Source Protected Admin",
role="admin",
must_setup=False,
)
token = login(admin["username"], "source-protected-password")["token"]
response = client.get(
f"/api/groups/{group['id']}", headers=_headers(token)
)
assert response.status_code == 200, response.get_json()
source = response.get_json()["group"]["input"]
assert source["product"] == "CRM"
assert "files" not in source
assert "file_text" not in source
assert "source-secret" not in response.get_data(as_text=True)
def test_super_admin_receives_full_group_view(client, user_store, login):
super_token = _setup_admin(user_store, login)
gid, analyzed = _ready_group(client, super_token)
assert analyzed["group"]["personas"][0]["pains"]
assert client.get(
f"/api/groups/{gid}/personas", headers=_headers(super_token)
).get_json()["personas"][0]["tolerance"]
def test_trainee_receives_revealable_persona_only(client, user_store, login):
super_token = _setup_admin(user_store, login)
gid, _ = _ready_group(client, super_token)
user = user_store.create_user(
org_id="org-default",
username="trainee-redaction",
password="trainee-redaction-password",
name="Trainee",
role="user",
must_setup=False,
)
trainee_token = login(user["username"], "trainee-redaction-password")["token"]
persona = client.get(
f"/api/groups/{gid}/personas", headers=_headers(trainee_token)
).get_json()["personas"][0]
assert "name" in persona
assert "profession" in persona
assert "pains" not in persona
assert "income" not in persona
assert "personality" not in persona
assert "tolerance" not in persona
def test_trainee_group_view_hides_uploaded_source_data(
client, user_store, login
):
super_token = _setup_admin(user_store, login)
gid, _ = _ready_group(client, super_token)
group_store = client.application.extensions["group_store"]
group = group_store.get(gid)
group_store.update(
gid,
input={
"product": "CRM",
"segment": "SME",
"description": "safe product summary",
"files": ["internal-brief.txt"],
"file_text": "CONFIDENT INTERNAL SOURCE CONTENT",
},
)
user = user_store.create_user(
org_id="org-default",
username="trainee-source-redaction",
password="trainee-source-redaction-password",
name="Trainee",
role="user",
must_setup=False,
)
trainee_token = login(user["username"], "trainee-source-redaction-password")["token"]
response = client.get(
f"/api/groups/{gid}", headers=_headers(trainee_token)
)
assert response.status_code == 200, response.get_json()
group_view = response.get_json()["group"]
assert group_view["input"]["product"] == "CRM"
assert "files" not in group_view["input"]
assert "file_text" not in group_view["input"]
def test_user_group_index_is_owner_scoped_and_revealable_only(tmp_path):
store = GroupStore(tmp_path)
private = store.create(
org_id="org-default", creator_id="owner-1", title="Private group"
)
store.update(
private["id"],
status="ready",
owner_user_id="owner-1",
input={"product": "CRM", "file_text": "private source"},
sales_kit={"productName": "CRM", "secret": "must-not-leak"},
report={"internal": "must-not-leak"},
personas=[
{
"id": "private-persona",
"name": "Private Persona",
"tier": "B",
"pains": [{"description": "hidden pain"}],
"future_hidden_signal": "must-not-leak",
}
],
)
shared = store.create(
org_id="org-default", creator_id="admin-1", title="Shared group"
)
store.update(
shared["id"],
status="ready",
input={"product": "Shared CRM"},
sales_kit={"productName": "Test product"},
report={"summary": "Test report"},
personas=[{"id": "shared-persona", "name": "Shared Persona", "tier": "A"}],
)
other_view = store.list_visible_to(
role="user", org_id="org-default", user_id="other-user"
)
assert {group["id"] for group in other_view} == {shared["id"]}
owner_view = store.list_visible_to(
role="user", org_id="org-default", user_id="owner-1"
)
private_view = next(group for group in owner_view if group["id"] == private["id"])
assert private_view["sales_kit"] is None
assert private_view["report"] is None
assert private_view["input"] == {"product": "CRM"}
assert private_view["personas"] == [
{
"id": "private-persona",
"name": "Private Persona",
"tier": "B",
"initiation_mode": "customer",
"channel": "facebook",
"profession": "",
"age_group": "",
"location": "",
"product_context": "",
}
]
def test_group_index_marks_user_owned_products_without_exposing_owner_id(
client, user_store, login
):
login_value = "owned-product-" + "index-password"
user = user_store.create_user(
org_id="org-default",
username="owned-product-index-user",
password=login_value,
name="Owned Product User",
role="user",
must_setup=False,
)
token = login(user["username"], login_value)["token"]
groups = client.application.extensions["group_store"]
private = groups.create(
org_id="org-default",
creator_id=user["id"],
owner_user_id=user["id"],
visibility="private",
title="Private product",
)
groups.update(
private["id"],
status="ready",
input={"product": "Private CRM"},
sales_kit={"productName": "Test product"},
report={"summary": "Test report"},
personas=[{"id": "private-persona", "name": "Private Persona"}],
)
failed = groups.create(
org_id="org-default",
creator_id=user["id"],
owner_user_id=user["id"],
visibility="private",
title="Failed product",
)
groups.update(
failed["id"],
status="failed",
error="analysis_failed",
input={"product": "Failed CRM"},
)
shared = groups.create(
org_id="org-default",
creator_id="admin",
title="Shared product",
)
groups.update(
shared["id"],
status="ready",
input={"product": "Shared POS"},
sales_kit={"productName": "Test product"},
report={"summary": "Test report"},
personas=[{"id": "shared-persona", "name": "Shared Persona"}],
)
response = client.get("/api/groups", headers=_headers(token))
assert response.status_code == 200, response.get_json()
rows = {row["id"]: row for row in response.get_json()["groups"]}
assert rows[private["id"]]["is_owned"] is True
assert rows[failed["id"]]["is_owned"] is True
assert rows[failed["id"]]["status"] == "failed"
assert rows[shared["id"]]["is_owned"] is False
assert "owner_user_id" not in rows[private["id"]]
private_response = client.get(
f"/api/groups/{private['id']}", headers=_headers(token)
)
assert private_response.status_code == 200, private_response.get_json()
private_payload = private_response.get_json()["group"]
assert private_payload["is_owned"] is True
assert "owner_user_id" not in private_payload
@pytest.mark.parametrize("status", ["draft", "failed"])
def test_owner_can_open_non_ready_private_product_for_retry(
client, user_store, login, status
):
password = "retry-product-owner-password"
user = user_store.create_user(
org_id="org-default",
username=f"retry-product-owner-{status}",
password=password,
name="Retry Product Owner",
role="user",
must_setup=False,
)
token = login(user["username"], password)["token"]
groups = client.application.extensions["group_store"]
product = groups.create(
org_id="org-default",
creator_id=user["id"],
owner_user_id=user["id"],
visibility="private",
title=f"{status.title()} product",
)
groups.update(
product["id"],
status=status,
error="analysis_failed" if status == "failed" else None,
input={"product": "Retry CRM"},
personas=[{"id": "stale-retry-persona", "name": "Stale Persona", "opener": "secret"}],
)
response = client.get(f"/api/groups/{product['id']}", headers=_headers(token))
assert response.status_code == 200, response.get_json()
payload = response.get_json()["group"]
assert payload["status"] == status
assert payload["is_owned"] is True
assert payload["personas"] == []
assert "owner_user_id" not in payload
def test_trainee_cannot_analyze(client, user_store, login):
super_token = _setup_admin(user_store, login)
gid, _ = _ready_group(client, super_token)
user = user_store.create_user(
org_id="org-default",
username="trainee-no-analyze",
password="trainee-no-analyze-password",
name="Trainee",
role="user",
must_setup=False,
)
token = login(user["username"], "trainee-no-analyze-password")["token"]
response = client.post(f"/api/groups/{gid}/analyze", headers=_headers(token))
assert response.status_code == 403, response.get_json()
def test_trainee_variant_response_is_revealable_only(client, user_store, login):
super_token = _setup_admin(user_store, login)
gid, analyzed = _ready_group(client, super_token)
user = user_store.create_user(
org_id="org-default",
username="trainee-variant",
password="trainee-variant-password",
name="Trainee",
role="user",
must_setup=False,
)
trainee_token = login(user["username"], "trainee-variant-password")["token"]
source_id = analyzed["personas"][0]["id"]
response = client.post(
f"/api/groups/{gid}/personas/{source_id}/variant",
headers=_headers(trainee_token),
)
assert response.status_code == 201, response.get_json()
persona = response.get_json()["persona"]
assert "name" in persona
assert "pains" not in persona
assert "income" not in persona
assert "tolerance" not in persona
def test_trainee_variant_is_stored_in_owner_private_group(
client, user_store, login
):
super_token = _setup_admin(user_store, login)
gid, analyzed = _ready_group(client, super_token)
user = user_store.create_user(
org_id="org-default",
username="private-variant-owner",
password="private-variant-owner-password",
name="Private Variant Owner",
role="user",
must_setup=False,
)
trainee_token = login(user["username"], "private-variant-owner-password")["token"]
headers = _headers(trainee_token)
source_id = analyzed["personas"][0]["id"]
response = client.post(
f"/api/groups/{gid}/personas/{source_id}/variant",
headers=headers,
)
assert response.status_code == 201, response.get_json()
payload = response.get_json()
variant = payload["persona"]
private_gid = payload["group_id"]
assert private_gid != gid
source_group = client.application.extensions["group_store"].get(gid)
private_group = client.application.extensions["group_store"].get(private_gid)
assert private_group["sales_kit"] == source_group["sales_kit"]
shared_personas = client.get(
f"/api/groups/{gid}/personas", headers=headers
).get_json()["personas"]
assert variant["id"] not in {p["id"] for p in shared_personas}
private_response = client.get(
f"/api/groups/{private_gid}/personas", headers=headers
)
assert private_response.status_code == 200, private_response.get_json()
private_personas = private_response.get_json()["personas"]
assert [p["id"] for p in private_personas] == [variant["id"]]
other_user = user_store.create_user(
org_id="org-default",
username="private-variant-other-user",
password="private-variant-other-password",
name="Other User",
role="user",
must_setup=False,
)
other_token = login(
other_user["username"], "private-variant-other-password"
)["token"]
other_headers = _headers(other_token)
assert client.get(
f"/api/groups/{private_gid}/personas", headers=other_headers
).status_code == 403
listed_ids = {
group["id"]
for group in client.get("/api/groups", headers=other_headers).get_json()["groups"]
}
assert private_gid not in listed_ids
Reference in New Issue View Git Blame Copy Permalink