786 lines
31 KiB
Python
786 lines
31 KiB
Python
"""BAT-022: REQ-053 — POST/PATCH /v1/admin/daemons (admin 데몬 CRUD)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.errors import DaemonInUseError
|
|
from app.domain.models import BrokerCredential, Daemon, TokenPrincipal, User
|
|
from app.main import create_app
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
_TEST_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
|
|
ADMIN_USER_ID = "admin-user"
|
|
OWNER_A_ID = "user-a"
|
|
OWNER_B_ID = "user-b"
|
|
DAEMON_ID = "daemon-001"
|
|
|
|
|
|
def _make_settings(**kwargs) -> Settings:
|
|
return Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_TEST_ENC_KEY,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def _make_client(settings: Settings, mock_repo) -> 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 = "admin-token") -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _make_daemon(
|
|
daemon_id: str = DAEMON_ID,
|
|
owner_user_id: str = OWNER_A_ID,
|
|
daemon_slug: str = "trex-orb",
|
|
display_name: str = "ORB",
|
|
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,
|
|
created_at=datetime(2026, 6, 28, 10, 0, 0, tzinfo=timezone.utc),
|
|
updated_at=datetime(2026, 6, 28, 10, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
|
|
|
|
def _make_admin_principal(user_id: str = ADMIN_USER_ID) -> TokenPrincipal:
|
|
return TokenPrincipal(
|
|
token_id="adm-1",
|
|
user_id=user_id,
|
|
permission="admin",
|
|
token_type="admin",
|
|
)
|
|
|
|
|
|
def _make_user_principal(user_id: str = OWNER_A_ID) -> TokenPrincipal:
|
|
return TokenPrincipal(
|
|
token_id="usr-1",
|
|
user_id=user_id,
|
|
permission="read",
|
|
token_type="user",
|
|
)
|
|
|
|
|
|
def _make_service_principal(user_id: str = OWNER_A_ID) -> TokenPrincipal:
|
|
return TokenPrincipal(
|
|
token_id="svc-1",
|
|
user_id=user_id,
|
|
permission="read",
|
|
token_type="service",
|
|
daemon_id=DAEMON_ID,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-1: POST 생성 → 201 + DaemonAdminResponse
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_create_daemon_returns_201() -> None:
|
|
"""REQ-053 AC-1: admin POST → 201 + id/daemon_slug/display_name/owner_user_id/is_active."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon = _make_daemon()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.create_daemon.return_value = daemon
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
"/v1/admin/daemons",
|
|
json={"daemon_slug": "trex-orb", "display_name": "ORB", "owner_user_id": OWNER_A_ID},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["id"] == DAEMON_ID
|
|
assert body["daemon_slug"] == "trex-orb"
|
|
assert body["display_name"] == "ORB"
|
|
assert body["owner_user_id"] == OWNER_A_ID
|
|
assert body["is_active"] is True
|
|
# Secret fields must not appear (REQ-054 AC-3)
|
|
assert "app_key" not in body
|
|
assert "app_secret" not in body
|
|
assert "credential_payload" not in body
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-2: 중복 slug → 409
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_create_daemon_duplicate_slug_returns_409() -> None:
|
|
"""REQ-053 AC-2: 중복 daemon_slug → 409 Conflict."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.create_daemon.side_effect = ValueError("이미 존재하는 daemon slug입니다: trex-orb")
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
"/v1/admin/daemons",
|
|
json={"daemon_slug": "trex-orb", "owner_user_id": OWNER_A_ID},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 409
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-3: display_name 변경 — BrokerCredential 유지
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_patch_display_name_credential_preserved() -> None:
|
|
"""REQ-053 AC-3: display_name 변경 → 200, BrokerCredential 삭제 없음."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_before = _make_daemon(display_name="OLD NAME")
|
|
daemon_after = _make_daemon(display_name="NEW NAME")
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_before
|
|
mock_repo.update_daemon.return_value = daemon_after
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
f"/v1/admin/daemons/{DAEMON_ID}",
|
|
json={"display_name": "NEW NAME"},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["display_name"] == "NEW NAME"
|
|
# update_daemon_owner must NOT be called (no owner change)
|
|
mock_repo.update_daemon_owner.assert_not_called()
|
|
# delete_credentials_by_daemon_id must NOT be called
|
|
mock_repo.delete_credentials_by_daemon_id.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-4: owner 재할당 → BrokerCredential 자동 삭제
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_patch_owner_invalidates_credentials() -> None:
|
|
"""REQ-053 AC-4: owner 재할당 → 200, BrokerCredential 자동 삭제(무효화)."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_before = _make_daemon(owner_user_id=OWNER_A_ID)
|
|
daemon_after = _make_daemon(owner_user_id=OWNER_B_ID)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_before
|
|
mock_repo.update_daemon_owner.return_value = daemon_after
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
f"/v1/admin/daemons/{DAEMON_ID}",
|
|
json={"owner_user_id": OWNER_B_ID},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["owner_user_id"] == OWNER_B_ID
|
|
# update_daemon_owner must be called (includes credential deletion)
|
|
mock_repo.update_daemon_owner.assert_called_once_with(DAEMON_ID, owner_user_id=OWNER_B_ID)
|
|
|
|
|
|
def test_admin_patch_referenced_daemon_owner_returns_409() -> None:
|
|
"""Issue #27: assigned daemon owner transfer fails without a partial mutation."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_before = _make_daemon(owner_user_id=OWNER_A_ID)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_before
|
|
mock_repo.update_daemon_owner.side_effect = DaemonInUseError(
|
|
"배정된 포트폴리오가 있는 daemon의 소유자는 변경할 수 없습니다"
|
|
)
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
f"/v1/admin/daemons/{DAEMON_ID}",
|
|
json={"owner_user_id": OWNER_B_ID},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 409
|
|
mock_repo.update_daemon_owner.assert_called_once_with(DAEMON_ID, owner_user_id=OWNER_B_ID)
|
|
mock_repo.update_daemon.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-5: owner 변경 후 이전 owner의 /v1/user_info → broker=null
|
|
# (verified via update_daemon_owner deleting credentials; AC-4 covers the mechanism)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_patch_owner_then_old_owner_loses_credential() -> None:
|
|
"""REQ-053 AC-5: owner 재할당 이후 이전 owner 인증정보 제거 확인 (update_daemon_owner 호출로 검증)."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_before = _make_daemon(owner_user_id=OWNER_A_ID)
|
|
daemon_after = _make_daemon(owner_user_id=OWNER_B_ID)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_before
|
|
mock_repo.update_daemon_owner.return_value = daemon_after
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
f"/v1/admin/daemons/{DAEMON_ID}",
|
|
json={"owner_user_id": OWNER_B_ID},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
# update_daemon_owner deletes credentials as part of the operation (see postgres.py implementation)
|
|
mock_repo.update_daemon_owner.assert_called_once_with(DAEMON_ID, owner_user_id=OWNER_B_ID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-6: 비활성화 → is_active=False + service token auto-revoke
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_deactivate_daemon_revokes_tokens() -> None:
|
|
"""REQ-053 AC-6: 비활성화 → 200 + is_active=False, service token auto-revoke."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_active = _make_daemon(is_active=True)
|
|
daemon_inactive = _make_daemon(is_active=False)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_active
|
|
mock_repo.update_daemon.return_value = daemon_inactive
|
|
mock_repo.revoke_tokens_by_daemon_id.return_value = 1
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(f"/v1/admin/daemons/{DAEMON_ID}/deactivate", headers=_auth_header())
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_active"] is False
|
|
mock_repo.update_daemon.assert_called_once_with(DAEMON_ID, is_active=False)
|
|
mock_repo.revoke_tokens_by_daemon_id.assert_called_once_with(DAEMON_ID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-7: 재활성화 → is_active=True
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_activate_daemon_returns_active() -> None:
|
|
"""REQ-053 AC-7: 재활성화 → 200 + is_active=True."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_inactive = _make_daemon(is_active=False)
|
|
daemon_active = _make_daemon(is_active=True)
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = daemon_inactive
|
|
mock_repo.update_daemon.return_value = daemon_active
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(f"/v1/admin/daemons/{DAEMON_ID}/activate", headers=_auth_header())
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_active"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-8: DELETE → 405 (하드 삭제 미제공)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_delete_daemon_returns_405() -> None:
|
|
"""REQ-053 AC-8: DELETE /v1/admin/daemons/{id} → 405 Method Not Allowed."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
|
|
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), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.delete(f"/v1/admin/daemons/{DAEMON_ID}", headers=_auth_header())
|
|
|
|
assert resp.status_code == 405
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-9: user-token(비admin) → 403
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_non_admin_user_token_create_daemon_returns_403() -> None:
|
|
"""REQ-053 AC-9: user-token(non-admin) → POST /v1/admin/daemons → 403."""
|
|
settings = _make_settings()
|
|
principal = _make_user_principal() # permission=read
|
|
|
|
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), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
"/v1/admin/daemons",
|
|
json={"daemon_slug": "test", "owner_user_id": OWNER_A_ID},
|
|
headers=_auth_header("user-token"),
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_non_admin_patch_returns_403() -> None:
|
|
"""REQ-053 AC-9: user-token → PATCH /v1/admin/daemons/{id} → 403."""
|
|
settings = _make_settings()
|
|
principal = _make_user_principal()
|
|
|
|
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), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
f"/v1/admin/daemons/{DAEMON_ID}",
|
|
json={"display_name": "X"},
|
|
headers=_auth_header("user-token"),
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_non_admin_deactivate_returns_403() -> None:
|
|
"""REQ-053 AC-9: user-token → POST /v1/admin/daemons/{id}/deactivate → 403."""
|
|
settings = _make_settings()
|
|
principal = _make_user_principal()
|
|
|
|
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), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
f"/v1/admin/daemons/{DAEMON_ID}/deactivate",
|
|
headers=_auth_header("user-token"),
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-10: service-token → 403
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_service_token_admin_crud_returns_403() -> None:
|
|
"""REQ-053 AC-10: service-token → POST /v1/admin/daemons → 403."""
|
|
settings = _make_settings()
|
|
svc_principal = _make_service_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = svc_principal
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
"/v1/admin/daemons",
|
|
json={"daemon_slug": "test", "owner_user_id": OWNER_A_ID},
|
|
headers=_auth_header("svc-token"),
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-11: 비인증 → 401
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_crud_no_token_returns_401() -> None:
|
|
"""REQ-053 AC-11: 비인증 → POST /v1/admin/daemons → 401."""
|
|
settings = _make_settings()
|
|
client = _make_client(settings, MagicMock())
|
|
resp = client.post("/v1/admin/daemons", json={"daemon_slug": "test", "owner_user_id": "u1"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-053 AC-12: 미존재 daemon_id → 404
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_patch_nonexistent_daemon_returns_404() -> None:
|
|
"""REQ-053 AC-12: PATCH 미존재 daemon → 404."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = None
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.patch(
|
|
"/v1/admin/daemons/nonexistent",
|
|
json={"display_name": "X"},
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_admin_deactivate_nonexistent_daemon_returns_404() -> None:
|
|
"""REQ-053 AC-12: deactivate 미존재 daemon → 404."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.get_daemon_by_id.return_value = None
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.post(
|
|
"/v1/admin/daemons/nonexistent/deactivate",
|
|
headers=_auth_header(),
|
|
)
|
|
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BAT-022 QC Major: update_daemon_owner 원자성 — 부분 실패 시 credential leak 방지
|
|
# (dcy-011 OQ-3: 이전 owner의 KIS 인증정보 신규 owner에게 상속 불가)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_mock_conn_for_owner_update(*, fail_on_update: bool = False):
|
|
"""
|
|
psycopg Connection 모의 객체를 반환.
|
|
conn.transaction() 컨텍스트 매니저를 지원하고, 내부 cursor 실행을 추적한다.
|
|
fail_on_update=True 이면 UPDATE daemons 실행 시 예외를 발생시켜
|
|
트랜잭션 롤백을 유발한다.
|
|
"""
|
|
from contextlib import contextmanager
|
|
from unittest.mock import MagicMock
|
|
import psycopg
|
|
|
|
executed_sqls: list[str] = []
|
|
rollback_called = False
|
|
commit_called = False
|
|
|
|
class TrackingCursor:
|
|
"""execute 호출을 기록하고 SELECT/UPDATE/DELETE에 대한 결과를 제공."""
|
|
def __init__(self) -> None:
|
|
self._last_sql = ""
|
|
self._rows: list = []
|
|
|
|
def execute(self, sql: str, params=None) -> None:
|
|
sql_stripped = sql.strip()
|
|
executed_sqls.append(sql_stripped)
|
|
if "FROM portfolios" in sql_stripped:
|
|
self._rows = []
|
|
elif sql_stripped.startswith("SELECT"):
|
|
# daemon exists
|
|
self._rows = [("daemon-001",)]
|
|
elif sql_stripped.startswith("DELETE"):
|
|
self._rows = []
|
|
elif sql_stripped.startswith("UPDATE"):
|
|
if fail_on_update:
|
|
raise psycopg.DatabaseError("simulated UPDATE failure")
|
|
self._rows = [(
|
|
"daemon-001",
|
|
"user-b",
|
|
"trex-orb",
|
|
"ORB",
|
|
True,
|
|
datetime(2026, 6, 28, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 28, tzinfo=timezone.utc),
|
|
)]
|
|
|
|
def fetchone(self):
|
|
if self._rows:
|
|
return self._rows[0]
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
class FakeTransaction:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
nonlocal rollback_called, commit_called
|
|
if exc_type is not None:
|
|
rollback_called = True
|
|
return False # re-raise
|
|
commit_called = True
|
|
return False
|
|
|
|
class FakeConn:
|
|
def transaction(self):
|
|
return FakeTransaction()
|
|
|
|
def cursor(self):
|
|
return TrackingCursor()
|
|
|
|
return FakeConn(), executed_sqls, lambda: rollback_called, lambda: commit_called
|
|
|
|
|
|
def test_update_daemon_owner_executes_delete_before_update() -> None:
|
|
"""BAT-022 QC Major / dcy-011 OQ-3: credential DELETE가 owner UPDATE보다 먼저 실행됨을 확인.
|
|
|
|
안전 순서 보장: DELETE가 먼저이므로 UPDATE 실패 시 owner 미변경+credential 없음(leak 없음).
|
|
"""
|
|
from unittest.mock import patch as _patch
|
|
|
|
fake_conn, executed_sqls, is_rollback, is_commit = _make_mock_conn_for_owner_update()
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
|
|
with _patch.object(repo, "_connect", return_value=fake_conn), \
|
|
_patch.object(repo, "ensure_schema"):
|
|
result = repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
|
|
|
|
# DELETE must come before UPDATE in execution order
|
|
delete_idx = next(i for i, s in enumerate(executed_sqls) if s.startswith("DELETE"))
|
|
update_idx = next(i for i, s in enumerate(executed_sqls) if s.startswith("UPDATE"))
|
|
assert delete_idx < update_idx, (
|
|
f"DELETE must execute before UPDATE to prevent credential leak. "
|
|
f"Actual order: DELETE={delete_idx}, UPDATE={update_idx}"
|
|
)
|
|
# Owner change committed
|
|
assert is_commit()
|
|
assert result.owner_user_id == "user-b"
|
|
|
|
|
|
def test_update_daemon_owner_rollback_on_update_failure_prevents_leak() -> None:
|
|
"""BAT-022 QC Major / dcy-011 OQ-3: UPDATE daemons 실패 시 트랜잭션 롤백(rollback 경로 확인).
|
|
|
|
부분 실패 시 "새 owner + 이전 credential 잔류" 상태(credential leak)가 발생하지 않아야 함.
|
|
트랜잭션 블록 안에서 UPDATE가 실패하면 예외가 전파되고 commit은 호출되지 않는다.
|
|
"""
|
|
import psycopg
|
|
from unittest.mock import patch as _patch
|
|
|
|
fake_conn, executed_sqls, is_rollback, is_commit = _make_mock_conn_for_owner_update(fail_on_update=True)
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
|
|
with _patch.object(repo, "_connect", return_value=fake_conn), \
|
|
_patch.object(repo, "ensure_schema"):
|
|
with pytest.raises(psycopg.DatabaseError):
|
|
repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
|
|
|
|
# DELETE was attempted before UPDATE
|
|
assert any(s.startswith("DELETE") for s in executed_sqls), \
|
|
"DELETE must have been attempted before UPDATE"
|
|
# UPDATE failure triggered rollback (not commit)
|
|
assert is_rollback(), "Transaction must roll back when UPDATE fails"
|
|
assert not is_commit(), "Transaction must NOT commit on UPDATE failure"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /v1/admin/daemons — 전체 데몬 목록 (admin 전용)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_admin_list_daemons_returns_200_with_all_daemons() -> None:
|
|
"""GET /v1/admin/daemons: admin → 200 + daemons/total 형태로 전체 목록 반환."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
daemon_a = _make_daemon(daemon_id="d-001", owner_user_id=OWNER_A_ID, daemon_slug="trex-orb", display_name="ORB")
|
|
daemon_b = _make_daemon(daemon_id="d-002", owner_user_id=OWNER_B_ID, daemon_slug="trex-tsl", display_name="TSL")
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.list_daemons_all.return_value = [daemon_a, daemon_b]
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.get("/v1/admin/daemons", headers=_auth_header())
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "daemons" in body
|
|
assert "total" in body
|
|
assert body["total"] == 2
|
|
ids = {d["id"] for d in body["daemons"]}
|
|
assert ids == {"d-001", "d-002"}
|
|
# 각 항목에 필수 필드 포함 확인
|
|
for item in body["daemons"]:
|
|
assert "id" in item
|
|
assert "daemon_slug" in item
|
|
assert "display_name" in item
|
|
assert "is_active" in item
|
|
assert "owner_user_id" in item
|
|
# 시크릿 필드 없음 확인
|
|
assert "app_key" not in body["daemons"][0]
|
|
assert "app_secret" not in body["daemons"][0]
|
|
assert "credential_payload" not in body["daemons"][0]
|
|
|
|
|
|
def test_admin_list_daemons_empty_returns_200() -> None:
|
|
"""GET /v1/admin/daemons: 데몬 없을 때 → 200 + daemons=[] total=0."""
|
|
settings = _make_settings()
|
|
principal = _make_admin_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.list_daemons_all.return_value = []
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.get("/v1/admin/daemons", headers=_auth_header())
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["daemons"] == []
|
|
assert body["total"] == 0
|
|
|
|
|
|
def test_admin_list_daemons_non_admin_returns_403() -> None:
|
|
"""GET /v1/admin/daemons: 비admin user token → 403."""
|
|
settings = _make_settings()
|
|
principal = _make_user_principal()
|
|
|
|
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), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.get("/v1/admin/daemons", headers=_auth_header("user-token"))
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_admin_list_daemons_no_token_returns_401() -> None:
|
|
"""GET /v1/admin/daemons: 미인증 → 401."""
|
|
settings = _make_settings()
|
|
client = _make_client(settings, MagicMock())
|
|
resp = client.get("/v1/admin/daemons")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_admin_list_daemons_service_token_returns_403() -> None:
|
|
"""GET /v1/admin/daemons: service-token → 403."""
|
|
settings = _make_settings()
|
|
svc_principal = _make_service_principal()
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = svc_principal
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_admin_daemons.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
client = _make_client(settings, mock_repo)
|
|
resp = client.get("/v1/admin/daemons", headers=_auth_header("svc-token"))
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_update_daemon_owner_both_ops_in_single_transaction() -> None:
|
|
"""BAT-022 QC Major: DELETE + UPDATE가 단일 conn.transaction() 블록 안에서 실행됨을 확인.
|
|
|
|
두 연산이 별도 커넥션이 아닌 같은 트랜잭션 안에 묶여 있어야 원자성이 보장된다.
|
|
"""
|
|
from unittest.mock import patch as _patch, call as _call
|
|
|
|
transaction_enter_count = [0]
|
|
executed_sqls: list[str] = []
|
|
|
|
class CountingTransaction:
|
|
def __enter__(self):
|
|
transaction_enter_count[0] += 1
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
class TrackingCursor:
|
|
def execute(self, sql: str, params=None) -> None:
|
|
executed_sqls.append(sql.strip())
|
|
|
|
def fetchone(self):
|
|
sql = executed_sqls[-1] if executed_sqls else ""
|
|
if "FROM portfolios" in sql:
|
|
return None
|
|
if sql.startswith("SELECT"):
|
|
return ("daemon-001",)
|
|
if sql.startswith("UPDATE"):
|
|
return (
|
|
"daemon-001", "user-b", "trex-orb", "ORB", True,
|
|
datetime(2026, 6, 28, tzinfo=timezone.utc),
|
|
datetime(2026, 6, 28, tzinfo=timezone.utc),
|
|
)
|
|
return None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
class FakeConn:
|
|
def transaction(self):
|
|
return CountingTransaction()
|
|
|
|
def cursor(self):
|
|
return TrackingCursor()
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
|
|
with _patch.object(repo, "_connect", return_value=FakeConn()), \
|
|
_patch.object(repo, "ensure_schema"):
|
|
repo.update_daemon_owner("daemon-001", owner_user_id="user-b")
|
|
|
|
# Exactly one transaction block entered (not two separate autocommit calls)
|
|
assert transaction_enter_count[0] == 1, (
|
|
f"Expected 1 transaction block, got {transaction_enter_count[0]}. "
|
|
"DELETE and UPDATE must be inside a single transaction."
|
|
)
|