From 0d9fbc76681f39c3a9a8a4fec5f52c636fe2936b Mon Sep 17 00:00:00 2001 From: Leopoldadmin Date: Wed, 3 Jun 2026 23:52:05 +0000 Subject: [PATCH] Upload tests/test_users_me.py --- tests/test_users_me.py | 195 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 tests/test_users_me.py diff --git a/tests/test_users_me.py b/tests/test_users_me.py new file mode 100644 index 0000000..313a974 --- /dev/null +++ b/tests/test_users_me.py @@ -0,0 +1,195 @@ +"""User /me endpoint tests + RBAC verification.""" + +from __future__ import annotations + +from httpx import AsyncClient + + +# === /users/me === + + +async def test_get_me_returns_user_data( + client: AsyncClient, auth_headers: dict[str, str], registered_user: dict +) -> None: + """GET /api/v1/users/me with a valid JWT returns 200 + email, name, role, org_id.""" + resp = await client.get("/api/v1/users/me", headers=auth_headers) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["email"] == registered_user["email"] + assert data["name"] == registered_user["name"] + assert "role" in data + assert "org_id" in data + assert "id" in data + assert "password_hash" not in data + assert "avatar_url" in data + assert "email_notifications" in data + + +async def test_get_me_invalid_token_format(client: AsyncClient) -> None: + """GET /api/v1/users/me with a malformed token returns 401.""" + resp = await client.get( + "/api/v1/users/me", + headers={"Authorization": "Bearer this-is-not-a-jwt"}, + ) + assert resp.status_code == 401, resp.text + + +async def test_get_me_bogus_token(client: AsyncClient) -> None: + """GET /api/v1/users/me with a token signed with the wrong key returns 401.""" + from jose import jwt + import time + + bogus = jwt.encode( + { + "sub": "1", + "org_id": 1, + "role": "admin", + "exp": int(time.time()) + 3600, + "iat": int(time.time()), + }, + "completely-different-secret-key-32chars-abc", + algorithm="HS256", + ) + resp = await client.get( + "/api/v1/users/me", + headers={"Authorization": f"Bearer {bogus}"}, + ) + assert resp.status_code == 401, resp.text + + +# === PATCH /users/{id} === + + +async def test_update_profile_self( + client: AsyncClient, auth_headers: dict[str, str], registered_user: dict +) -> None: + """PATCH /api/v1/users/{self_id} updates the current user's name.""" + user_id = registered_user["user"]["id"] + resp = await client.patch( + f"/api/v1/users/{user_id}", + json={"name": "Updated Name", "email_notifications": False}, + headers=auth_headers, + ) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["name"] == "Updated Name" + assert data["email_notifications"] is False + # email and id should be unchanged + assert data["email"] == registered_user["email"] + assert data["id"] == user_id + + +async def test_update_profile_cannot_change_other( + client: AsyncClient, auth_headers: dict[str, str], registered_user: dict +) -> None: + """Non-admin cannot update another user's profile (403).""" + # Bootstrap user IS admin (first registered user). To test the 403 case, + # create a sales_rep and try to update someone else. + # First, create a 2nd user (admin can do this) + create_resp = await client.post( + "/api/v1/users/", + json={ + "email": "rep@test.com", + "password": "RepPass123!", + "name": "Test Rep", + "role": "sales_rep", + }, + headers=auth_headers, + ) + assert create_resp.status_code == 201, create_resp.text + other_id = create_resp.json()["id"] + + # Now log in as the rep and try to update the admin + login_resp = await client.post( + "/api/v1/auth/login", + data={"username": "rep@test.com", "password": "RepPass123!"}, + ) + assert login_resp.status_code == 200, login_resp.text + rep_token = login_resp.json()["access_token"] + rep_headers = {"Authorization": f"Bearer {rep_token}"} + + # Rep tries to update the admin → 403 + admin_id = registered_user["user"]["id"] + resp = await client.patch( + f"/api/v1/users/{admin_id}", + json={"name": "Hacked"}, + headers=rep_headers, + ) + assert resp.status_code == 403, resp.text + + +# === GET /users/ (admin list) === + + +async def test_list_users_as_admin( + client: AsyncClient, auth_headers: dict[str, str], registered_user: dict +) -> None: + """GET /api/v1/users/ as admin returns 200 + paginated list.""" + resp = await client.get("/api/v1/users/", headers=auth_headers) + assert resp.status_code == 200, resp.text + data = resp.json() + assert "items" in data + assert "total" in data + assert "page" in data + assert "page_size" in data + assert data["total"] >= 1 + assert len(data["items"]) >= 1 + # The bootstrap user is in the list + emails = [u["email"] for u in data["items"]] + assert registered_user["email"] in emails + + +async def test_list_users_as_sales_rep_forbidden( + client: AsyncClient, auth_headers: dict[str, str] +) -> None: + """GET /api/v1/users/ as sales_rep → 403.""" + # Create a sales_rep via the admin + create_resp = await client.post( + "/api/v1/users/", + json={ + "email": "rep@test.com", + "password": "RepPass123!", + "name": "Test Rep", + "role": "sales_rep", + }, + headers=auth_headers, + ) + assert create_resp.status_code == 201, create_resp.text + + # Log in as the rep + login_resp = await client.post( + "/api/v1/auth/login", + data={"username": "rep@test.com", "password": "RepPass123!"}, + ) + assert login_resp.status_code == 200, login_resp.text + rep_token = login_resp.json()["access_token"] + rep_headers = {"Authorization": f"Bearer {rep_token}"} + + # Rep tries to list all users → 403 + resp = await client.get("/api/v1/users/", headers=rep_headers) + assert resp.status_code == 403, resp.text + + +# === Refresh & Logout === + + +async def test_refresh_token( + client: AsyncClient, auth_headers: dict[str, str] +) -> None: + """POST /api/v1/auth/refresh issues a new token for the current user.""" + resp = await client.post("/api/v1/auth/refresh", headers=auth_headers) + assert resp.status_code == 200, resp.text + data = resp.json() + assert "access_token" in data + assert data["token_type"] == "bearer" + assert data["expires_in"] > 0 + + +async def test_logout( + client: AsyncClient, auth_headers: dict[str, str] +) -> None: + """POST /api/v1/auth/logout returns 200 with confirmation message.""" + resp = await client.post("/api/v1/auth/logout", headers=auth_headers) + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["message"] == "logged out"