430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""Tests for portfolio comment history API endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.api.portfolios as portfolios_api
|
|
from app.core.config import get_settings
|
|
from app.domain.models import LedgerEntry, Portfolio, PortfolioComment, StrategySet
|
|
from app.main import app
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, portfolio: Portfolio | None = None) -> None:
|
|
self._portfolio = portfolio
|
|
self.strategy_set = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:99914b932bd3")
|
|
self._comments: dict[str, PortfolioComment] = {}
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def ensure_schema(self) -> None:
|
|
pass
|
|
|
|
def ping(self) -> None:
|
|
pass
|
|
|
|
def list_users(self):
|
|
return []
|
|
|
|
def get(self, portfolio_id: str) -> Portfolio:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
return self._portfolio
|
|
|
|
def get_or_create(self, portfolio_id: str, symbol: str, **kwargs) -> Portfolio:
|
|
raise NotImplementedError()
|
|
|
|
def create_portfolio(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio:
|
|
portfolio = self.get(portfolio_id)
|
|
for key, value in fields.items():
|
|
if value is not None:
|
|
setattr(portfolio, key, value)
|
|
return portfolio
|
|
|
|
def list_portfolios(self, *, broker=None, status=None, owner_user_id=None):
|
|
if owner_user_id is not None and self._portfolio and self._portfolio.owner_user_id != owner_user_id:
|
|
return []
|
|
return [self._portfolio] if self._portfolio else []
|
|
|
|
def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
|
|
return self.strategy_set
|
|
|
|
def get_strategy_set(self, id: str) -> StrategySet | None:
|
|
return self.strategy_set if id == self.strategy_set.id else None
|
|
|
|
def list_strategy_sets(self):
|
|
return [self.strategy_set]
|
|
|
|
def delete_strategy_set(self, id: str) -> None:
|
|
pass
|
|
|
|
def save(self, portfolio, *, current_price, action, trade_price, quantity, fee_amount, tax_amount,
|
|
total_value_after, strategy_set_id="manual", note="") -> None:
|
|
pass
|
|
|
|
def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
|
|
return []
|
|
|
|
def reset(self, portfolio_id: str) -> Portfolio:
|
|
return self.get(portfolio_id)
|
|
|
|
def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio:
|
|
return self.get(portfolio_id)
|
|
|
|
def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio:
|
|
return self.get(portfolio_id)
|
|
|
|
def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None:
|
|
self._portfolio = None
|
|
|
|
def get_user(self, user_id: str):
|
|
raise ValueError("사용자를 찾을 수 없습니다")
|
|
|
|
def get_user_by_username(self, username: str):
|
|
return None
|
|
|
|
def create_user(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def update_user(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def delete_user(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def update_user_last_login(self, user_id: str) -> None:
|
|
pass
|
|
|
|
def create_auth_token(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def list_auth_tokens(self, user_id: str):
|
|
return []
|
|
|
|
def revoke_auth_token(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def create_bot_principal(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def get_bot_principal(self, bot_principal_id: str):
|
|
raise ValueError("봇을 찾을 수 없습니다")
|
|
|
|
def list_bot_principals(self, *, is_active=None, owner_user_id=None):
|
|
return []
|
|
|
|
def deactivate_bot_principal(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def list_bot_auth_tokens(self, bot_principal_id: str):
|
|
return []
|
|
|
|
def revoke_bot_auth_token(self, *args, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
def get_token_principal(self, token_hash: str):
|
|
return None
|
|
|
|
def create_comment(self, portfolio_id: str, *, content: str) -> PortfolioComment:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
now = datetime.now(timezone.utc)
|
|
comment = PortfolioComment(
|
|
id=str(uuid.uuid4()),
|
|
portfolio_id=portfolio_id,
|
|
content=content,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self._comments[comment.id] = comment
|
|
return comment
|
|
|
|
def list_comments(self, portfolio_id: str) -> list[PortfolioComment]:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
return sorted(
|
|
[c for c in self._comments.values() if c.portfolio_id == portfolio_id],
|
|
key=lambda c: c.created_at or datetime.min.replace(tzinfo=timezone.utc),
|
|
reverse=True,
|
|
)
|
|
|
|
def update_comment(self, portfolio_id: str, comment_id: str, *, content: str) -> PortfolioComment:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
comment = self._comments.get(comment_id)
|
|
if comment is None or comment.portfolio_id != portfolio_id:
|
|
raise ValueError("댓글을 찾을 수 없습니다")
|
|
comment.content = content
|
|
comment.updated_at = datetime.now(timezone.utc)
|
|
return comment
|
|
|
|
def delete_comment(self, portfolio_id: str, comment_id: str) -> None:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
comment = self._comments.get(comment_id)
|
|
if comment is None or comment.portfolio_id != portfolio_id:
|
|
raise ValueError("댓글을 찾을 수 없습니다")
|
|
del self._comments[comment_id]
|
|
|
|
|
|
def _portfolio() -> Portfolio:
|
|
return Portfolio(
|
|
portfolio_id="005930:tv",
|
|
broker="kis",
|
|
status="enabled",
|
|
name="Samsung Electronics",
|
|
symbol="005930",
|
|
owner_user_id="legacy-owner",
|
|
cash_balance=1_000_000.0,
|
|
asset_quantity=1.0,
|
|
price=70_000.0,
|
|
profit_buffer=50_000.0,
|
|
active_strategy_id="set-1",
|
|
avabile_strategies=["set-1"],
|
|
)
|
|
|
|
|
|
def _setup_client(monkeypatch, repo: FakeRepo) -> TestClient:
|
|
monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
|
|
monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
|
|
monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "write-token")
|
|
monkeypatch.setenv("PORTFOLIOHUB_ADMIN_TOKENS", "admin-token")
|
|
get_settings.cache_clear()
|
|
app.dependency_overrides.clear()
|
|
monkeypatch.setattr(portfolios_api, "get_portfolio_repository_from_settings", lambda settings: repo)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_create_comment_returns_201(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "First comment"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["content"] == "First comment"
|
|
assert data["portfolio_id"] == "005930:tv"
|
|
assert "id" in data
|
|
assert "created_at" in data
|
|
|
|
|
|
def test_list_comments_returns_200_sorted_by_created_desc(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "First"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "Second"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
response = client.get(
|
|
"/portfolios/005930:tv/comments",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 200
|
|
items = response.json()
|
|
assert len(items) == 2
|
|
assert items[0]["content"] == "Second"
|
|
assert items[1]["content"] == "First"
|
|
|
|
|
|
def test_update_comment_returns_200(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
create_resp = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "Original"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
comment_id = create_resp.json()["id"]
|
|
|
|
update_resp = client.patch(
|
|
f"/portfolios/005930:tv/comments/{comment_id}",
|
|
json={"content": "Updated"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert update_resp.status_code == 200
|
|
data = update_resp.json()
|
|
assert data["content"] == "Updated"
|
|
assert data["id"] == comment_id
|
|
|
|
|
|
def test_delete_comment_returns_204(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
create_resp = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "To delete"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
comment_id = create_resp.json()["id"]
|
|
|
|
delete_resp = client.delete(
|
|
f"/portfolios/005930:tv/comments/{comment_id}",
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
list_resp = client.get(
|
|
"/portfolios/005930:tv/comments",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert delete_resp.status_code == 204
|
|
assert all(c["id"] != comment_id for c in list_resp.json())
|
|
|
|
|
|
def test_create_comment_nonexistent_portfolio_returns_404(monkeypatch) -> None:
|
|
repo = FakeRepo()
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.post(
|
|
"/portfolios/nonexistent/comments",
|
|
json={"content": "Should fail"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_list_comments_nonexistent_portfolio_returns_404(monkeypatch) -> None:
|
|
repo = FakeRepo()
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.get(
|
|
"/portfolios/nonexistent/comments",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_update_nonexistent_comment_returns_404(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.patch(
|
|
"/portfolios/005930:tv/comments/no-such-id",
|
|
json={"content": "Updated"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_delete_nonexistent_comment_returns_404(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.delete(
|
|
"/portfolios/005930:tv/comments/no-such-id",
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_create_comment_without_token_returns_401(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "No auth"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_create_comment_with_read_token_returns_403(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
response = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "Should be forbidden"},
|
|
headers={"Authorization": "Bearer read-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_comment_multiline_text_stored_and_retrieved(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
multiline = "Line 1\nLine 2\nLine 3"
|
|
create_resp = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": multiline},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
comment_id = create_resp.json()["id"]
|
|
|
|
list_resp = client.get(
|
|
"/portfolios/005930:tv/comments",
|
|
headers={"Authorization": "Bearer read-token"},
|
|
)
|
|
|
|
app.dependency_overrides.clear()
|
|
assert create_resp.status_code == 201
|
|
assert create_resp.json()["content"] == multiline
|
|
matching = [c for c in list_resp.json() if c["id"] == comment_id]
|
|
assert matching[0]["content"] == multiline
|
|
|
|
|
|
def test_update_comment_preserves_id_and_updates_updated_at(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
client = _setup_client(monkeypatch, repo)
|
|
|
|
create_resp = client.post(
|
|
"/portfolios/005930:tv/comments",
|
|
json={"content": "Original"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
original = create_resp.json()
|
|
comment_id = original["id"]
|
|
|
|
update_resp = client.patch(
|
|
f"/portfolios/005930:tv/comments/{comment_id}",
|
|
json={"content": "Modified"},
|
|
headers={"Authorization": "Bearer write-token"},
|
|
)
|
|
updated = update_resp.json()
|
|
|
|
app.dependency_overrides.clear()
|
|
assert updated["id"] == comment_id
|
|
assert updated["content"] == "Modified"
|