"""Tests for v1 API endpoints (health, assets, user_info, ledger/trades).""" from __future__ import annotations from datetime import datetime, timezone from decimal import Decimal from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from app.core.auth import AuthContext, Permission from app.core.config import Settings, get_settings from app.domain.models import AuthToken, BrokerCredential, Portfolio, TokenPrincipal, User, LedgerEntry from app.main import create_app # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- # Valid Fernet test key (32 null bytes base64-encoded) _TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" def _make_settings(**kwargs) -> Settings: return Settings( allowed_origins_raw="http://localhost:3000", **kwargs, ) def _make_app(settings: Settings | None = None) -> TestClient: get_settings.cache_clear() s = settings or _make_settings() app = create_app(settings_factory=lambda: s) app.dependency_overrides[get_settings] = lambda: s return TestClient(app) def _make_app_db(monkeypatch, **extra_settings) -> Settings: """Return settings with db_enabled=True and a valid encryption key for startup.""" monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") monkeypatch.setenv("CREDENTIAL_ENCRYPTION_KEY", _TEST_ENC_KEY) return _make_settings(**extra_settings) def _service_token_header(token: str = "svc-token") -> dict[str, str]: return {"Authorization": f"Bearer {token}"} class _RuntimeIdRoundTripRepo: def __init__(self) -> None: self.users = { "user-1": User( id="user-1", username="alice", email="alice@example.com", permission="read", is_active=True, ) } self.users_by_username = {"alice": self.users["user-1"]} self.tokens_by_hash: dict[str, AuthToken] = {} self._seq = 1 def close(self) -> None: return None def get_user_by_username(self, username: str) -> User | None: return self.users_by_username.get(username) def get_user(self, user_id: str) -> User: user = self.users.get(user_id) if user is None: raise ValueError("사용자를 찾을 수 없습니다") return user def create_auth_token( self, user_id: str, *, name: str, token_hash: str, token_prefix: str, permission: str, token_type: str, expires_at: datetime | None = None, runtime_id: str | None = None, daemon_id: str | None = None, ) -> AuthToken: token = AuthToken( id=f"token-{self._seq}", user_id=user_id, name=name, token_prefix=token_prefix, permission=permission, token_type=token_type, expires_at=expires_at, created_at=datetime.now(timezone.utc), runtime_id=runtime_id, daemon_id=daemon_id, ) self._seq += 1 self.tokens_by_hash[token_hash] = token return token def get_token_principal(self, token_hash: str) -> TokenPrincipal | None: token = self.tokens_by_hash.get(token_hash) if token is None: return None return TokenPrincipal( token_id=token.id, user_id=token.user_id, permission=token.permission, token_type=token.token_type, runtime_id=token.runtime_id, daemon_id=token.daemon_id, ) def list_broker_credentials(self, user_id: str) -> list[BrokerCredential]: return [] # --------------------------------------------------------------------------- # GET /health # --------------------------------------------------------------------------- def test_health_returns_full_response() -> None: client = _make_app() resp = client.get("/health") assert resp.status_code == 200 body = resp.json() assert body["status"] == "ok" assert body["service"] == "portfolio-hub" assert "version" in body assert "checked_at" in body # checked_at should contain KST offset (+09:00) assert "+09:00" in body["checked_at"] # --------------------------------------------------------------------------- # POST /auth/service-tokens (B-1: REQ-005 AC2) # --------------------------------------------------------------------------- def test_auth_issue_service_token_returns_runtime_id(monkeypatch) -> None: repo = _RuntimeIdRoundTripRepo() settings = _make_settings( db_enabled=True, admin_tokens_raw="admin-token", credential_encryption_key=_TEST_ENC_KEY, ) with patch("app.services.auth_service.get_portfolio_repository_from_settings", return_value=repo), \ patch("app.core.auth.get_portfolio_repository_from_settings", return_value=repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=repo): app = create_app(settings_factory=lambda: settings) app.dependency_overrides[get_settings] = lambda: settings client = TestClient(app) issue_resp = client.post( "/auth/service-tokens", headers={"Authorization": "Bearer admin-token"}, json={"name": "bot-1", "username": "alice"}, ) assert issue_resp.status_code == 201 issue_body = issue_resp.json() assert issue_body["runtime_id"] is not None assert "token" in issue_body user_info_resp = client.get( "/v1/user_info", headers=_service_token_header(issue_body["token"]), ) assert user_info_resp.status_code == 200 user_info_body = user_info_resp.json() assert user_info_body["runtime_id"] == issue_body["runtime_id"] assert user_info_body["username"] == "alice" def test_auth_issue_service_token_generates_distinct_runtime_ids_per_token(monkeypatch) -> None: repo = _RuntimeIdRoundTripRepo() settings = _make_settings( db_enabled=True, admin_tokens_raw="admin-token", credential_encryption_key=_TEST_ENC_KEY, ) with patch("app.services.auth_service.get_portfolio_repository_from_settings", return_value=repo): app = create_app(settings_factory=lambda: settings) app.dependency_overrides[get_settings] = lambda: settings client = TestClient(app) first = client.post( "/auth/service-tokens", headers={"Authorization": "Bearer admin-token"}, json={"name": "bot-1", "username": "alice"}, ) second = client.post( "/auth/service-tokens", headers={"Authorization": "Bearer admin-token"}, json={"name": "bot-2", "username": "alice"}, ) assert first.status_code == 201 assert second.status_code == 201 assert first.json()["runtime_id"] is not None assert second.json()["runtime_id"] is not None assert first.json()["runtime_id"] != second.json()["runtime_id"] # --------------------------------------------------------------------------- # GET /v1/user_info # --------------------------------------------------------------------------- def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None: settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id=None, # no runtime_id ) user = User(id="user-1", username="alice", permission="read", is_active=True) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.return_value = [] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 assert resp.json()["runtime_id"] is None def test_v1_user_info_returns_user(monkeypatch) -> None: settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-abc" ) user = User( id="user-1", username="alice", email="alice@example.com", permission="read", is_active=True, ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.return_value = [] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert body["runtime_id"] == "rt-abc" assert body["username"] == "alice" assert body["trading_enabled"] is True assert body["disabled_reason"] is None assert body["brokers"] == [] def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None: settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-2", permission="read", token_type="service", runtime_id="rt-xyz" ) user = User( id="user-2", username="bob", email=None, permission="read", is_active=False, ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.return_value = [] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert body["trading_enabled"] is False assert body["disabled_reason"] == "account_inactive" def test_v1_user_info_returns_decrypted_broker_credentials(monkeypatch) -> None: """REQ-007 AC5: brokers[]에 복호화된 credential_payload 반환.""" from app.core.encryption import encrypt_payload import json as _json settings = _make_app_db(monkeypatch) plain = _json.dumps({"app_key": "mykey", "app_secret": "mysecret"}) encrypted = encrypt_payload(_TEST_ENC_KEY, plain) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-abc" ) user = User(id="user-1", username="alice", permission="read", is_active=True) cred = BrokerCredential( user_id="user-1", broker="kis", account_ref="12345678", encrypted_payload=encrypted, credential_version=1, ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.return_value = [cred] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): settings = _make_settings(credential_encryption_key=_TEST_ENC_KEY) from app.core.config import get_settings as _gs app_instance = create_app(settings_factory=lambda: settings) app_instance.dependency_overrides[_gs] = lambda: settings client = TestClient(app_instance) resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert len(body["brokers"]) == 1 broker = body["brokers"][0] assert broker["broker"] == "kis" assert broker["credential_payload"] == {"app_key": "mykey", "app_secret": "mysecret"} # --------------------------------------------------------------------------- # POST /v1/assets/{asset_id}/ledger/trades # --------------------------------------------------------------------------- def _trade_payload(**overrides) -> dict: base = { "order_id": "ord-001", "execution_id": "exec-001", "side": "buy", "quantity": "10.5", "price": "50000", "fee_amount": "100", "tax_amount": "0", "executed_at": "2026-05-01T10:00:00+09:00", } base.update(overrides) return base def _make_ledger_trade(**kwargs) -> LedgerEntry: defaults = dict( ledger_entry_id=1, portfolio_id="005930", order_id="ord-001", execution_id="exec-001", side="buy", quantity=10.5, price=50000.0, fee_amount=100.0, tax_amount=0.0, executed_at=datetime(2026, 5, 1, 1, 0, 0, tzinfo=timezone.utc), recorded_at=datetime(2026, 5, 1, 1, 0, 0, tzinfo=timezone.utc), ) defaults.update(kwargs) return LedgerEntry(**defaults) def _make_ledger_principal() -> TokenPrincipal: return TokenPrincipal( token_id="t1", user_id="user-1", permission="trade-write", token_type="service", runtime_id="rt-1", daemon_id="d-1" ) def _make_mock_repo_for_ledger(monkeypatch, trade: LedgerEntry, created: bool, found: bool = True) -> MagicMock: settings = _make_app_db(monkeypatch) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = _make_ledger_principal() if found: mock_portfolio = MagicMock() mock_portfolio.managed_by_daemon_id = "d-1" mock_repo.get.return_value = mock_portfolio else: mock_repo.get.side_effect = Exception("not found") mock_repo.insert_ledger_trade.return_value = (trade, created) return mock_repo def test_v1_ledger_trade_created(monkeypatch) -> None: trade = _make_ledger_trade() mock_repo = _make_mock_repo_for_ledger(monkeypatch, trade, created=True) with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.post( "/v1/portfolios/005930/ledger/trades", json=_trade_payload(), headers=_service_token_header(), ) assert resp.status_code == 201 body = resp.json() assert body["status"] == "new" assert body["portfolio_id"] == "005930" def test_v1_ledger_trade_idempotent_same_payload(monkeypatch) -> None: """Duplicate with same payload → HTTP 200 + status='duplicate'.""" trade = _make_ledger_trade() mock_repo = _make_mock_repo_for_ledger(monkeypatch, trade, created=False) with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.post( "/v1/portfolios/005930/ledger/trades", json=_trade_payload(), # same payload headers=_service_token_header(), ) assert resp.status_code == 200 body = resp.json() assert body["status"] == "duplicate" def test_v1_ledger_trade_conflict_different_payload(monkeypatch) -> None: """Duplicate key with different payload → HTTP 409.""" trade = _make_ledger_trade() # side=buy, quantity=10.5 mock_repo = _make_mock_repo_for_ledger(monkeypatch, trade, created=False) with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.post( "/v1/portfolios/005930/ledger/trades", json=_trade_payload(quantity="99.0"), # different quantity → conflict headers=_service_token_header(), ) assert resp.status_code == 409 def test_v1_ledger_trade_asset_not_found(monkeypatch) -> None: """Non-existent portfolio_id → HTTP 404.""" trade = _make_ledger_trade() mock_repo = _make_mock_repo_for_ledger(monkeypatch, trade, created=True, found=False) with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.post( "/v1/portfolios/nonexistent/ledger/trades", json=_trade_payload(), headers=_service_token_header(), ) assert resp.status_code == 404 def test_v1_ledger_trade_requires_trade_write_permission(monkeypatch) -> None: """Read-only service token must be rejected before asset lookup.""" _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.post( "/v1/portfolios/005930/ledger/trades", json=_trade_payload(), headers=_service_token_header(), ) assert resp.status_code == 403 assert resp.json()["detail"] == "Insufficient permission" mock_repo.get.assert_not_called() mock_repo.insert_ledger_trade.assert_not_called() # --------------------------------------------------------------------------- # REQ-010: startup validation # --------------------------------------------------------------------------- def test_startup_fails_when_encryption_key_missing() -> None: """B-10: Settings.db_enabled=True + no key → lifespan raises RuntimeError.""" import pytest settings = _make_settings(db_enabled=True, credential_encryption_key="") with pytest.raises(RuntimeError, match="CREDENTIAL_ENCRYPTION_KEY"): with TestClient(create_app(settings_factory=lambda: settings)): pass # lifespan should raise before this def test_startup_ok_when_encryption_key_present() -> None: """B-10 AC1: valid key → app starts normally.""" settings = _make_settings(db_enabled=False, credential_encryption_key=_TEST_ENC_KEY) with TestClient(create_app(settings_factory=lambda: settings)) as client: resp = client.get("/health") assert resp.status_code == 200 # --------------------------------------------------------------------------- # REQ-010: encrypt/decrypt round-trip (B-11 evidence) # --------------------------------------------------------------------------- def test_encrypt_decrypt_roundtrip() -> None: """Fernet encrypt→upsert→decrypt round-trip works for credential payload.""" import json as _json from app.core.encryption import encrypt_payload, decrypt_payload plain = _json.dumps({"app_key": "mykey", "app_secret": "mysecret"}) ciphertext = encrypt_payload(_TEST_ENC_KEY, plain) # ciphertext is not plaintext assert ciphertext != plain assert "mykey" not in ciphertext # roundtrip recovered = decrypt_payload(_TEST_ENC_KEY, ciphertext) assert _json.loads(recovered) == {"app_key": "mykey", "app_secret": "mysecret"} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_settings(**kwargs) -> Settings: return Settings( allowed_origins_raw="http://localhost:3000", **kwargs, ) def _make_app(settings: Settings | None = None) -> TestClient: get_settings.cache_clear() s = settings or _make_settings() app = create_app(settings_factory=lambda: s) app.dependency_overrides[get_settings] = lambda: s return TestClient(app) def _service_token_header(token: str = "svc-token") -> dict[str, str]: return {"Authorization": f"Bearer {token}"} # --------------------------------------------------------------------------- # GET /health # --------------------------------------------------------------------------- def test_health_returns_full_response() -> None: client = _make_app() resp = client.get("/health") assert resp.status_code == 200 body = resp.json() assert body["status"] == "ok" assert body["service"] == "portfolio-hub" assert "version" in body assert "checked_at" in body # checked_at should contain KST offset (+09:00) assert "+09:00" in body["checked_at"] # --------------------------------------------------------------------------- # GET /v1/assets # --------------------------------------------------------------------------- def _make_portfolio(portfolio_id: str = "p1") -> Portfolio: return Portfolio( portfolio_id=portfolio_id, broker="kis", status="enabled", name="Test", symbol="005930", tags=["tag1"], cash_balance=1000.0, asset_quantity=10.0, price=50000.0, profit_buffer=0.0, active_strategy_id=None, avabile_strategies=None, owner_user_id="user-1", managed_by_daemon_id="d-1", ) def test_v1_assets_requires_service_token(monkeypatch) -> None: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false") settings = _make_settings(db_enabled=False) client = _make_app(settings) resp = client.get("/v1/assets") assert resp.status_code == 401 def test_v1_assets_rejects_env_token() -> None: from app.core.config import get_settings settings = _make_settings(db_enabled=False, read_tokens_raw="env-token") app = create_app(settings_factory=lambda: settings) app.dependency_overrides[get_settings] = lambda: settings client = TestClient(app) resp = client.get("/v1/assets", headers=_service_token_header("env-token")) assert resp.status_code == 403 def test_v1_assets_returns_asset_list(monkeypatch) -> None: """GET /v1/assets requires admin — service token is rejected.""" monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-1", daemon_id="d-1" ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app() resp = client.get("/v1/assets", headers=_service_token_header()) assert resp.status_code == 403 # --------------------------------------------------------------------------- # REQ-010 CHG-20260521-02: GET /v1/assets admin catalog (path 분리) # --------------------------------------------------------------------------- def test_v1_assets_user_token(monkeypatch) -> None: """User token is rejected — admin catalog endpoint requires admin permission.""" settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="user", runtime_id="rt-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/assets", headers=_service_token_header()) assert resp.status_code == 403 def test_v1_assets_admin_token_default(monkeypatch) -> None: """Admin token returns catalog list — REQ-010 AC-9.""" import datetime as _dt from app.domain.models import Asset settings = _make_app_db(monkeypatch) asset = Asset( asset_id="a1-uuid", symbol="005930", markets=["KRX"], display_name="Samsung", status="active", created_at=_dt.datetime.now(_dt.timezone.utc), updated_at=_dt.datetime.now(_dt.timezone.utc), ) principal = TokenPrincipal( token_id="t1", user_id="admin-1", permission="admin", token_type="admin", runtime_id="rt-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.list_assets.return_value = [asset] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/assets", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert body["total"] == 1 item = body["items"][0] assert item["symbol"] == "005930" assert item["status"] == "active" assert "portfolio_id" not in item mock_repo.list_assets.assert_called_once() def test_v1_assets_admin_token_user_id_param(monkeypatch) -> None: """user_id param is not supported on catalog endpoint — returns 200 with full catalog.""" import datetime as _dt from app.domain.models import Asset settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="admin-1", permission="admin", token_type="admin", runtime_id="rt-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.list_assets.return_value = [] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/assets?user_id=other-user", headers=_service_token_header()) assert resp.status_code == 200 assert resp.json()["total"] == 0 mock_repo.list_assets.assert_called_once() def test_v1_assets_market_filter(monkeypatch) -> None: """Service token is rejected by admin catalog endpoint — REQ-010 AC-12.""" settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-1", daemon_id="d-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/assets?market=KRX", headers=_service_token_header()) assert resp.status_code == 403 def test_v1_assets_asset_id_is_uuid(monkeypatch) -> None: """Service token is rejected by admin catalog endpoint.""" settings = _make_app_db(monkeypatch) principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-1", daemon_id="d-1", ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app(_make_settings(credential_encryption_key=_TEST_ENC_KEY)) resp = client.get("/v1/assets", headers=_service_token_header()) assert resp.status_code == 403 # --------------------------------------------------------------------------- # GET /v1/user_info # --------------------------------------------------------------------------- def test_v1_user_info_null_runtime_id_returns_200_with_null(monkeypatch) -> None: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id=None, # no runtime_id ) user = User(id="user-1", username="alice", permission="read", is_active=True) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.return_value = [] with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app() resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 assert resp.json()["runtime_id"] is None def test_v1_user_info_returns_user(monkeypatch) -> None: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-abc" ) user = User( id="user-1", username="alice", email="alice@example.com", permission="read", is_active=True, ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app() resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert body["runtime_id"] == "rt-abc" assert body["username"] == "alice" assert body["disabled_reason"] is None def test_v1_user_info_degrades_when_broker_credentials_lookup_fails(monkeypatch) -> None: """issue #9 regression: broker_credentials lookup failure (missing table / DB error) must NOT crash /v1/user_info with 500. daemon_id/user_id/permission do not depend on broker credentials — degrade gracefully with empty brokers.""" monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") principal = TokenPrincipal( token_id="t1", user_id="user-1", permission="read", token_type="service", runtime_id="rt-1", daemon_id="daemon-9", ) user = User(id="user-1", username="alice", permission="read", is_active=True) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user mock_repo.list_broker_credentials.side_effect = RuntimeError( 'relation "broker_credentials" does not exist' ) with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app() resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 body = resp.json() assert body["daemon_id"] == "daemon-9" assert body["user_id"] == "user-1" assert body["brokers"] == [] def test_v1_user_info_inactive_user_has_disabled_reason(monkeypatch) -> None: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") principal = TokenPrincipal( token_id="t1", user_id="user-2", permission="read", token_type="service", runtime_id="rt-xyz" ) user = User( id="user-2", username="bob", email=None, permission="read", is_active=False, ) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get_user.return_value = user with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \ patch("app.api.v1_user_info.get_portfolio_repository_from_settings", return_value=mock_repo): client = _make_app() resp = client.get("/v1/user_info", headers=_service_token_header()) assert resp.status_code == 200 assert resp.json()["disabled_reason"] == "account_inactive"