e05472615b
P3-A: Add account_id to audit_logs model + migration (backfill via user_id → users.account_id). log_audit() gains optional account_id param with fallback SELECT to avoid churn across 40 call sites. P3-B: Add account_id to tree_shares model + migration (backfill via created_by → users.account_id). TreeShare constructor updated in trees.py. P3-C: Enable RLS on 6 remaining tables: step_ratings, step_usage_log, target_lists, session_shares, audit_logs, tree_shares. P3-D: Drop team_id from target_lists — endpoint, schema, and model now use account_id as the sole isolation key. P3-E: Append Phase 3 RLS isolation tests for all 6 tables. test_target_lists.py: fix cross-account test to use Account model (not Team) and set account_id on new User. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""Tests for target lists CRUD."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.user import User
|
|
from sqlalchemy import select
|
|
|
|
|
|
@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
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cannot_access_other_accounts_list(client: AsyncClient, auth_headers: dict, test_db):
|
|
"""User from account B cannot access account A's target list."""
|
|
import uuid
|
|
from app.models.account import Account
|
|
from app.models.user import User
|
|
from app.core.security import get_password_hash
|
|
|
|
# Create account A list using existing auth_headers
|
|
create = await client.post(
|
|
"/api/v1/target-lists/",
|
|
json={"name": "Account A List", "targets": [{"label": "SRV-A"}]},
|
|
headers=auth_headers,
|
|
)
|
|
assert create.status_code == 201
|
|
list_id = create.json()["id"]
|
|
|
|
# Create a separate account B with its own user
|
|
account_b = Account(
|
|
name=f"Account B {uuid.uuid4()}",
|
|
display_code=f"AB{str(uuid.uuid4())[:6].upper()}",
|
|
)
|
|
test_db.add(account_b)
|
|
await test_db.flush()
|
|
|
|
user_b = User(
|
|
email=f"userb_{uuid.uuid4()}@test.com",
|
|
password_hash=get_password_hash("password123"),
|
|
name="User B",
|
|
is_active=True,
|
|
account_id=account_b.id,
|
|
account_role="engineer",
|
|
role="engineer",
|
|
)
|
|
test_db.add(user_b)
|
|
await test_db.flush()
|
|
await test_db.commit()
|
|
|
|
# Get auth token for user B
|
|
login = await client.post(
|
|
"/api/v1/auth/login/json",
|
|
json={"email": user_b.email, "password": "password123"},
|
|
)
|
|
assert login.status_code == 200
|
|
token_b = login.json()["access_token"]
|
|
headers_b = {"Authorization": f"Bearer {token_b}"}
|
|
|
|
# Account B cannot access Account A's list
|
|
resp = await client.get(f"/api/v1/target-lists/{list_id}", headers=headers_b)
|
|
assert resp.status_code == 404
|