Senad Uka
646917c322
- Embeddable vanilla-JS chat widget (one script tag + tenant public key):
WebSocket, resumable sessions (24h localStorage token), honeypot + per-IP
and per-message rate limits, Bosnian UI (§7)
- Session auth (bcrypt + signed cookies), owner accounts + super-admins,
admin impersonation ('otvori kao salon', §11)
- Owner dashboard (§10, Bosnian): requests with same actions as email,
call/chat history with transcripts, usage bar; settings for profile,
working hours (manual + LLM paste-to-parse), services CRUD + 'Zalijepi
cjenovnik' LLM import with review step, scheduling (Google OAuth free/busy
read-only + calendar mappings + buffers, partner status), telephony
(forwarding MMI codes per operator, ring group, worker SIP creds + QR),
agent voice/price-mode/notes + widget snippet, notifications/SMS templates
- Super-admin panel (§11): tenant CRUD incl. plan/minutes/paid-until,
partner API config (encrypted secrets), number/SIM registry + assignment,
connectivity test (live availability + dry-run push), versioned prompt
template editor with publish/rollback, playground, usage overview, health
- Fixes: WAL + busy_timeout for sqlite tests, no awaits in WS disconnect
path (deadlock), template publish autoflush bug
- 76 tests green (incl. widget WS e2e booking flow); dashboard, widget and
admin verified live in a browser against Postgres
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Owner dashboard + super-admin panel over HTTP (M3)."""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from gogo.auth import hash_password
|
|
from gogo.domain import BookingStatus
|
|
from gogo.models import Admin, PromptTemplate, Service, Tenant, User
|
|
from tests.test_proposal_lifecycle import create as create_request
|
|
|
|
|
|
@pytest.fixture
|
|
async def owner_user(session, tenant):
|
|
user = User(
|
|
tenant_id=tenant.id,
|
|
email="vlasnica@salon-merima.ba",
|
|
password_hash=hash_password("tajna123"),
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
async def admin_user(session):
|
|
admin = Admin(email="admin@gogotelefon.ba", password_hash=hash_password("admin123"))
|
|
session.add(admin)
|
|
await session.commit()
|
|
return admin
|
|
|
|
|
|
async def login_owner(client, email="vlasnica@salon-merima.ba", password="tajna123"):
|
|
resp = await client.post("/login", data={"email": email, "password": password})
|
|
assert resp.status_code == 303
|
|
assert resp.headers["location"] == "/dash"
|
|
|
|
|
|
async def login_admin(client):
|
|
resp = await client.post(
|
|
"/admin/login", data={"email": "admin@gogotelefon.ba", "password": "admin123"}
|
|
)
|
|
assert resp.status_code == 303
|
|
|
|
|
|
# -- owner auth ---------------------------------------------------------------
|
|
|
|
|
|
async def test_dashboard_requires_login(gogo_client, session, tenant):
|
|
resp = await gogo_client.get("/dash")
|
|
assert resp.status_code == 303
|
|
assert resp.headers["location"] == "/login"
|
|
|
|
|
|
async def test_login_wrong_password(gogo_client, session, owner_user):
|
|
resp = await gogo_client.post(
|
|
"/login", data={"email": owner_user.email, "password": "pogresna"}
|
|
)
|
|
assert resp.status_code == 303
|
|
assert "/login" in resp.headers["location"]
|
|
|
|
|
|
async def test_login_and_requests_list(gogo_client, session, tenant, owner_user, emails, sms):
|
|
await create_request(session, tenant)
|
|
await session.commit()
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.get("/dash")
|
|
assert resp.status_code == 200
|
|
assert "Amra Hodžić" in resp.text
|
|
assert "na čekanju" in resp.text
|
|
|
|
|
|
async def test_dashboard_resolve_action(gogo_client, session, tenant, owner_user, emails, sms):
|
|
req, _ = await create_request(session, tenant)
|
|
await session.commit()
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.post(f"/dash/requests/{req.id}/resolve")
|
|
assert resp.status_code == 303
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.resolved_by_owner.value
|
|
assert sms == []
|
|
|
|
|
|
async def test_request_detail_confirm(gogo_client, session, tenant, owner_user, emails, sms):
|
|
req, _ = await create_request(session, tenant)
|
|
await session.commit()
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.get(f"/dash/requests/{req.id}")
|
|
assert "Potvrdi" in resp.text
|
|
resp = await gogo_client.post(f"/dash/requests/{req.id}/confirm/0")
|
|
assert resp.status_code == 303
|
|
await session.refresh(req)
|
|
assert req.status == BookingStatus.confirmed.value
|
|
assert len(sms) == 1
|
|
|
|
|
|
# -- settings -------------------------------------------------------------------
|
|
|
|
|
|
async def test_settings_save_hours(gogo_client, session, tenant, owner_user):
|
|
await login_owner(gogo_client)
|
|
data = {
|
|
"name": "Salon Merima",
|
|
"address": "Nova adresa 5",
|
|
"city": "Banja Luka",
|
|
"phone": "+38751000000",
|
|
"website": "",
|
|
"wh_mon": "09:00-13:00, 14:00-18:00",
|
|
"wh_tue": "9-18",
|
|
"wh_wed": "09:00-18:00",
|
|
"wh_thu": "09:00-18:00",
|
|
"wh_fri": "09:00-18:00",
|
|
"wh_sat": "09:00-14:00",
|
|
"wh_sun": "",
|
|
}
|
|
resp = await gogo_client.post("/dash/settings", data=data)
|
|
assert resp.status_code == 303
|
|
await session.refresh(tenant)
|
|
assert tenant.address == "Nova adresa 5"
|
|
assert tenant.working_hours["mon"] == [["09:00", "13:00"], ["14:00", "18:00"]]
|
|
assert tenant.working_hours["tue"] == [["09:00", "18:00"]]
|
|
assert tenant.working_hours["sun"] == []
|
|
|
|
|
|
async def test_settings_bad_hours_rejected(gogo_client, session, tenant, owner_user):
|
|
await login_owner(gogo_client)
|
|
data = {"name": "Salon Merima", "wh_mon": "od devet do šest"}
|
|
resp = await gogo_client.post("/dash/settings", data=data)
|
|
assert resp.status_code == 303
|
|
assert "err=" in resp.headers["location"]
|
|
|
|
|
|
async def test_services_crud(gogo_client, session, tenant, owner_user):
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.post(
|
|
"/dash/services",
|
|
data={"name": "Nova usluga", "duration_min": "60", "price_min": "40",
|
|
"price_max": "50", "agent_note": "napomena"},
|
|
)
|
|
assert resp.status_code == 303
|
|
svc = (
|
|
await session.execute(select(Service).where(Service.name == "Nova usluga"))
|
|
).scalar_one()
|
|
assert svc.duration_min == 60
|
|
assert svc.price_min == 40
|
|
|
|
resp = await gogo_client.post(f"/dash/services/{svc.id}/delete")
|
|
assert resp.status_code == 303
|
|
await session.refresh(svc)
|
|
assert svc.active is False
|
|
|
|
|
|
async def test_paste_price_list_with_scripted_llm(
|
|
gogo_client, session, tenant, owner_user, monkeypatch
|
|
):
|
|
from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse, set_llm
|
|
|
|
set_llm(
|
|
ScriptedLLM(
|
|
[
|
|
LLMResponse(
|
|
text="",
|
|
tool_uses=[
|
|
ToolUse(
|
|
id="t1",
|
|
name="save_services",
|
|
input={
|
|
"services": [
|
|
{"name": "Trajna", "duration_min": 90,
|
|
"price_min": 40, "price_max": 60},
|
|
]
|
|
},
|
|
)
|
|
],
|
|
stop_reason="tool_use",
|
|
)
|
|
]
|
|
)
|
|
)
|
|
try:
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.post(
|
|
"/dash/services/parse", data={"text": "Trajna 40-60KM"}
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "Provjeri i potvrdi" in resp.text
|
|
assert "Trajna" in resp.text
|
|
|
|
# confirm the parsed row → imported
|
|
resp = await gogo_client.post(
|
|
"/dash/services/import",
|
|
data={"count": "1", "row_0": "on", "name_0": "Trajna",
|
|
"duration_0": "90", "pmin_0": "40", "pmax_0": "60", "note_0": ""},
|
|
)
|
|
assert resp.status_code == 303
|
|
svc = (
|
|
await session.execute(select(Service).where(Service.name == "Trajna"))
|
|
).scalar_one()
|
|
assert svc.duration_min == 90
|
|
finally:
|
|
set_llm(None)
|
|
|
|
|
|
async def test_workers_add_and_qr(gogo_client, session, tenant, owner_user):
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.post("/dash/workers", data={"name": "Merima"})
|
|
assert resp.status_code == 303
|
|
from gogo.models import Worker
|
|
|
|
worker = (await session.execute(select(Worker))).scalar_one()
|
|
assert worker.name == "Merima"
|
|
assert worker.sip_username.startswith("salon-merima-")
|
|
|
|
resp = await gogo_client.get(f"/dash/workers/{worker.id}/qr.png")
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"] == "image/png"
|
|
assert resp.content[:8] == b"\x89PNG\r\n\x1a\n"
|
|
|
|
|
|
async def test_notifications_save(gogo_client, session, tenant, owner_user):
|
|
await login_owner(gogo_client)
|
|
resp = await gogo_client.post(
|
|
"/dash/notifications",
|
|
data={
|
|
"notify_emails": "a@b.ba\nc@d.ba",
|
|
"sms_request_received": "on",
|
|
"sms_confirmation": "Vaš termin: {usluga} {dan} u {vrijeme}h. {salon}",
|
|
},
|
|
)
|
|
assert resp.status_code == 303
|
|
await session.refresh(tenant)
|
|
assert tenant.notify_emails == ["a@b.ba", "c@d.ba"]
|
|
assert tenant.sms_request_received is True
|
|
assert "confirmation" in tenant.sms_templates
|
|
|
|
|
|
# -- super-admin ------------------------------------------------------------------
|
|
|
|
|
|
async def test_admin_create_tenant_and_owner(gogo_client, session, admin_user):
|
|
await login_admin(gogo_client)
|
|
resp = await gogo_client.post(
|
|
"/admin/tenants/new",
|
|
data={
|
|
"name": "Salon Aida", "slug": "salon-aida", "city": "Sarajevo",
|
|
"status": "active", "plan": "gogo_start", "included_minutes": "300",
|
|
"paid_until": "2026-12-31", "proposal_ttl_hours": "24",
|
|
"scheduling_provider": "partner_api",
|
|
"partner_base_url": "http://partner.test",
|
|
"partner_api_key": "test-partner-key",
|
|
"partner_webhook_secret": "test-webhook-secret",
|
|
"owner_email": "aida@salon.ba", "owner_password": "lozinka1",
|
|
"phone_number_id": "", "llm_model": "", "prompt_override": "",
|
|
},
|
|
)
|
|
assert resp.status_code == 303, resp.text
|
|
t = (
|
|
await session.execute(select(Tenant).where(Tenant.slug == "salon-aida"))
|
|
).scalar_one()
|
|
assert t.scheduling_provider == "partner_api"
|
|
assert t.paid_until is not None
|
|
user = (
|
|
await session.execute(select(User).where(User.tenant_id == t.id))
|
|
).scalar_one()
|
|
assert user.email == "aida@salon.ba"
|
|
|
|
from gogo.models import ProviderConfig
|
|
|
|
cfg = (
|
|
await session.execute(select(ProviderConfig).where(ProviderConfig.tenant_id == t.id))
|
|
).scalar_one()
|
|
assert cfg.config["base_url"] == "http://partner.test"
|
|
assert cfg.config["api_key_encrypted"] # stored encrypted
|
|
|
|
|
|
async def test_admin_connectivity_test_against_fake_partner(
|
|
gogo_client, session, admin_user, partner_tenant, wire_fake_partner
|
|
):
|
|
wire_fake_partner.slots["SRV-12"] = [
|
|
{"start": "2026-07-15T14:30:00+02:00", "end": "2026-07-15T15:15:00+02:00"}
|
|
]
|
|
await login_admin(gogo_client)
|
|
resp = await gogo_client.post(f"/admin/tenants/{partner_tenant.id}/test-connection")
|
|
assert resp.status_code == 200
|
|
assert "GET /availability" in resp.text
|
|
assert "dry_run" in resp.text or "dry-run" in resp.text
|
|
assert "ok=True" in resp.text
|
|
# dry-run must not create anything in the partner system
|
|
assert wire_fake_partner.requests == {}
|
|
|
|
|
|
async def test_admin_prompt_template_versioning(gogo_client, session, admin_user, tenant):
|
|
await login_admin(gogo_client)
|
|
resp = await gogo_client.post(
|
|
"/admin/templates",
|
|
data={"body": "V1 {salon_profile} {greeting}", "notes": "prva", "publish": "on"},
|
|
)
|
|
assert resp.status_code == 303
|
|
resp = await gogo_client.post(
|
|
"/admin/templates",
|
|
data={"body": "V2 {salon_profile} {greeting}", "notes": "druga", "publish": "on"},
|
|
)
|
|
assert resp.status_code == 303
|
|
|
|
versions = (await session.execute(select(PromptTemplate))).scalars().all()
|
|
published = [v for v in versions if v.published]
|
|
assert len(versions) == 2
|
|
assert len(published) == 1 and published[0].version == 2
|
|
|
|
# composed prompt uses the published template
|
|
from gogo.agent.prompt import compose_system_prompt
|
|
|
|
prompt = await compose_system_prompt(session, tenant, "voice")
|
|
assert prompt.startswith("V2")
|
|
|
|
# rollback to v1
|
|
resp = await gogo_client.post("/admin/templates/1/publish")
|
|
assert resp.status_code == 303
|
|
session.expire_all()
|
|
await session.refresh(tenant) # async-safe reload (expired attrs can't lazy-load)
|
|
prompt = await compose_system_prompt(session, tenant, "voice")
|
|
assert prompt.startswith("V1")
|
|
|
|
|
|
async def test_admin_impersonation(gogo_client, session, admin_user, tenant):
|
|
await login_admin(gogo_client)
|
|
resp = await gogo_client.post(f"/admin/impersonate/{tenant.id}")
|
|
assert resp.status_code == 303
|
|
resp = await gogo_client.get("/dash")
|
|
assert resp.status_code == 200
|
|
assert "admin pregled" in resp.text
|
|
assert tenant.name in resp.text
|
|
|
|
|
|
async def test_admin_requires_login(gogo_client, session):
|
|
resp = await gogo_client.get("/admin")
|
|
assert resp.status_code == 303
|
|
assert resp.headers["location"] == "/admin/login"
|
|
|
|
|
|
async def test_admin_health_page(gogo_client, session, admin_user):
|
|
await login_admin(gogo_client)
|
|
resp = await gogo_client.get("/admin/health")
|
|
assert resp.status_code == 200
|
|
assert "Baza podataka" in resp.text
|