"""Tests for daemon assignment authorization on v1 endpoints (REQ-006, REQ-007, REQ-008).""" from __future__ import annotations from datetime import datetime, timezone 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 Portfolio, TokenPrincipal, User from app.main import create_app def _make_settings(**kwargs) -> Settings: return Settings(allowed_origins_raw="http://localhost:3000", **kwargs) def _svc_header(token: str = "svc-token") -> dict[str, str]: return {"Authorization": f"Bearer {token}"} def _app_with_db(monkeypatch) -> TestClient: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") get_settings.cache_clear() s = _make_settings() app = create_app(settings_factory=lambda: s) app.dependency_overrides[get_settings] = lambda: s return TestClient(app) def _principal(daemon_id: str | None = "d-1") -> TokenPrincipal: return TokenPrincipal( token_id="t1", user_id="user-1", permission="trade-write", token_type="service", runtime_id="rt-1", daemon_id=daemon_id, ) def _portfolio_with_daemon(daemon_id: str | None = "d-1") -> MagicMock: p = MagicMock(spec=Portfolio) p.managed_by_daemon_id = daemon_id p.portfolio_id = "asset-1" p.status = "enabled" return p def _ledger_trade_payload() -> dict: return { "order_id": "ord-1", "execution_id": "exec-1", "side": "buy", "quantity": "10", "price": "50000", "fee_amount": "0", "tax_amount": "0", "executed_at": "2026-05-13T10:00:00+09:00", } # --------------------------------------------------------------------------- # REQ-010 CHG-20260521-02: GET /v1/assets admin catalog — service token rejected # (daemon-scoped portfolio list moved to GET /v1/portfolios) # --------------------------------------------------------------------------- def test_assets_200_when_token_has_no_daemon_id(monkeypatch) -> None: """REQ-044: GET /v1/assets now requires only read — trade-write service token (no daemon_id) gets 200.""" client = _app_with_db(monkeypatch) principal = _principal(daemon_id=None) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.list_assets.return_value = [] mock_repo.get_tags_by_ids.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): resp = client.get("/v1/assets", headers=_svc_header()) assert resp.status_code == 200 assert resp.json()["total"] == 0 def test_assets_200_with_daemon_scoped_filter(monkeypatch) -> None: """REQ-044: GET /v1/assets now requires only read — trade-write service token with daemon_id also gets 200.""" client = _app_with_db(monkeypatch) principal = _principal(daemon_id="d-1") mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.list_assets.return_value = [] mock_repo.get_tags_by_ids.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): resp = client.get("/v1/assets", headers=_svc_header()) assert resp.status_code == 200 assert resp.json()["total"] == 0 # --------------------------------------------------------------------------- # REQ-007: GET /v1/user_info daemon_id field + NULL diagnostic # --------------------------------------------------------------------------- def test_user_info_returns_daemon_id_in_response(monkeypatch) -> None: client = _app_with_db(monkeypatch) principal = _principal(daemon_id="d-1") 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): resp = client.get("/v1/user_info", headers=_svc_header()) assert resp.status_code == 200 assert resp.json()["daemon_id"] == "d-1" def test_user_info_null_daemon_id_returns_403(monkeypatch) -> None: """REQ-049 AC-3: daemon_id=None service token → 403 Forbidden (BAT-020 정책 변경).""" client = _app_with_db(monkeypatch) principal = _principal(daemon_id=None) 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): resp = client.get("/v1/user_info", headers=_svc_header()) # REQ-049: daemon_id 없는 service token은 user_info 403 (daemon-scoped 인가) assert resp.status_code == 403 # --------------------------------------------------------------------------- # REQ-008/029: POST /v1/portfolios/{portfolio_id}/ledger/trades daemon assignment authorization # --------------------------------------------------------------------------- def _setup_ledger_mocks(monkeypatch, principal: TokenPrincipal, portfolio: MagicMock, trade: MagicMock) -> tuple[MagicMock, TestClient]: monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true") monkeypatch.setenv("CREDENTIAL_ENCRYPTION_KEY", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") get_settings.cache_clear() s = _make_settings(credential_encryption_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=") app = create_app(settings_factory=lambda: s) app.dependency_overrides[get_settings] = lambda: s client = TestClient(app) mock_repo = MagicMock() mock_repo.get_token_principal.return_value = principal mock_repo.get.return_value = portfolio mock_repo.insert_ledger_trade.return_value = (trade, True) return mock_repo, client def test_ledger_trade_403_when_token_has_no_daemon_id(monkeypatch) -> None: from app.domain.models import LedgerEntry principal = _principal(daemon_id=None) portfolio = _portfolio_with_daemon(daemon_id="d-1") trade = MagicMock(spec=LedgerEntry) mock_repo, client = _setup_ledger_mocks(monkeypatch, principal, portfolio, trade) 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): resp = client.post( "/v1/portfolios/asset-1/ledger/trades", json=_ledger_trade_payload(), headers=_svc_header(), ) assert resp.status_code == 403 assert "daemon_id" in resp.json()["detail"] def test_ledger_trade_403_when_asset_has_no_daemon_assignment(monkeypatch) -> None: from app.domain.models import LedgerEntry principal = _principal(daemon_id="d-1") portfolio = _portfolio_with_daemon(daemon_id=None) trade = MagicMock(spec=LedgerEntry) mock_repo, client = _setup_ledger_mocks(monkeypatch, principal, portfolio, trade) 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): resp = client.post( "/v1/portfolios/asset-1/ledger/trades", json=_ledger_trade_payload(), headers=_svc_header(), ) assert resp.status_code == 403 assert "not assigned" in resp.json()["detail"] def test_ledger_trade_403_when_daemon_mismatch(monkeypatch) -> None: from app.domain.models import LedgerEntry principal = _principal(daemon_id="d-1") portfolio = _portfolio_with_daemon(daemon_id="d-2") trade = MagicMock(spec=LedgerEntry) mock_repo, client = _setup_ledger_mocks(monkeypatch, principal, portfolio, trade) 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): resp = client.post( "/v1/portfolios/asset-1/ledger/trades", json=_ledger_trade_payload(), headers=_svc_header(), ) assert resp.status_code == 403 assert "mismatch" in resp.json()["detail"].lower() or "daemon" in resp.json()["detail"].lower() def test_ledger_trade_201_when_daemon_matches(monkeypatch) -> None: from app.domain.models import LedgerEntry principal = _principal(daemon_id="d-1") portfolio = _portfolio_with_daemon(daemon_id="d-1") trade = MagicMock() trade.ledger_entry_id = 1 trade.portfolio_id = "asset-1" trade.order_id = "ord-1" trade.execution_id = "exec-1" trade.side = "buy" trade.quantity = 10.0 trade.price = 50000.0 trade.fee_amount = 0.0 trade.tax_amount = 0.0 trade.executed_at = datetime(2026, 5, 13, 1, 0, 0, tzinfo=timezone.utc) trade.recorded_at = datetime(2026, 5, 13, 1, 0, 0, tzinfo=timezone.utc) mock_repo, client = _setup_ledger_mocks(monkeypatch, principal, portfolio, trade) 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): resp = client.post( "/v1/portfolios/asset-1/ledger/trades", json=_ledger_trade_payload(), headers=_svc_header(), ) assert resp.status_code == 201 assert resp.json()["status"] == "new"