Upload tests/test_users_me.py

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"