feat(REQ-001): add portfolio comment history API, CLI, repository, migration, and tests
Agent-Logs-Url: https://github.com/genichin/portfolio-hub/sessions/64fc2cec-78f5-41d1-99a4-45f868781edd Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
co-authored by
genichin
genichin
parent
4684597353
commit
e8f782bec6
+155
-2
@@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import app.cli.commands.health as health_commands
|
||||
@@ -8,7 +9,7 @@ from typer.testing import CliRunner
|
||||
|
||||
from app.cli.main import create_app
|
||||
from app.core.config import Settings
|
||||
from app.domain.models import LedgerEntry, Portfolio, StrategySet, User
|
||||
from app.domain.models import LedgerEntry, Portfolio, PortfolioComment, StrategySet, User
|
||||
|
||||
|
||||
class FakeRepo:
|
||||
@@ -19,6 +20,7 @@ class FakeRepo:
|
||||
self._strategy_sets: dict[str, StrategySet] = {self.strategy_set.id: self.strategy_set}
|
||||
self._users: dict[str, User] = {}
|
||||
self._user_seq = 1
|
||||
self._comments: dict[str, PortfolioComment] = {}
|
||||
self.closed = False
|
||||
|
||||
def close(self) -> None:
|
||||
@@ -262,6 +264,47 @@ class FakeRepo:
|
||||
if clear_ledger:
|
||||
self.entries = []
|
||||
|
||||
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 _settings() -> Settings:
|
||||
return Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000")
|
||||
@@ -525,4 +568,114 @@ def test_cli_user_update_and_delete_flow() -> None:
|
||||
assert update_result.exit_code == 0
|
||||
assert "alice-updated" in update_result.stdout
|
||||
assert delete_result.exit_code == 0
|
||||
assert f"Deleted user: {user.id}" in delete_result.stdout
|
||||
assert f"Deleted user: {user.id}" in delete_result.stdout
|
||||
|
||||
def test_cli_comment_add_and_list() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
runner = build_runner(repo)
|
||||
|
||||
add_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "comment", "add", "005930:tv", "--content", "First comment"],
|
||||
)
|
||||
list_result = runner.invoke(runner.app, ["portfolio", "comment", "list", "005930:tv"])
|
||||
|
||||
assert add_result.exit_code == 0
|
||||
assert "First comment" in add_result.stdout
|
||||
assert list_result.exit_code == 0
|
||||
assert "First comment" in list_result.stdout
|
||||
|
||||
|
||||
def test_cli_comment_add_supports_json_output() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "comment", "add", "005930:tv", "--content", "JSON comment"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.stdout)
|
||||
assert payload["content"] == "JSON comment"
|
||||
assert payload["portfolio_id"] == "005930:tv"
|
||||
assert "id" in payload
|
||||
|
||||
|
||||
def test_cli_comment_update_flow() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
runner = build_runner(repo)
|
||||
|
||||
add_result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "comment", "add", "005930:tv", "--content", "Original"],
|
||||
)
|
||||
assert add_result.exit_code == 0
|
||||
comment_id = json.loads(add_result.stdout)["id"]
|
||||
|
||||
update_result = runner.invoke(
|
||||
runner.app,
|
||||
[
|
||||
"--format",
|
||||
"json",
|
||||
"portfolio",
|
||||
"comment",
|
||||
"update",
|
||||
"005930:tv",
|
||||
comment_id,
|
||||
"--content",
|
||||
"Updated",
|
||||
"--force",
|
||||
],
|
||||
)
|
||||
|
||||
assert update_result.exit_code == 0
|
||||
updated = json.loads(update_result.stdout)
|
||||
assert updated["content"] == "Updated"
|
||||
assert updated["id"] == comment_id
|
||||
|
||||
|
||||
def test_cli_comment_delete_flow() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
runner = build_runner(repo)
|
||||
|
||||
add_result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "comment", "add", "005930:tv", "--content", "To delete"],
|
||||
)
|
||||
assert add_result.exit_code == 0
|
||||
comment_id = json.loads(add_result.stdout)["id"]
|
||||
|
||||
delete_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "comment", "delete", "005930:tv", comment_id, "--force"],
|
||||
)
|
||||
list_result = runner.invoke(runner.app, ["portfolio", "comment", "list", "005930:tv"])
|
||||
|
||||
assert delete_result.exit_code == 0
|
||||
assert comment_id in delete_result.stdout
|
||||
assert "To delete" not in list_result.stdout
|
||||
|
||||
|
||||
def test_cli_comment_multiline_content() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
runner = build_runner(repo)
|
||||
|
||||
multiline = "Line 1\nLine 2\nLine 3"
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "comment", "add", "005930:tv", "--content", multiline],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.stdout)
|
||||
assert payload["content"] == multiline
|
||||
|
||||
|
||||
def test_cli_comment_list_nonexistent_portfolio() -> None:
|
||||
repo = FakeRepo()
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(runner.app, ["portfolio", "comment", "list", "nonexistent:id"])
|
||||
|
||||
assert result.exit_code == 1
|
||||
|
||||
@@ -0,0 +1,426 @@
|
||||
"""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):
|
||||
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):
|
||||
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",
|
||||
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"
|
||||
Reference in New Issue
Block a user