genichinandClaude Sonnet 4.6
6add0ade29
- confirm-req-implemented: REQ-001~018 (17건, REQ-013 제외) Approved → Implemented - BAT-001(REQ-009/011/014), BAT-002(REQ-010/012/015), BAT-003(REQ-001/002/003/006) - BAT-004(REQ-004/005), BAT-005(REQ-007/008), BAT-006(REQ-016/017/018) - CHG-20260521-01 entry 각 REQ에 추가; SRS index Recent Change Log 갱신 - REQ-010 CHG-20260521-02: GET /v1/assets admin catalog 전용 재정의 - 기존 token_type 분기 portfolio 조회 코드 제거, admin permission + list_assets() 호출로 대체 - AssetCatalogListResponse 스키마 추가 (items/total) - REQ-010 Status: Implemented → Approved (재검증 필요) - 테스트 343건 전체 통과: service/user token 403, admin token 200 + catalog schema 검증 - migration 보강: portfolios_asset_migration market 컬럼 부재 fallback, portfolio_tags 빈 테이블 early-return, service_token index fix autocommit_block 정리 - docs/releases/rel-001: REL-001 draft 추가 (BAT-001~011 포함) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
959 lines
36 KiB
Python
959 lines
36 KiB
Python
"""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, LedgerTrade
|
|
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/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:
|
|
settings = _make_app_db(monkeypatch)
|
|
|
|
portfolio = _make_portfolio()
|
|
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
|
|
mock_repo.list_portfolios_with_strategies.return_value = [portfolio]
|
|
mock_repo.get_strategy_set.return_value = None
|
|
|
|
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["assets"][0]
|
|
assert item["asset_id"] == "p1"
|
|
assert "market" not in item
|
|
assert item["broker_type"] == "kis"
|
|
assert item["cash"] == 1000.0
|
|
assert item["quantity"] == 10.0
|
|
assert item["strategy_params"] == {}
|
|
assert item["strategy_type"] is None
|
|
mock_repo.list_portfolios_with_strategies.assert_called_once_with(
|
|
owner_user_id="user-1",
|
|
include_disabled=False,
|
|
managed_by_daemon_id="d-1",
|
|
market=None,
|
|
)
|
|
|
|
|
|
def test_v1_assets_status_filter(monkeypatch) -> None:
|
|
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?status=active", headers=_service_token_header())
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_v1_assets_strategy_params(monkeypatch) -> None:
|
|
"""GET /v1/assets is admin catalog only — service token is rejected (strategy data is on GET /v1/portfolios)."""
|
|
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
|
|
|
|
|
|
def test_v1_assets_strategy_params_fallback_on_error(monkeypatch) -> None:
|
|
"""GET /v1/assets is admin catalog only — service token is rejected."""
|
|
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:
|
|
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) -> LedgerTrade:
|
|
defaults = dict(
|
|
ledger_entry_id=1,
|
|
asset_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 LedgerTrade(**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: LedgerTrade, 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["asset_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_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"
|