97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.device_type import DeviceType
|
|
from app.models.user import User
|
|
from app.core.service_account import PLATFORM_ACCOUNT_ID
|
|
|
|
|
|
async def _login_headers(client, email: str, password: str) -> dict[str, str]:
|
|
response = await client.post(
|
|
"/api/v1/auth/login/json",
|
|
json={"email": email, "password": password},
|
|
)
|
|
assert response.status_code == 200
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_device_types_include_platform_and_account_custom(client, test_db, auth_headers, test_user):
|
|
result = await test_db.execute(select(User).where(User.email == test_user["email"]))
|
|
user = result.scalar_one()
|
|
|
|
test_db.add(
|
|
DeviceType(
|
|
id=uuid.uuid4(),
|
|
slug="platform-router",
|
|
label="Platform Router",
|
|
category="network",
|
|
is_system=True,
|
|
account_id=PLATFORM_ACCOUNT_ID,
|
|
sort_order=0,
|
|
)
|
|
)
|
|
await test_db.commit()
|
|
|
|
create_response = await client.post(
|
|
"/api/v1/device-types/",
|
|
json={
|
|
"slug": "tenant-appliance",
|
|
"label": "Tenant Appliance",
|
|
"category": "network",
|
|
"sort_order": 3,
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert create_response.status_code == 201
|
|
assert create_response.json()["account_id"] == str(user.account_id)
|
|
|
|
list_response = await client.get("/api/v1/device-types/", headers=auth_headers)
|
|
assert list_response.status_code == 200
|
|
payload = list_response.json()
|
|
slugs = {item["slug"] for item in payload}
|
|
|
|
assert "platform-router" in slugs
|
|
assert "tenant-appliance" in slugs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_diagrams_are_account_scoped(client, test_db, auth_headers, test_user):
|
|
other_user = {
|
|
"email": "other-network@example.com",
|
|
"password": "TestPassword123!",
|
|
"name": "Other Network User",
|
|
}
|
|
register_response = await client.post("/api/v1/auth/register", json=other_user)
|
|
assert register_response.status_code in (200, 201)
|
|
other_headers = await _login_headers(client, other_user["email"], other_user["password"])
|
|
|
|
owner_result = await test_db.execute(select(User).where(User.email == test_user["email"]))
|
|
owner = owner_result.scalar_one()
|
|
|
|
create_response = await client.post(
|
|
"/api/v1/network-diagrams/",
|
|
json={
|
|
"name": "HQ Core",
|
|
"client_name": "Acme",
|
|
"description": "Primary topology",
|
|
"nodes": [],
|
|
"edges": [],
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert create_response.status_code == 201
|
|
diagram = create_response.json()
|
|
assert diagram["account_id"] == str(owner.account_id)
|
|
|
|
own_get = await client.get(f"/api/v1/network-diagrams/{diagram['id']}", headers=auth_headers)
|
|
assert own_get.status_code == 200
|
|
|
|
other_get = await client.get(f"/api/v1/network-diagrams/{diagram['id']}", headers=other_headers)
|
|
assert other_get.status_code == 404
|