0c2d4ba685
Introduces the TargetList model (team-scoped JSONB target arrays), Pydantic schemas, and full CRUD REST API at /target-lists/. Includes Alembic migration and 5 integration tests (TDD). Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
"""Tests for target lists CRUD."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.team import Team
|
|
from app.models.user import User
|
|
from sqlalchemy import select
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_headers(client: AsyncClient, test_db: AsyncSession, test_user: dict):
|
|
"""Override auth_headers to ensure the test user has a team_id assigned."""
|
|
# Fetch the user from DB and assign a team
|
|
result = await test_db.execute(select(User).where(User.email == test_user["email"]))
|
|
user = result.scalar_one()
|
|
|
|
# Create a team and assign the user to it
|
|
team = Team(name="Test Team")
|
|
test_db.add(team)
|
|
await test_db.flush()
|
|
|
|
user.team_id = team.id
|
|
await test_db.commit()
|
|
|
|
# Re-login to get a fresh token
|
|
login_data = {
|
|
"email": test_user["email"],
|
|
"password": test_user["password"],
|
|
}
|
|
resp = await client.post("/api/v1/auth/login/json", json=login_data)
|
|
assert resp.status_code == 200
|
|
token_data = resp.json()
|
|
return {"Authorization": f"Bearer {token_data['access_token']}"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_target_list(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.post(
|
|
"/api/v1/target-lists/",
|
|
json={
|
|
"name": "RDS Farm A",
|
|
"description": "Production RDS servers",
|
|
"targets": [
|
|
{"label": "RDS-01", "notes": "192.168.1.10"},
|
|
{"label": "RDS-02", "notes": "192.168.1.11"},
|
|
],
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
data = resp.json()
|
|
assert data["name"] == "RDS Farm A"
|
|
assert len(data["targets"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_target_lists(client: AsyncClient, auth_headers: dict):
|
|
resp = await client.get("/api/v1/target-lists/", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert isinstance(resp.json(), list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_target_list(client: AsyncClient, auth_headers: dict):
|
|
create = await client.post(
|
|
"/api/v1/target-lists/",
|
|
json={"name": "Get Test", "targets": [{"label": "SRV-01"}]},
|
|
headers=auth_headers,
|
|
)
|
|
list_id = create.json()["id"]
|
|
resp = await client.get(f"/api/v1/target-lists/{list_id}", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Get Test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_target_list(client: AsyncClient, auth_headers: dict):
|
|
create = await client.post(
|
|
"/api/v1/target-lists/",
|
|
json={"name": "Old Name", "targets": [{"label": "SRV-01"}]},
|
|
headers=auth_headers,
|
|
)
|
|
list_id = create.json()["id"]
|
|
resp = await client.put(
|
|
f"/api/v1/target-lists/{list_id}",
|
|
json={"name": "New Name", "targets": [{"label": "SRV-01"}, {"label": "SRV-02"}]},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "New Name"
|
|
assert len(resp.json()["targets"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_target_list(client: AsyncClient, auth_headers: dict):
|
|
create = await client.post(
|
|
"/api/v1/target-lists/",
|
|
json={"name": "To Delete", "targets": [{"label": "X"}]},
|
|
headers=auth_headers,
|
|
)
|
|
list_id = create.json()["id"]
|
|
resp = await client.delete(f"/api/v1/target-lists/{list_id}", headers=auth_headers)
|
|
assert resp.status_code == 204
|
|
|
|
get = await client.get(f"/api/v1/target-lists/{list_id}", headers=auth_headers)
|
|
assert get.status_code == 404
|