104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""B2B profile endpoint integration tests (ticket 004).
|
|
|
|
GET/PUT /api/v1/b2b/profile with X-API-Key auth — same fields as the web
|
|
PATCH surface, but idempotent full-replace PUT (B2B convention).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from tests.integration.conftest import user_token
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _unique_tg() -> int:
|
|
return abs(int(uuid.uuid4().int % 900000000000)) + 100000000000
|
|
|
|
|
|
async def _setup_key(client: httpx.AsyncClient, infra: dict[str, str]) -> str:
|
|
tg = _unique_tg()
|
|
token = await user_token(client, infra, tg)
|
|
r = await client.post(
|
|
"/api/v1/b2b/keys",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"name": f"profile-test-{tg}"},
|
|
)
|
|
assert r.status_code == 201, r.text
|
|
return r.json()["api_key"]
|
|
|
|
|
|
class TestB2BProfile:
|
|
async def test_get_profile_defaults(
|
|
self, client: httpx.AsyncClient, infra: dict[str, str]
|
|
) -> None:
|
|
key = await _setup_key(client, infra)
|
|
r = await client.get("/api/v1/b2b/profile", headers={"X-API-Key": key})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["language"] is None
|
|
assert r.json()["notif_prefs"]["marketing"] is False
|
|
|
|
async def test_put_is_idempotent(
|
|
self, client: httpx.AsyncClient, infra: dict[str, str]
|
|
) -> None:
|
|
key = await _setup_key(client, infra)
|
|
body = {
|
|
"language": "en",
|
|
"timezone": "Europe/Berlin",
|
|
"notif_prefs": {"report_ready": False, "security": True, "marketing": True},
|
|
"dashboard_prefs": {"severity_filter": "high", "per_page": 25, "density": "compact"},
|
|
}
|
|
r1 = await client.put("/api/v1/b2b/profile", json=body, headers={"X-API-Key": key})
|
|
assert r1.status_code == 200, r1.text
|
|
r2 = await client.put("/api/v1/b2b/profile", json=body, headers={"X-API-Key": key})
|
|
assert r2.status_code == 200, r2.text
|
|
assert r1.json() == r2.json()
|
|
|
|
r3 = await client.get("/api/v1/b2b/profile", headers={"X-API-Key": key})
|
|
assert r3.json() == body
|
|
|
|
async def test_put_invalid_timezone_422(
|
|
self, client: httpx.AsyncClient, infra: dict[str, str]
|
|
) -> None:
|
|
key = await _setup_key(client, infra)
|
|
r = await client.put(
|
|
"/api/v1/b2b/profile",
|
|
json={"timezone": "not-a-zone"},
|
|
headers={"X-API-Key": key},
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
async def test_invalid_key_401(self, client: httpx.AsyncClient) -> None:
|
|
r = await client.get("/api/v1/b2b/profile", headers={"X-API-Key": "nope"})
|
|
assert r.status_code == 401
|
|
r = await client.put("/api/v1/b2b/profile", json={}, headers={"X-API-Key": "nope"})
|
|
assert r.status_code == 401
|
|
|
|
async def test_missing_key_401(self, client: httpx.AsyncClient) -> None:
|
|
r = await client.get("/api/v1/b2b/profile")
|
|
assert r.status_code == 401
|
|
|
|
async def test_b2b_profile_matches_web_profile_row(
|
|
self, client: httpx.AsyncClient, infra: dict[str, str]
|
|
) -> None:
|
|
"""The b2b profile is the same row the web API serves (key owner)."""
|
|
tg = _unique_tg()
|
|
token = await user_token(client, infra, tg)
|
|
r = await client.post(
|
|
"/api/v1/b2b/keys",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"name": f"profile-shared-{tg}"},
|
|
)
|
|
key = r.json()["api_key"]
|
|
|
|
await client.patch(
|
|
"/api/v1/me/profile",
|
|
json={"language": "ru"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
r = await client.get("/api/v1/b2b/profile", headers={"X-API-Key": key})
|
|
assert r.json()["language"] == "ru"
|