genichin
ce63c75c9c
- Add V1 API endpoints: assets, ledger, user-info - Implement Fernet-based credential payload encryption (req-010) - Extend health response schema with runtime info (req-004) - Add service token runtime ID issuance (req-005) - Implement database schema migration for V1 API (req-009) - Add comprehensive test coverage for V1 endpoints - Update SRS with 10 new requirements (REQ-004 through REQ-011) - Create batch planning, design, and verification for BAT-001 - Integrate review-discovery StagePilot skill - Update documentation indices and baseline specs
126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import Settings
|
|
from app.core.config import get_settings
|
|
from app.core.security import mask_secret
|
|
from app.domain.models import User
|
|
from app.main import app, create_app
|
|
|
|
|
|
def test_healthcheck() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "ok"
|
|
assert body["service"] == "portfolio-hub"
|
|
assert "version" in body
|
|
assert "checked_at" in body
|
|
|
|
|
|
def test_docs_are_exposed() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/docs")
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_settings_default_origin() -> None:
|
|
settings = get_settings()
|
|
|
|
assert settings.allowed_origins == ["http://localhost:3000"]
|
|
|
|
|
|
class _BootstrapRepo:
|
|
def __init__(self, users: list[User] | None = None) -> None:
|
|
self.users = list(users or [])
|
|
self.created: list[tuple[str, str | None, str | None, str]] = []
|
|
self.closed = False
|
|
|
|
def list_users(self) -> list[User]:
|
|
return list(self.users)
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
*,
|
|
email: str | None = None,
|
|
password_hash: str | None = None,
|
|
permission: str = "read",
|
|
is_active: bool = True,
|
|
) -> User:
|
|
self.created.append((username, email, password_hash, permission))
|
|
user = User(
|
|
id=f"user-{len(self.users) + 1}",
|
|
username=username,
|
|
email=email,
|
|
permission=permission,
|
|
is_active=is_active,
|
|
password_hash=password_hash,
|
|
)
|
|
self.users.append(user)
|
|
return user
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def test_startup_bootstraps_initial_user_when_users_table_empty() -> None:
|
|
repo = _BootstrapRepo()
|
|
settings = Settings(
|
|
db_enabled=True,
|
|
bootstrap_username="admin",
|
|
bootstrap_email="admin@example.com",
|
|
bootstrap_password="bootstrap123",
|
|
allowed_origins_raw="http://localhost:3000",
|
|
)
|
|
bootstrap_app = create_app(settings_factory=lambda: settings, repository_factory=lambda current: repo)
|
|
|
|
with TestClient(bootstrap_app) as client:
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert repo.created[0][0] == "admin"
|
|
assert repo.created[0][1] == "admin@example.com"
|
|
assert repo.created[0][2] is not None
|
|
assert repo.created[0][3] == "admin"
|
|
assert repo.closed is True
|
|
|
|
|
|
def test_startup_does_not_bootstrap_when_users_exist() -> None:
|
|
repo = _BootstrapRepo(users=[User(id="user-1", username="existing")])
|
|
settings = Settings(
|
|
db_enabled=True,
|
|
bootstrap_username="admin",
|
|
bootstrap_email="admin@example.com",
|
|
bootstrap_password="bootstrap123",
|
|
allowed_origins_raw="http://localhost:3000",
|
|
)
|
|
bootstrap_app = create_app(settings_factory=lambda: settings, repository_factory=lambda current: repo)
|
|
|
|
with TestClient(bootstrap_app) as client:
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert repo.created == []
|
|
assert repo.closed is True
|
|
|
|
|
|
def test_mask_secret_keeps_prefix_and_suffix() -> None:
|
|
masked = mask_secret("abcdefghijklmnopqrstuvwxyz", prefix=6, suffix=4)
|
|
|
|
assert masked.startswith("abcdef")
|
|
assert masked.endswith("wxyz")
|
|
assert "ghijklmnopqrstuv" not in masked
|
|
|
|
|
|
def test_mask_secret_short_value_is_fully_masked() -> None:
|
|
masked = mask_secret("abcd", prefix=3, suffix=2)
|
|
|
|
assert masked == "****"
|