"""Backend integration tests for Srikandi Diamond Motors Pekanbaru dashboard.""" import os import io import pytest import requests BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://frosty-nightingale-6.preview.emergentagent.com").rstrip("/") ADMIN = {"email": "admin@srikandi.id", "password": "admin123"} SA = {"email": "sa@srikandi.id", "password": "sa12345"} MGR = {"email": "manager@srikandi.id", "password": "manager123"} @pytest.fixture(scope="session") def session(): s = requests.Session() s.headers.update({"Content-Type": "application/json"}) return s def _login(session, creds): r = session.post(f"{BASE_URL}/api/auth/login", json=creds, timeout=30) assert r.status_code == 200, f"Login failed: {r.status_code} {r.text}" return r.json() @pytest.fixture(scope="session") def admin_token(session): return _login(session, ADMIN)["access_token"] @pytest.fixture(scope="session") def sa_token(session): return _login(session, SA)["access_token"] @pytest.fixture(scope="session") def manager_token(session): return _login(session, MGR)["access_token"] def _hdr(token): return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} # -------- AUTH TESTS -------- class TestAuth: def test_login_admin(self, session): data = _login(session, ADMIN) assert "access_token" in data and len(data["access_token"]) > 10 assert data["user"]["role"] == "admin" assert data["user"]["email"] == ADMIN["email"] def test_login_sa(self, session): data = _login(session, SA) assert data["user"]["role"] == "service_advisor" def test_login_manager(self, session): data = _login(session, MGR) assert data["user"]["role"] == "manager" def test_login_invalid(self, session): r = session.post(f"{BASE_URL}/api/auth/login", json={"email": "wrong@x.id", "password": "bad"}) assert r.status_code == 401 assert "detail" in r.json() def test_me_with_token(self, session, admin_token): r = requests.get(f"{BASE_URL}/api/auth/me", headers=_hdr(admin_token)) assert r.status_code == 200 assert r.json()["role"] == "admin" def test_me_no_token(self, session): r = requests.get(f"{BASE_URL}/api/auth/me") assert r.status_code == 401 # -------- DASHBOARD ENDPOINTS -------- class TestDashboards: def test_home(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/home", headers=_hdr(admin_token)) assert r.status_code == 200 d = r.json() assert "today" in d for k in ["wo", "unit_entry", "revenue_total", "revenue_jasa", "revenue_sparepart", "service_order"]: assert k in d["today"], f"missing {k} in today" assert isinstance(d["trend_7days"], list) assert isinstance(d["mtd_chart"], list) assert len(d["top_service_advisors"]) <= 5 assert len(d["top_parts"]) <= 10 assert len(d["top_models"]) <= 5 def test_daily(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/daily", headers=_hdr(admin_token)) assert r.status_code == 200 assert "chart" in r.json() def test_service_advisor(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/service-advisor", headers=_hdr(admin_token)) assert r.status_code == 200 assert "ranking" in r.json() def test_sparepart(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/sparepart", headers=_hdr(admin_token)) assert r.status_code == 200 d = r.json() assert isinstance(d["fast_moving"], list) assert isinstance(d["slow_moving"], list) assert isinstance(d["total_revenue_sparepart"], (int, float)) assert isinstance(d["total_qty_sold"], (int, float)) def test_vehicle_population(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/vehicle-population", headers=_hdr(admin_token)) assert r.status_code == 200 d = r.json() assert len(d["passenger_car"]) == 5 assert len(d["light_commercial"]) == 3 for p in d["passenger_car"] + d["light_commercial"]: assert p["count"] > 0 def test_target(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/target", headers=_hdr(admin_token)) assert r.status_code == 200 d = r.json() assert len(d["rows"]) == 9 for row in d["rows"]: assert {"target", "actual", "achievement"}.issubset(row.keys()) assert isinstance(d["elapsed_percent"], (int, float)) def test_morning(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/morning-meeting", headers=_hdr(admin_token)) assert r.status_code == 200 def test_revenue(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/revenue", headers=_hdr(admin_token)) assert r.status_code == 200 d = r.json() assert "total_revenue" in d and "profit_jasa" in d def test_unit_entry(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/unit-entry", headers=_hdr(admin_token)) assert r.status_code == 200 def test_customer_retention(self, admin_token): r = requests.get(f"{BASE_URL}/api/dashboard/customer-retention", headers=_hdr(admin_token)) assert r.status_code == 200 assert "overall_retention_rate" in r.json() def test_dashboards_with_sa_role(self, sa_token): # SA should be able to read all dashboards r = requests.get(f"{BASE_URL}/api/dashboard/home", headers=_hdr(sa_token)) assert r.status_code == 200 # -------- TARGETS -------- class TestTargets: def test_get_targets_any_role(self, sa_token, manager_token): for t in [sa_token, manager_token]: r = requests.get(f"{BASE_URL}/api/targets", headers=_hdr(t)) assert r.status_code == 200 assert isinstance(r.json(), list) r = requests.get(f"{BASE_URL}/api/targets/current", headers=_hdr(sa_token)) assert r.status_code == 200 def test_upsert_target_admin(self, admin_token): payload = { "month": "2099-12", "wo": 999, "unit_entry": 999, "revenue_jasa": 1, "revenue_sparepart": 2, "total_revenue": 3, "service_order": 999, "profit_jasa": 1, "profit_sparepart": 2, "customer_retention": 50.0, } r = requests.post(f"{BASE_URL}/api/targets", json=payload, headers=_hdr(admin_token)) assert r.status_code == 200 # verify persistence lst = requests.get(f"{BASE_URL}/api/targets", headers=_hdr(admin_token)).json() assert any(t["month"] == "2099-12" and t["wo"] == 999 for t in lst) def test_upsert_target_sa_forbidden(self, sa_token): r = requests.post(f"{BASE_URL}/api/targets", json={"month": "2099-11", "wo": 1}, headers=_hdr(sa_token)) assert r.status_code == 403 def test_upsert_target_manager_forbidden(self, manager_token): r = requests.post(f"{BASE_URL}/api/targets", json={"month": "2099-11", "wo": 1}, headers=_hdr(manager_token)) assert r.status_code == 403 # -------- UPLOAD -------- class TestUpload: def test_upload_csv_admin(self, admin_token): csv_data = ( "date,wo,unit_entry,service_order,revenue_jasa,revenue_sparepart,profit_jasa,profit_sparepart\n" "2025-02-01,30,35,38,20000000,30000000,10000000,7000000\n" "2025-02-02,32,36,40,21000000,31000000,10500000,7200000\n" ) files = {"file": ("test.csv", io.BytesIO(csv_data.encode()), "text/csv")} r = requests.post(f"{BASE_URL}/api/upload/daily-report", files=files, headers={"Authorization": f"Bearer {admin_token}"}) assert r.status_code == 200, r.text d = r.json() assert d["total_rows"] == 2 assert (d["inserted"] + d["updated"]) == 2 assert isinstance(d["errors"], list) def test_upload_missing_columns(self, admin_token): bad = "date,wo\n2025-02-01,30\n" files = {"file": ("bad.csv", io.BytesIO(bad.encode()), "text/csv")} r = requests.post(f"{BASE_URL}/api/upload/daily-report", files=files, headers={"Authorization": f"Bearer {admin_token}"}) assert r.status_code == 400 def test_upload_sa_forbidden(self, sa_token): csv_data = "date,wo,unit_entry,service_order,revenue_jasa,revenue_sparepart,profit_jasa,profit_sparepart\n2025-02-01,30,35,38,20000000,30000000,10000000,7000000\n" files = {"file": ("t.csv", io.BytesIO(csv_data.encode()), "text/csv")} r = requests.post(f"{BASE_URL}/api/upload/daily-report", files=files, headers={"Authorization": f"Bearer {sa_token}"}) assert r.status_code == 403 # -------- ADMIN RESET/SEED -------- class TestAdminOps: def test_reseed_admin(self, admin_token): # ensure seeded r = requests.post(f"{BASE_URL}/api/admin/seed-dummy", headers=_hdr(admin_token)) assert r.status_code == 200 # verify data populated h = requests.get(f"{BASE_URL}/api/dashboard/home", headers=_hdr(admin_token)).json() assert len(h["trend_7days"]) > 0 def test_reset_then_reseed(self, admin_token): r = requests.post(f"{BASE_URL}/api/admin/reset-data", headers=_hdr(admin_token)) assert r.status_code == 200 # dashboards must not crash h = requests.get(f"{BASE_URL}/api/dashboard/home", headers=_hdr(admin_token)) assert h.status_code == 200 # reseed for follow-up tests / demo rr = requests.post(f"{BASE_URL}/api/admin/seed-dummy", headers=_hdr(admin_token)) assert rr.status_code == 200 def test_reset_sa_forbidden(self, sa_token): r = requests.post(f"{BASE_URL}/api/admin/reset-data", headers=_hdr(sa_token)) assert r.status_code == 403 def test_seed_manager_forbidden(self, manager_token): r = requests.post(f"{BASE_URL}/api/admin/seed-dummy", headers=_hdr(manager_token)) assert r.status_code == 403 # -------- REGISTER -------- class TestRegister: def test_register_admin_only(self, admin_token, sa_token): # SA cannot register r = requests.post(f"{BASE_URL}/api/auth/register", json={"email": "test_blocked@srikandi.id", "password": "x", "name": "X", "role": "service_advisor"}, headers=_hdr(sa_token)) assert r.status_code == 403 # Admin can register import uuid as _u email = f"test_{_u.uuid4().hex[:6]}@srikandi.id" r2 = requests.post(f"{BASE_URL}/api/auth/register", json={"email": email, "password": "pass1234", "name": "Test User", "role": "service_advisor"}, headers=_hdr(admin_token)) assert r2.status_code == 200 assert r2.json()["role"] == "service_advisor"