Fusion
73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
from typing import Any
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from app.models.secret import Secret
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_secret_encrypt_decrypt(
|
|
auth_client: AsyncClient, db_session: async_sessionmaker[Any]
|
|
) -> None:
|
|
resp = await auth_client.post(
|
|
"/api/v1/projects",
|
|
json={"name": "SecretTest", "slug": "secret-test"},
|
|
)
|
|
project_id = resp.json()["id"]
|
|
|
|
resp = await auth_client.post(
|
|
"/api/v1/secrets",
|
|
json={
|
|
"scope_type": "project",
|
|
"scope_id": str(project_id),
|
|
"key": "api_key",
|
|
"value": "super-secret",
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["value"] == "super-secret"
|
|
secret_id = data["id"]
|
|
|
|
resp = await auth_client.get(f"/api/v1/secrets/{secret_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["value"] == "super-secret"
|
|
|
|
async with db_session() as session:
|
|
result = await session.execute(select(Secret).where(Secret.id == secret_id))
|
|
secret = result.scalar_one()
|
|
assert secret.encrypted_value != "super-secret"
|
|
|
|
resp = await auth_client.put(
|
|
f"/api/v1/secrets/{secret_id}",
|
|
json={"value": "new-secret"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["value"] == "new-secret"
|
|
|
|
resp = await auth_client.delete(f"/api/v1/secrets/{secret_id}")
|
|
assert resp.status_code == 204
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_secret_ownership_enforced(auth_client: AsyncClient) -> None:
|
|
resp = await auth_client.post("/api/v1/projects", json={"name": "P1", "slug": "p1"})
|
|
p1 = resp.json()["id"]
|
|
|
|
resp = await auth_client.post("/api/v1/projects", json={"name": "P2", "slug": "p2"})
|
|
p2 = resp.json()["id"]
|
|
|
|
resp = await auth_client.post(
|
|
"/api/v1/secrets",
|
|
json={"scope_type": "project", "scope_id": str(p1), "key": "k1", "value": "v1"},
|
|
)
|
|
|
|
resp = await auth_client.get("/api/v1/secrets", params={"scope_id": str(p2)})
|
|
assert resp.status_code == 200
|
|
secrets = resp.json()
|
|
for s in secrets:
|
|
assert s["scope_id"] != str(p1) or s["scope_type"] != "project"
|