genichin
3509606c7e
- Add V1 API routes for assets, ledger, and user info with complete CRUD operations - Implement authentication scopes and request validation via auth_routes.py - Extend schemas with V1-specific models for accounts, holdings, and positions - Add database queries and aggregations for P&L, performance, and account data - Include comprehensive test coverage for all V1 endpoints and integration flows - Update batch and release documentation with implementation and verification evidence - Sync project structure and runtime flows per V1 API contract requirements Closes implementation phase of BAT-001 (TrEx Basic V1 API Contract)
131 lines
3.7 KiB
Python
131 lines
3.7 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
|
|
|
|
|
|
_TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
|
|
|
|
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",
|
|
credential_encryption_key=_TEST_ENC_KEY,
|
|
)
|
|
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",
|
|
credential_encryption_key=_TEST_ENC_KEY,
|
|
)
|
|
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 == "****"
|