genichin
deaaa96c46
* docs: preserve original issue 5 body * feat: add service token rename command * test: normalize rich option errors
425 lines
14 KiB
Python
425 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
import psycopg.errors
|
|
|
|
from app.core.auth import AuthContext, Permission
|
|
from app.core.config import Settings
|
|
from app.core.config import get_settings
|
|
from app.domain.models import AuthToken
|
|
from app.main import app
|
|
from app.services.auth_service import AuthService
|
|
|
|
|
|
def _client_with_tokens() -> TestClient:
|
|
get_settings.cache_clear()
|
|
app.dependency_overrides.clear()
|
|
return TestClient(app)
|
|
|
|
|
|
def test_read_route_requires_token(monkeypatch) -> None:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
|
|
get_settings.cache_clear()
|
|
client = _client_with_tokens()
|
|
|
|
response = client.get("/portfolios")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing bearer token"
|
|
|
|
|
|
def test_read_route_accepts_read_token(monkeypatch) -> None:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
|
|
get_settings.cache_clear()
|
|
client = _client_with_tokens()
|
|
|
|
response = client.get("/portfolios", headers={"Authorization": "Bearer read-token"})
|
|
|
|
assert response.status_code == 501
|
|
assert response.json()["detail"] == "Portfolio persistence is not implemented yet"
|
|
|
|
|
|
def test_trade_route_rejects_read_token(monkeypatch) -> None:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
|
|
monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "trade-token")
|
|
get_settings.cache_clear()
|
|
client = _client_with_tokens()
|
|
|
|
response = client.post(
|
|
"/portfolios/p1/buy",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
json={"quantity": 1, "price": 1000, "fee": 0, "tax": 0, "note": ""},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "Insufficient permission"
|
|
|
|
|
|
def test_legacy_buy_sell_routes_emit_deprecation_header(monkeypatch) -> None:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "trade-token")
|
|
get_settings.cache_clear()
|
|
client = _client_with_tokens()
|
|
|
|
for path in ("/portfolios/p1/buy", "/portfolios/p1/sell"):
|
|
response = client.post(
|
|
path,
|
|
headers={"Authorization": "Bearer trade-token"},
|
|
json={"quantity": 1, "price": 1000, "fee": 0, "tax": 0, "note": ""},
|
|
)
|
|
|
|
assert response.status_code == 501
|
|
assert response.headers["Deprecation"] == "true"
|
|
|
|
|
|
def test_admin_route_accepts_admin_token(monkeypatch) -> None:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_ADMIN_TOKENS", "admin-token")
|
|
get_settings.cache_clear()
|
|
client = _client_with_tokens()
|
|
|
|
response = client.delete("/portfolios/p1", headers={"Authorization": "Bearer admin-token"})
|
|
|
|
assert response.status_code == 501
|
|
assert response.json()["detail"] == "Portfolio persistence is not implemented yet"
|
|
|
|
|
|
@dataclass
|
|
class _AuthRepo:
|
|
def __post_init__(self) -> None:
|
|
self.users: dict[str, object] = {}
|
|
self.tokens: dict[str, AuthToken] = {}
|
|
self._daemons: dict[str, object] = {}
|
|
self._seq = 1
|
|
self.rename_calls = 0
|
|
|
|
def get_user_by_username(self, username: str):
|
|
return self.users.get(username)
|
|
|
|
def get_daemon_by_id(self, daemon_id: str):
|
|
return self._daemons.get(daemon_id)
|
|
|
|
def get_service_token_by_name(self, name: str) -> AuthToken | None:
|
|
for token in self.tokens.values():
|
|
if token.name == name and token.token_type == "service" and token.revoked_at is None:
|
|
return token
|
|
return None
|
|
|
|
def rename_service_token(self, name: str, new_name: str) -> AuthToken:
|
|
self.rename_calls += 1
|
|
token = self.get_service_token_by_name(name)
|
|
if token is None:
|
|
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
|
|
if self.get_service_token_by_name(new_name) is not None:
|
|
raise psycopg.errors.UniqueViolation("duplicate service-token name")
|
|
token.name = new_name
|
|
return token
|
|
|
|
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[token.id] = token
|
|
return token
|
|
|
|
def revoke_auth_token(self, user_id: str, token_id: str) -> None:
|
|
token = self.tokens.get(token_id)
|
|
if token is None or token.user_id != user_id:
|
|
raise ValueError("토큰을 찾을 수 없습니다")
|
|
token.revoked_at = datetime.now(timezone.utc)
|
|
|
|
|
|
def _service(repo: _AuthRepo) -> AuthService:
|
|
return AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))
|
|
|
|
|
|
def test_issue_service_token_defaults_to_365_days() -> None:
|
|
from dataclasses import dataclass as _dc
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("my-token", "alice")
|
|
|
|
assert result.auth_token.expires_at is not None
|
|
assert (result.auth_token.expires_at - datetime.now(timezone.utc)).days >= 364
|
|
assert result.auth_token.token_type == "service"
|
|
assert result.auth_token.name == "my-token"
|
|
assert result.auth_token.runtime_id is not None
|
|
|
|
|
|
def test_issue_service_token_generates_distinct_runtime_ids() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
|
|
first = service.issue_service_token("my-token-1", "alice")
|
|
second = service.issue_service_token("my-token-2", "alice")
|
|
|
|
assert first.auth_token.runtime_id is not None
|
|
assert second.auth_token.runtime_id is not None
|
|
assert first.auth_token.runtime_id != second.auth_token.runtime_id
|
|
|
|
|
|
def test_issue_service_token_no_expiry() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("my-token", "alice", no_expiry=True)
|
|
|
|
assert result.auth_token.expires_at is None
|
|
|
|
|
|
def test_issue_service_token_unknown_user_raises() -> None:
|
|
repo = _AuthRepo()
|
|
service = _service(repo)
|
|
|
|
try:
|
|
service.issue_service_token("my-token", "unknown")
|
|
except ValueError as exc:
|
|
assert "unknown" in str(exc)
|
|
else:
|
|
raise AssertionError("Expected ValueError for unknown user")
|
|
|
|
|
|
def test_revoke_service_token_success() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
issued = service.issue_service_token("my-token", "alice")
|
|
|
|
revoked_at = service.revoke_service_token("my-token")
|
|
|
|
assert revoked_at is not None
|
|
assert repo.tokens[issued.auth_token.id].revoked_at is not None
|
|
|
|
|
|
def test_revoke_service_token_not_found_raises() -> None:
|
|
repo = _AuthRepo()
|
|
service = _service(repo)
|
|
|
|
try:
|
|
service.revoke_service_token("nonexistent")
|
|
except ValueError as exc:
|
|
assert "nonexistent" in str(exc)
|
|
else:
|
|
raise AssertionError("Expected ValueError for missing service token")
|
|
|
|
|
|
def test_rename_service_token_changes_only_name() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
issued = _service(repo).issue_service_token("old-name", "alice")
|
|
before = issued.auth_token
|
|
protected = (
|
|
before.id,
|
|
before.user_id,
|
|
before.token_prefix,
|
|
before.permission,
|
|
before.token_type,
|
|
before.expires_at,
|
|
before.last_used_at,
|
|
before.revoked_at,
|
|
before.created_at,
|
|
before.runtime_id,
|
|
before.daemon_id,
|
|
)
|
|
|
|
result = _service(repo).rename_service_token("old-name", "new-name")
|
|
|
|
assert result.changed is True
|
|
assert result.auth_token.name == "new-name"
|
|
assert repo.get_service_token_by_name("old-name") is None
|
|
assert repo.get_service_token_by_name("new-name") is result.auth_token
|
|
assert repo.rename_calls == 1
|
|
assert protected == (
|
|
result.auth_token.id,
|
|
result.auth_token.user_id,
|
|
result.auth_token.token_prefix,
|
|
result.auth_token.permission,
|
|
result.auth_token.token_type,
|
|
result.auth_token.expires_at,
|
|
result.auth_token.last_used_at,
|
|
result.auth_token.revoked_at,
|
|
result.auth_token.created_at,
|
|
result.auth_token.runtime_id,
|
|
result.auth_token.daemon_id,
|
|
)
|
|
|
|
|
|
def test_rename_service_token_same_name_is_noop_without_write() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
issued = _service(repo).issue_service_token("unchanged", "alice")
|
|
|
|
result = _service(repo).rename_service_token("unchanged", "unchanged")
|
|
|
|
assert result.changed is False
|
|
assert result.auth_token is issued.auth_token
|
|
assert repo.rename_calls == 0
|
|
|
|
|
|
def test_rename_service_token_missing_and_duplicate_errors() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
service.issue_service_token("source", "alice")
|
|
service.issue_service_token("target", "alice")
|
|
|
|
for old_name, new_name, message in (
|
|
("missing", "new", "서비스 토큰을 찾을 수 없습니다: missing"),
|
|
("source", "target", "이미 존재하는 서비스 토큰 이름입니다: target"),
|
|
):
|
|
try:
|
|
service.rename_service_token(old_name, new_name)
|
|
except ValueError as exc:
|
|
assert str(exc) == message
|
|
else:
|
|
raise AssertionError("Expected rename failure")
|
|
|
|
|
|
# REQ-018: daemon_id forward 보정 (service layer)
|
|
|
|
def test_issue_service_token_with_daemon_id_stores_it() -> None:
|
|
from app.domain.models import Daemon, User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
daemon = Daemon(
|
|
id="daemon-1",
|
|
owner_user_id="user-1",
|
|
daemon_slug="my-daemon",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
repo._daemons = {"daemon-1": daemon}
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("tok", "alice", no_expiry=True, daemon_id="daemon-1")
|
|
|
|
assert result.auth_token.daemon_id == "daemon-1"
|
|
|
|
|
|
def test_issue_service_token_with_nonexistent_daemon_raises() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
repo._daemons = {}
|
|
service = _service(repo)
|
|
|
|
try:
|
|
service.issue_service_token("tok", "alice", no_expiry=True, daemon_id="no-such-daemon")
|
|
except ValueError as exc:
|
|
assert "no-such-daemon" in str(exc)
|
|
else:
|
|
raise AssertionError("Expected ValueError for nonexistent daemon")
|
|
|
|
|
|
def test_issue_service_token_without_daemon_id_is_unbound() -> None:
|
|
from app.domain.models import User
|
|
|
|
repo = _AuthRepo()
|
|
repo.users["alice"] = User(id="user-1", username="alice", permission="read")
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("tok", "alice", no_expiry=True)
|
|
|
|
assert result.auth_token.daemon_id is None
|
|
|
|
|
|
# REQ-016: POST /auth/tokens 제거 — HTTP AC-2
|
|
|
|
def test_post_auth_tokens_returns_404(monkeypatch) -> None:
|
|
import os
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
|
|
from app.core.config import get_settings as _get_settings
|
|
_get_settings.cache_clear()
|
|
from app.main import app as _app
|
|
client = TestClient(_app)
|
|
|
|
response = client.post(
|
|
"/auth/tokens",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
json={"name": "test", "permission": "read"},
|
|
)
|
|
|
|
_app.dependency_overrides.clear()
|
|
assert response.status_code in (404, 405)
|
|
|
|
|
|
# REQ-034: LoginRequest extra="forbid" enforcement (dcy-005 FR-2)
|
|
|
|
def test_login_rejects_no_expiry_extra_field(monkeypatch) -> None:
|
|
"""REQ-034 AC-2: session 발급 경로에 no_expiry 시도 → HTTP 422 (extra="forbid")."""
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
client = TestClient(app)
|
|
|
|
resp = client.post(
|
|
"/auth/login",
|
|
json={"username": "anyone", "password": "any", "no_expiry": True},
|
|
)
|
|
|
|
assert resp.status_code == 422
|
|
body = resp.json()
|
|
assert any("extra" in (err.get("type") or "") for err in body.get("detail", []))
|
|
|
|
|
|
def test_login_rejects_arbitrary_extra_field(monkeypatch) -> None:
|
|
"""REQ-034 AC-3: 임의 추가 필드 → HTTP 422 (extra="forbid" 일반 동작)."""
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
client = TestClient(app)
|
|
|
|
resp = client.post(
|
|
"/auth/login",
|
|
json={"username": "anyone", "password": "any", "arbitrary_field": "x"},
|
|
)
|
|
|
|
assert resp.status_code == 422
|