feat(bat-019): REQ-044 GET /v1/assets require_read + REQ-045 GET /v1/daemons caller-scope

REQ-044: list_catalog_assets 의존성을 require_admin → require_read로 완화.
asset 쓰기(POST/PATCH/DELETE)는 require_admin 유지.

REQ-045: GET /v1/daemons 신규 엔드포인트 — caller 소유 데몬만 반환(require_read).
user_id=None(env 토큰)은 빈 목록 200. DaemonResponse/DaemonListResponse 스키마 추가.

기존 테스트(test_v1.py, test_asset_catalog.py, test_daemon_assignment.py, test_tags_master.py)
REQ-044 권한 완화에 맞게 업데이트. test_v1_daemons.py 신규 추가(13 케이스).

pytest 472 passed, 0 failed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
This commit is contained in:
2026-06-27 10:36:04 +09:00
co-authored by Claude Sonnet 4.6
parent d43ff430c2
commit c5288a8c42
13 changed files with 769 additions and 39 deletions
+20 -5
View File
@@ -1023,25 +1023,40 @@ def test_get_v1_assets_market_filter() -> None:
assert items[0]["symbol"] == "005930"
# REQ-010 AC-12: non-admin → HTTP 403
def test_get_v1_assets_user_token_returns_403() -> None:
# REQ-044: GET /v1/assets is now require_read — user/service tokens return 200
def test_get_v1_assets_user_token_returns_200() -> None:
"""REQ-044: read user token accepted by GET /v1/assets."""
client, _, patches = _make_catalog_get_client(_USER_KW, [], {})
try:
resp = client.get("/v1/assets", headers={"Authorization": "Bearer user"})
finally:
for p in patches:
p.stop()
assert resp.status_code == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_get_v1_assets_service_token_returns_403() -> None:
def test_get_v1_assets_service_token_returns_200() -> None:
"""REQ-044: trade-write service token accepted by GET /v1/assets (trade-write >= read)."""
client, _, patches = _make_catalog_get_client(_SERVICE_KW, [], {})
try:
resp = client.get("/v1/assets", headers={"Authorization": "Bearer svc"})
finally:
for p in patches:
p.stop()
assert resp.status_code == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_get_v1_assets_no_token_returns_401() -> None:
"""GET /v1/assets without token returns 401 (require_read still gates auth)."""
client, _, patches = _make_catalog_get_client(_ADMIN_KW, [], {})
try:
resp = client.get("/v1/assets") # no Authorization header
finally:
for p in patches:
p.stop()
assert resp.status_code == 401
# ---------------------------------------------------------------------------
+15 -7
View File
@@ -64,30 +64,38 @@ def _ledger_trade_payload() -> dict:
# (daemon-scoped portfolio list moved to GET /v1/portfolios)
# ---------------------------------------------------------------------------
def test_assets_403_when_token_has_no_daemon_id(monkeypatch) -> None:
"""GET /v1/assets requires admin — service token with no daemon_id gets 403 (Insufficient permission)."""
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):
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 == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_assets_200_with_daemon_scoped_filter(monkeypatch) -> None:
"""GET /v1/assets requires admin — service token with daemon_id is also rejected."""
"""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):
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 == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
# ---------------------------------------------------------------------------
+53 -9
View File
@@ -320,54 +320,98 @@ def test_asset_item_empty_tag_ids_returns_none() -> None:
def test_v1_assets_active_tag_not_archived(monkeypatch) -> None:
"""GET /v1/assets는 admin 전용 — user token은 403 반환 (archived 표시는 GET /v1/portfolios로 이관)."""
"""REQ-044: GET /v1/assets now accepts read token — active tag returns archived=False."""
from app.domain.models import Asset, Tag
import datetime as _dt
settings = _make_v1_settings(monkeypatch)
principal = TokenPrincipal(
token_id="t1", user_id=_USER_ID, permission="read",
token_type="user", runtime_id="rt-1",
)
asset = Asset(
asset_id="a-1", symbol="005930", markets=["KRX"],
status="active", tag_ids=[_TEST_TAG_ID_1],
created_at=_dt.datetime.now(_dt.timezone.utc),
updated_at=_dt.datetime.now(_dt.timezone.utc),
)
active_tag = Tag(tag_id=_TEST_TAG_ID_1, name="kospi200", created_by_user_id="admin", is_active=True)
mock_repo = MagicMock()
mock_repo.get_token_principal.return_value = principal
mock_repo.list_assets.return_value = [asset]
mock_repo.get_tags_by_ids.return_value = {_TEST_TAG_ID_1: active_tag}
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo):
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_v1_client(settings)
resp = client.get("/v1/assets", headers={"Authorization": "Bearer svc-token"})
assert resp.status_code == 403
assert resp.status_code == 200
item = resp.json()["items"][0]
assert item["tag_ids"] == [{"tag_id": _TEST_TAG_ID_1, "archived": False}]
def test_v1_assets_inactive_tag_archived_true(monkeypatch) -> None:
"""GET /v1/assets는 admin 전용 — user token은 403 반환."""
"""REQ-044: GET /v1/assets now accepts read token — inactive tag returns archived=True."""
from app.domain.models import Asset, Tag
import datetime as _dt
settings = _make_v1_settings(monkeypatch)
principal = TokenPrincipal(
token_id="t1", user_id=_USER_ID, permission="read",
token_type="user", runtime_id="rt-1",
)
asset = Asset(
asset_id="a-1", symbol="005930", markets=["KRX"],
status="active", tag_ids=[_TEST_TAG_ID_1, _TEST_TAG_ID_2],
created_at=_dt.datetime.now(_dt.timezone.utc),
updated_at=_dt.datetime.now(_dt.timezone.utc),
)
active_tag = Tag(tag_id=_TEST_TAG_ID_1, name="kospi200", created_by_user_id="admin", is_active=True)
inactive_tag = Tag(tag_id=_TEST_TAG_ID_2, name="legacy", created_by_user_id="admin", is_active=False)
mock_repo = MagicMock()
mock_repo.get_token_principal.return_value = principal
mock_repo.list_assets.return_value = [asset]
mock_repo.get_tags_by_ids.return_value = {_TEST_TAG_ID_1: active_tag, _TEST_TAG_ID_2: inactive_tag}
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo):
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_v1_client(settings)
resp = client.get("/v1/assets", headers={"Authorization": "Bearer svc-token"})
assert resp.status_code == 403
assert resp.status_code == 200
item = resp.json()["items"][0]
by_id = {t["tag_id"]: t for t in item["tag_ids"]}
assert by_id[_TEST_TAG_ID_1]["archived"] is False
assert by_id[_TEST_TAG_ID_2]["archived"] is True
def test_v1_assets_empty_tag_ids_returns_none(monkeypatch) -> None:
"""GET /v1/assets는 admin 전용 — user token은 403 반환."""
"""REQ-044: GET /v1/assets now accepts read token — asset with no tags returns empty tag_ids."""
from app.domain.models import Asset
import datetime as _dt
settings = _make_v1_settings(monkeypatch)
principal = TokenPrincipal(
token_id="t1", user_id=_USER_ID, permission="read",
token_type="user", runtime_id="rt-1",
)
asset = Asset(
asset_id="a-1", symbol="005930", markets=["KRX"],
status="active", tag_ids=[],
created_at=_dt.datetime.now(_dt.timezone.utc),
updated_at=_dt.datetime.now(_dt.timezone.utc),
)
mock_repo = MagicMock()
mock_repo.get_token_principal.return_value = principal
mock_repo.list_assets.return_value = [asset]
mock_repo.get_tags_by_ids.return_value = {}
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo):
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_v1_client(settings)
resp = client.get("/v1/assets", headers={"Authorization": "Bearer svc-token"})
assert resp.status_code == 403
assert resp.status_code == 200
item = resp.json()["items"][0]
assert item["tag_ids"] == []
# ---------------------------------------------------------------------------
+56 -16
View File
@@ -610,18 +610,26 @@ def test_v1_assets_requires_service_token(monkeypatch) -> None:
assert resp.status_code == 401
def test_v1_assets_rejects_env_token() -> None:
def test_v1_assets_env_read_token_returns_200(monkeypatch) -> None:
"""REQ-044: env read token is now accepted (require_read) — list_catalog_assets."""
from app.core.config import get_settings
from app.domain.models import Asset
import datetime as _dt
settings = _make_settings(db_enabled=False, read_tokens_raw="env-token")
mock_repo = MagicMock()
mock_repo.list_assets.return_value = []
mock_repo.get_tags_by_ids.return_value = {}
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
with patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo):
resp = client.get("/v1/assets", headers=_service_token_header("env-token"))
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_v1_assets_returns_asset_list(monkeypatch) -> None:
"""GET /v1/assets requires admin — service token is rejected."""
"""REQ-044: GET /v1/assets now accepts read service token — returns 200 with catalog list."""
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
@@ -629,12 +637,16 @@ def test_v1_assets_returns_asset_list(monkeypatch) -> 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):
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()
resp = client.get("/v1/assets", headers=_service_token_header())
assert resp.status_code == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
# ---------------------------------------------------------------------------
@@ -642,7 +654,7 @@ def test_v1_assets_returns_asset_list(monkeypatch) -> None:
# ---------------------------------------------------------------------------
def test_v1_assets_user_token(monkeypatch) -> None:
"""User token is rejected — admin catalog endpoint requires admin permission."""
"""REQ-044: user read token is now accepted (require_read) — returns 200."""
settings = _make_app_db(monkeypatch)
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
@@ -650,12 +662,16 @@ def test_v1_assets_user_token(monkeypatch) -> 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):
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 == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_v1_assets_admin_token_default(monkeypatch) -> None:
@@ -716,24 +732,44 @@ def test_v1_assets_admin_token_user_id_param(monkeypatch) -> None:
def test_v1_assets_market_filter(monkeypatch) -> None:
"""Service token is rejected by admin catalog endpoint — REQ-010 AC-12."""
"""REQ-044: read service token with market filter now returns 200 (require_read)."""
from app.domain.models import Asset
import datetime as _dt
settings = _make_app_db(monkeypatch)
krx_asset = Asset(
asset_id="a-krx", 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),
)
nasdaq_asset = Asset(
asset_id="a-nasdaq", symbol="AAPL", markets=["NASDAQ"],
display_name="Apple", 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="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_assets.return_value = [krx_asset, nasdaq_asset]
mock_repo.get_tags_by_ids.return_value = {}
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo):
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?market=KRX", headers=_service_token_header())
assert resp.status_code == 403
assert resp.status_code == 200
items = resp.json()["items"]
assert len(items) == 1
assert items[0]["symbol"] == "005930"
def test_v1_assets_asset_id_is_uuid(monkeypatch) -> None:
"""Service token is rejected by admin catalog endpoint."""
def test_v1_assets_read_token_200(monkeypatch) -> None:
"""REQ-044: read service token accepted by GET /v1/assets (require_read)."""
settings = _make_app_db(monkeypatch)
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
@@ -741,12 +777,16 @@ def test_v1_assets_asset_id_is_uuid(monkeypatch) -> 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):
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 == 403
assert resp.status_code == 200
assert resp.json()["total"] == 0
# ---------------------------------------------------------------------------
+349
View File
@@ -0,0 +1,349 @@
"""Tests for BAT-019: GET /v1/daemons — caller-scoped daemon list (REQ-045)."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from app.core.config import Settings, get_settings
from app.domain.models import Daemon, TokenPrincipal
from app.main import create_app
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
def _make_settings(**kwargs) -> Settings:
return Settings(
allowed_origins_raw="http://localhost:3000",
**kwargs,
)
def _make_app(settings: Settings) -> TestClient:
get_settings.cache_clear()
app = create_app(settings_factory=lambda: settings)
app.dependency_overrides[get_settings] = lambda: settings
return TestClient(app)
def _auth_header(token: str = "test-token") -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def _make_daemon(
daemon_id: str = "d-1",
owner_user_id: str = "user-1",
daemon_slug: str = "trex-orb-v1",
display_name: str | None = "ORB v1",
is_active: bool = True,
) -> Daemon:
return Daemon(
id=daemon_id,
owner_user_id=owner_user_id,
daemon_slug=daemon_slug,
display_name=display_name,
is_active=is_active,
)
def _make_mock_repo_with_principal(principal: TokenPrincipal, daemons: list[Daemon]) -> MagicMock:
mock_repo = MagicMock()
mock_repo.get_token_principal.return_value = principal
mock_repo.list_daemons_for_user.return_value = daemons
return mock_repo
_SETTINGS_DB = _make_settings(db_enabled=True, credential_encryption_key=_TEST_ENC_KEY)
# ---------------------------------------------------------------------------
# REQ-045 AC: 사용자A(데몬2) → 자기 2개만
# ---------------------------------------------------------------------------
def test_daemons_returns_caller_owned_daemons() -> None:
"""REQ-045 AC: 사용자A 소유 2개 데몬 → 2개 반환."""
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
daemons = [
_make_daemon("d-1", "user-1", "trex-orb-v1", "ORB v1", is_active=True),
_make_daemon("d-2", "user-1", "trex-tsl-v1", "TSL v1", is_active=True),
]
mock_repo = _make_mock_repo_with_principal(principal, daemons)
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 2
assert len(body["daemons"]) == 2
slugs = {d["daemon_slug"] for d in body["daemons"]}
assert slugs == {"trex-orb-v1", "trex-tsl-v1"}
# ---------------------------------------------------------------------------
# REQ-045 AC: 필드 4개 포함
# ---------------------------------------------------------------------------
def test_daemons_response_has_required_fields() -> None:
"""REQ-045 AC: id, daemon_slug, display_name, is_active 필드 포함."""
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
daemon = _make_daemon("d-1", "user-1", "trex-orb-v1", "ORB v1", is_active=True)
mock_repo = _make_mock_repo_with_principal(principal, [daemon])
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
item = resp.json()["daemons"][0]
assert item["id"] == "d-1"
assert item["daemon_slug"] == "trex-orb-v1"
assert item["display_name"] == "ORB v1"
assert item["is_active"] is True
# ---------------------------------------------------------------------------
# REQ-045 AC: display_name None 허용
# ---------------------------------------------------------------------------
def test_daemons_display_name_can_be_none() -> None:
"""REQ-045: display_name=None인 데몬도 정상 반환."""
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
daemon = _make_daemon("d-3", "user-1", "trex-custom", None, is_active=True)
mock_repo = _make_mock_repo_with_principal(principal, [daemon])
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
item = resp.json()["daemons"][0]
assert item["display_name"] is None
# ---------------------------------------------------------------------------
# REQ-045 AC: 타사용자 데몬 비노출 (사용자B 데몬은 user-2 소유)
# ---------------------------------------------------------------------------
def test_daemons_does_not_expose_other_user_daemons() -> None:
"""REQ-045 AC: 사용자B 데몬은 사용자A 조회 시 반환되지 않는다."""
principal_a = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
# list_daemons_for_user("user-1")은 user-1 소유만 반환 (repository 레벨 격리)
daemons_for_user_a = [
_make_daemon("d-1", "user-1", "trex-orb-v1", "ORB v1", is_active=True),
]
mock_repo = _make_mock_repo_with_principal(principal_a, daemons_for_user_a)
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 1
# user-2 소유 데몬은 없음
owner_ids = {d.get("owner_user_id") for d in body["daemons"]}
# DaemonResponse는 owner_user_id 미포함이므로, repository 호출 인수 검증
mock_repo.list_daemons_for_user.assert_called_once_with("user-1")
# ---------------------------------------------------------------------------
# REQ-045 AC: 데몬 없으면 빈 목록 200
# ---------------------------------------------------------------------------
def test_daemons_empty_list_returns_200() -> None:
"""REQ-045 AC: 데몬 없는 사용자 → 빈 목록 200."""
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
mock_repo = _make_mock_repo_with_principal(principal, [])
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
body = resp.json()
assert body["daemons"] == []
assert body["total"] == 0
# ---------------------------------------------------------------------------
# REQ-045 AC: 무토큰 → 401
# ---------------------------------------------------------------------------
def test_daemons_no_token_returns_401() -> None:
"""REQ-045 AC: 무토큰 → 401."""
settings = _make_settings(db_enabled=False)
client = _make_app(settings)
resp = client.get("/v1/daemons")
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# REQ-045 AC: admin도 caller 스코프 (admin user_id=None → 빈 목록)
# ---------------------------------------------------------------------------
def test_daemons_admin_token_with_user_id_none_returns_empty() -> None:
"""REQ-045: env admin 토큰(user_id=None)은 빈 목록 200 반환."""
settings = _make_settings(db_enabled=False, admin_tokens_raw="admin-env-token")
mock_repo = MagicMock()
mock_repo.list_daemons_for_user.return_value = []
with patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(settings)
resp = client.get("/v1/daemons", headers=_auth_header("admin-env-token"))
assert resp.status_code == 200
body = resp.json()
assert body["daemons"] == []
assert body["total"] == 0
# user_id=None이므로 list_daemons_for_user는 호출되지 않아야 함
mock_repo.list_daemons_for_user.assert_not_called()
def test_daemons_admin_token_with_user_id_returns_caller_daemons() -> None:
"""REQ-045: admin DB 토큰(user_id='admin-1')은 admin-1 소유 데몬만 반환."""
principal = TokenPrincipal(
token_id="t-admin", user_id="admin-1", permission="admin",
token_type="admin", runtime_id="rt-admin",
)
daemons = [
_make_daemon("d-admin-1", "admin-1", "admin-bot", "Admin Bot", is_active=True),
]
mock_repo = _make_mock_repo_with_principal(principal, daemons)
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 1
assert body["daemons"][0]["id"] == "d-admin-1"
mock_repo.list_daemons_for_user.assert_called_once_with("admin-1")
# ---------------------------------------------------------------------------
# REQ-045: is_active=False인 데몬도 반환 (소유 데몬 전체)
# ---------------------------------------------------------------------------
def test_daemons_includes_inactive_daemons() -> None:
"""REQ-045: inactive 데몬도 is_active=False로 반환됨."""
principal = TokenPrincipal(
token_id="t1", user_id="user-1", permission="read",
token_type="service", runtime_id="rt-1",
)
daemons = [
_make_daemon("d-1", "user-1", "trex-orb-v1", "ORB v1", is_active=True),
_make_daemon("d-2", "user-1", "old-bot", "Old Bot", is_active=False),
]
mock_repo = _make_mock_repo_with_principal(principal, daemons)
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
patch("app.api.v1_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/daemons", headers=_auth_header())
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 2
active_flags = {d["daemon_slug"]: d["is_active"] for d in body["daemons"]}
assert active_flags["trex-orb-v1"] is True
assert active_flags["old-bot"] is False
# ---------------------------------------------------------------------------
# REQ-044 회귀: GET /v1/assets read 토큰 → 200, 쓰기는 admin 필요 → 403
# ---------------------------------------------------------------------------
def test_assets_read_token_200_regression() -> None:
"""REQ-044 회귀: read 토큰으로 GET /v1/assets → 200."""
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
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):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/assets", headers=_auth_header())
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_assets_write_requires_admin_regression() -> None:
"""REQ-044 회귀: read 토큰으로 POST /v1/assets → 403 (쓰기는 require_admin 유지)."""
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(_SETTINGS_DB)
resp = client.post(
"/v1/assets",
json={"symbol": "005930", "markets": ["KRX"]},
headers=_auth_header(),
)
assert resp.status_code == 403
def test_assets_no_token_returns_401_regression() -> None:
"""REQ-044 회귀: 무토큰 GET /v1/assets → 401."""
settings = _make_settings(db_enabled=False)
client = _make_app(settings)
resp = client.get("/v1/assets")
assert resp.status_code == 401
def test_assets_admin_get_200_regression() -> None:
"""REQ-044 회귀: admin 토큰 GET /v1/assets → 200 (기존 동작 유지)."""
principal = TokenPrincipal(
token_id="t-admin", 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 = []
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):
client = _make_app(_SETTINGS_DB)
resp = client.get("/v1/assets", headers=_auth_header())
assert resp.status_code == 200