feat: implement portfolio UUID migration and v1 API finalization (BAT-006, BAT-001 re-verified)
Core Changes: - Migrate portfolio_id from string to UUID in schema and repository layer - Update portfolio CLI commands to use name+user combination for identification - Update HTTP API to accept portfolio_id as UUID path parameter - Implement portfolio ownership validation in repository queries Requirements Completed (REQ-016 to REQ-019): - REQ-016: portfolios table UUID schema migration with data migration script - REQ-017: CLI portfolio commands with name+user identification - REQ-018: HTTP API portfolio path parameters and v1 asset_id UUID returns - REQ-019: Portfolio UUID baseline documentation updates BAT-001 Re-verification: - REQ-004~010 marked as Implemented after verification evidence validation - All v1 API endpoints verified with comprehensive test coverage Documentation Updates: - Update data-model.md with portfolio UUID structure - Update interface-contract.md with UUID path parameters - Update runtime-flows.md with ownership-based resolution flows - Update project-structure.md with database layer changes - Add migration script for portfolio UUID transition Testing: - Add test_portfolio_uuid.py for migration validation - Update test_cli.py and test_portfolio_ownership.py for ownership checks
This commit is contained in:
+23
-10
@@ -208,12 +208,11 @@ class FakeRepo:
|
||||
|
||||
def create_portfolio(
|
||||
self,
|
||||
portfolio_id: str,
|
||||
name: str,
|
||||
*,
|
||||
owner_user_id: str,
|
||||
broker: str,
|
||||
status: str,
|
||||
name: str,
|
||||
symbol: str,
|
||||
cash_balance: float,
|
||||
asset_quantity: float,
|
||||
@@ -222,6 +221,7 @@ class FakeRepo:
|
||||
active_strategy_id: str | None,
|
||||
avabile_strategies: list[str] | None,
|
||||
) -> Portfolio:
|
||||
portfolio_id = str(uuid.uuid4())
|
||||
self._portfolio = Portfolio(
|
||||
portfolio_id=portfolio_id,
|
||||
broker=broker,
|
||||
@@ -238,6 +238,13 @@ class FakeRepo:
|
||||
)
|
||||
return self._portfolio
|
||||
|
||||
def get_portfolio_by_name_and_user(self, owner_user_id: str, name: str) -> Portfolio | None:
|
||||
if self._portfolio is None:
|
||||
return None
|
||||
if self._portfolio.owner_user_id == owner_user_id and self._portfolio.name == name:
|
||||
return self._portfolio
|
||||
return None
|
||||
|
||||
def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio:
|
||||
portfolio = self.get(portfolio_id)
|
||||
for key, value in fields.items():
|
||||
@@ -484,7 +491,7 @@ def _portfolio() -> Portfolio:
|
||||
portfolio_id="005930:tv",
|
||||
broker="kis",
|
||||
status="enabled",
|
||||
name="Samsung Electronics",
|
||||
name="005930:tv",
|
||||
symbol="005930",
|
||||
cash_balance=1_000_000.0,
|
||||
asset_quantity=1.0,
|
||||
@@ -511,10 +518,14 @@ def test_cli_list_renders_table() -> None:
|
||||
|
||||
|
||||
def test_cli_status_supports_json_output() -> None:
|
||||
repo = FakeRepo(_portfolio())
|
||||
repo = FakeRepo()
|
||||
user = repo.create_user("alice", permission="read")
|
||||
p = _portfolio()
|
||||
p.owner_user_id = user.id
|
||||
repo._portfolio = p
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(runner.app, ["--format", "json", "portfolio", "status", "005930:tv"])
|
||||
result = runner.invoke(runner.app, ["--format", "json", "portfolio", "status", "005930:tv", "--user", "alice"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.stdout)
|
||||
@@ -549,6 +560,8 @@ def test_cli_create_buy_and_delete_flow() -> None:
|
||||
"portfolio",
|
||||
"buy",
|
||||
"005930:tv",
|
||||
"--user",
|
||||
"alice",
|
||||
"--quantity",
|
||||
"2",
|
||||
"--price",
|
||||
@@ -560,12 +573,12 @@ def test_cli_create_buy_and_delete_flow() -> None:
|
||||
)
|
||||
assert repo.entries[0].action == "BUY"
|
||||
assert repo._portfolio.owner_user_id == user.id
|
||||
delete_result = runner.invoke(runner.app, ["portfolio", "delete", "005930:tv", "--keep-ledger", "--force"])
|
||||
delete_result = runner.invoke(runner.app, ["portfolio", "delete", "005930:tv", "--user", "alice", "--keep-ledger", "--force"])
|
||||
|
||||
assert create_result.exit_code == 0
|
||||
assert buy_result.exit_code == 0
|
||||
assert delete_result.exit_code == 0
|
||||
assert "Deleted portfolio: 005930:tv" in delete_result.stdout
|
||||
assert "Deleted portfolio: 005930:tv" in delete_result.stdout # name used as identifier
|
||||
|
||||
|
||||
def test_cli_write_command_prompts_before_create() -> None:
|
||||
@@ -590,7 +603,7 @@ def test_cli_write_command_prompts_before_create() -> None:
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Create portfolio 005930:tv for symbol 005930?" in result.stdout
|
||||
assert "Create portfolio '005930:tv' for symbol 005930 (user: alice)?" in result.stdout
|
||||
|
||||
|
||||
def test_cli_write_command_can_abort() -> None:
|
||||
@@ -627,8 +640,8 @@ def test_cli_portfolio_create_without_user_exits_error() -> None:
|
||||
["portfolio", "create", "005930:tv", "--symbol", "005930", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "--user" in result.stdout
|
||||
assert result.exit_code != 0
|
||||
assert "--user" in result.output
|
||||
|
||||
|
||||
def test_cli_portfolio_create_with_valid_user_sets_owner() -> None:
|
||||
|
||||
@@ -164,7 +164,7 @@ def test_db_enabled_env_token_without_user_context_is_rejected(monkeypatch) -> N
|
||||
response = client.post(
|
||||
"/portfolios",
|
||||
headers={"Authorization": "Bearer trade-env-token"},
|
||||
json={"portfolio_id": "new-1", "symbol": "005930"},
|
||||
json={"name": "new-1", "symbol": "005930"},
|
||||
)
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@@ -0,0 +1,563 @@
|
||||
"""Tests for BAT-006: portfolio UUID migration and name+user identification."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
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, PortfolioComment, PortfolioTag, StrategySet, User
|
||||
|
||||
|
||||
class FakeRepo:
|
||||
def __init__(self) -> None:
|
||||
self._portfolios: list[Portfolio] = []
|
||||
self._users: dict[str, User] = {}
|
||||
self._user_seq = 1
|
||||
self._strategy_set = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:abc")
|
||||
self._strategy_sets: dict[str, StrategySet] = {self._strategy_set.id: self._strategy_set}
|
||||
self._entries: list[LedgerEntry] = []
|
||||
self._comments: dict[str, PortfolioComment] = {}
|
||||
self._tags: dict[str, PortfolioTag] = {}
|
||||
self.closed = False
|
||||
|
||||
def close(self) -> None:
|
||||
self.closed = True
|
||||
|
||||
def ensure_schema(self) -> None:
|
||||
return None
|
||||
|
||||
def ping(self) -> None:
|
||||
return None
|
||||
|
||||
def create_user(
|
||||
self,
|
||||
username: str,
|
||||
*,
|
||||
email: str | None = None,
|
||||
password_hash: str | None = None,
|
||||
permission: str = "read",
|
||||
is_active: bool = True,
|
||||
) -> User:
|
||||
now = datetime.now(timezone.utc)
|
||||
user = User(
|
||||
id=f"user-{self._user_seq}",
|
||||
username=username,
|
||||
email=email,
|
||||
permission=permission,
|
||||
is_active=is_active,
|
||||
password_hash=password_hash,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
self._user_seq += 1
|
||||
self._users[user.id] = user
|
||||
return user
|
||||
|
||||
def get_user(self, user_id: str) -> User:
|
||||
user = self._users.get(user_id)
|
||||
if user is None:
|
||||
raise ValueError("사용자를 찾을 수 없습니다")
|
||||
return user
|
||||
|
||||
def get_user_by_username(self, username: str) -> User | None:
|
||||
return next((u for u in self._users.values() if u.username == username), None)
|
||||
|
||||
def list_users(self) -> list[User]:
|
||||
return list(self._users.values())
|
||||
|
||||
def create_portfolio(
|
||||
self,
|
||||
name: str,
|
||||
*,
|
||||
owner_user_id: str,
|
||||
broker: str,
|
||||
status: str,
|
||||
symbol: str,
|
||||
cash_balance: float,
|
||||
asset_quantity: float,
|
||||
asset_price: float,
|
||||
profit_buffer: float,
|
||||
active_strategy_id: str | None,
|
||||
avabile_strategies: list[str] | None,
|
||||
) -> Portfolio:
|
||||
if any(
|
||||
p.owner_user_id == owner_user_id and p.name == name
|
||||
for p in self._portfolios
|
||||
):
|
||||
raise ValueError(f"이미 사용 중인 이름입니다: {name}")
|
||||
portfolio_id = str(uuid.uuid4())
|
||||
portfolio = Portfolio(
|
||||
portfolio_id=portfolio_id,
|
||||
broker=broker,
|
||||
status=status,
|
||||
name=name,
|
||||
symbol=symbol,
|
||||
owner_user_id=owner_user_id,
|
||||
cash_balance=cash_balance,
|
||||
asset_quantity=asset_quantity,
|
||||
price=asset_price,
|
||||
profit_buffer=profit_buffer,
|
||||
active_strategy_id=active_strategy_id,
|
||||
avabile_strategies=avabile_strategies,
|
||||
)
|
||||
self._portfolios.append(portfolio)
|
||||
return portfolio
|
||||
|
||||
def get_portfolio_by_name_and_user(self, owner_user_id: str, name: str) -> Portfolio | None:
|
||||
return next(
|
||||
(p for p in self._portfolios if p.owner_user_id == owner_user_id and p.name == name),
|
||||
None,
|
||||
)
|
||||
|
||||
def get(self, portfolio_id: str) -> Portfolio:
|
||||
portfolio = next((p for p in self._portfolios if p.portfolio_id == portfolio_id), None)
|
||||
if portfolio is None:
|
||||
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
||||
return portfolio
|
||||
|
||||
def get_or_create(
|
||||
self,
|
||||
portfolio_id: str,
|
||||
symbol: str,
|
||||
*,
|
||||
name: str | None = None,
|
||||
broker: str = "kis",
|
||||
status: str = "enabled",
|
||||
active_strategy_id: str | None = None,
|
||||
avabile_strategies: list[str] | None = None,
|
||||
owner_user_id: str | None = None,
|
||||
initial_cash: float = 0.0,
|
||||
) -> Portfolio:
|
||||
existing = next((p for p in self._portfolios if p.portfolio_id == portfolio_id), None)
|
||||
if existing:
|
||||
return existing
|
||||
portfolio = Portfolio(
|
||||
portfolio_id=portfolio_id,
|
||||
broker=broker,
|
||||
status=status,
|
||||
name=name or symbol,
|
||||
symbol=symbol,
|
||||
owner_user_id=owner_user_id,
|
||||
cash_balance=initial_cash,
|
||||
active_strategy_id=active_strategy_id,
|
||||
avabile_strategies=avabile_strategies,
|
||||
)
|
||||
self._portfolios.append(portfolio)
|
||||
return portfolio
|
||||
|
||||
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: str | None = None,
|
||||
status: str | None = None,
|
||||
owner_user_id: str | None = None,
|
||||
include_tags: list[str] | None = None,
|
||||
exclude_tags: list[str] | None = None,
|
||||
match_mode: str | None = None,
|
||||
) -> list[Portfolio]:
|
||||
results = self._portfolios
|
||||
if broker is not None:
|
||||
results = [p for p in results if p.broker == broker]
|
||||
if status is not None:
|
||||
results = [p for p in results if p.status == status]
|
||||
if owner_user_id is not None:
|
||||
results = [p for p in results if p.owner_user_id == owner_user_id]
|
||||
return results
|
||||
|
||||
def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
|
||||
s = StrategySet(id="set-1", name=name, config=config, nick=f"{name}:abc")
|
||||
self._strategy_set = s
|
||||
self._strategy_sets[s.id] = s
|
||||
return s
|
||||
|
||||
def get_strategy_set(self, id: str) -> StrategySet | None:
|
||||
return self._strategy_sets.get(id)
|
||||
|
||||
def list_strategy_sets(self) -> list[StrategySet]:
|
||||
return list(self._strategy_sets.values())
|
||||
|
||||
def delete_strategy_set(self, id: str) -> None:
|
||||
self._strategy_sets.pop(id, None)
|
||||
|
||||
def save(
|
||||
self,
|
||||
portfolio: Portfolio,
|
||||
*,
|
||||
current_price: float,
|
||||
action: str,
|
||||
trade_price: float,
|
||||
quantity: float,
|
||||
fee_amount: float,
|
||||
tax_amount: float,
|
||||
total_value_after: float,
|
||||
strategy_set_id: str = "manual",
|
||||
note: str = "",
|
||||
) -> None:
|
||||
idx = next((i for i, p in enumerate(self._portfolios) if p.portfolio_id == portfolio.portfolio_id), None)
|
||||
if idx is not None:
|
||||
self._portfolios[idx] = portfolio
|
||||
self._entries.insert(
|
||||
0,
|
||||
LedgerEntry(
|
||||
portfolio_id=portfolio.portfolio_id,
|
||||
recorded_at=datetime.now(timezone.utc),
|
||||
action=action,
|
||||
price=trade_price,
|
||||
quantity=quantity,
|
||||
fee_amount=fee_amount,
|
||||
tax_amount=tax_amount,
|
||||
cash_balance_after=portfolio.cash_balance,
|
||||
asset_quantity_after=portfolio.asset_quantity,
|
||||
total_value_after=total_value_after,
|
||||
profit_buffer_after=portfolio.profit_buffer,
|
||||
strategy_set_id=strategy_set_id,
|
||||
note=note,
|
||||
),
|
||||
)
|
||||
|
||||
def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
|
||||
return [e for e in self._entries if e.portfolio_id == portfolio_id][:limit]
|
||||
|
||||
def reset(self, portfolio_id: str) -> Portfolio:
|
||||
portfolio = self.get(portfolio_id)
|
||||
portfolio.cash_balance = 0.0
|
||||
portfolio.asset_quantity = 0.0
|
||||
portfolio.price = 0.0
|
||||
portfolio.profit_buffer = 0.0
|
||||
return portfolio
|
||||
|
||||
def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio:
|
||||
portfolio = self.get(portfolio_id)
|
||||
if action == "INVEST":
|
||||
portfolio.profit_buffer += amount
|
||||
else:
|
||||
portfolio.profit_buffer -= amount
|
||||
return portfolio
|
||||
|
||||
def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio:
|
||||
portfolio = self.get(portfolio_id)
|
||||
if direction == "to-buffer":
|
||||
portfolio.cash_balance -= amount
|
||||
portfolio.profit_buffer += amount
|
||||
else:
|
||||
portfolio.cash_balance += amount
|
||||
portfolio.profit_buffer -= amount
|
||||
return portfolio
|
||||
|
||||
def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None:
|
||||
self._portfolios = [p for p in self._portfolios if p.portfolio_id != portfolio_id]
|
||||
if clear_ledger:
|
||||
self._entries = [e for e in self._entries if e.portfolio_id != portfolio_id]
|
||||
|
||||
def add_tag(self, portfolio_id: str, *, tag: str) -> PortfolioTag:
|
||||
key = f"{portfolio_id}:{tag}"
|
||||
if key in self._tags:
|
||||
raise ValueError(f"이미 연결된 태그입니다: {tag}")
|
||||
item = PortfolioTag(portfolio_id=portfolio_id, tag=tag, created_at=datetime.now(timezone.utc))
|
||||
self._tags[key] = item
|
||||
return item
|
||||
|
||||
def list_tags(self, portfolio_id: str) -> list[PortfolioTag]:
|
||||
return [t for t in self._tags.values() if t.portfolio_id == portfolio_id]
|
||||
|
||||
def remove_tag(self, portfolio_id: str, *, tag: str) -> None:
|
||||
key = f"{portfolio_id}:{tag}"
|
||||
if key not in self._tags:
|
||||
raise ValueError("태그를 찾을 수 없습니다")
|
||||
del self._tags[key]
|
||||
|
||||
def create_comment(self, portfolio_id: str, *, content: str) -> PortfolioComment:
|
||||
self.get(portfolio_id)
|
||||
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]:
|
||||
self.get(portfolio_id)
|
||||
return [c for c in self._comments.values() if c.portfolio_id == portfolio_id]
|
||||
|
||||
def update_comment(self, portfolio_id: str, comment_id: str, *, content: str) -> PortfolioComment:
|
||||
self.get(portfolio_id)
|
||||
comment = self._comments.get(comment_id)
|
||||
if comment is None:
|
||||
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:
|
||||
self.get(portfolio_id)
|
||||
if comment_id not in self._comments:
|
||||
raise ValueError("댓글을 찾을 수 없습니다")
|
||||
del self._comments[comment_id]
|
||||
|
||||
|
||||
def _settings() -> Settings:
|
||||
return Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000")
|
||||
|
||||
|
||||
def build_runner(repo: FakeRepo) -> CliRunner:
|
||||
app = create_app(settings_factory=_settings, repository_factory=lambda settings: repo)
|
||||
runner = CliRunner()
|
||||
runner.app = app
|
||||
return runner
|
||||
|
||||
|
||||
# --- REQ-016: portfolio_id is a server-generated UUID ---
|
||||
|
||||
def test_create_portfolio_assigns_uuid_id() -> None:
|
||||
repo = FakeRepo()
|
||||
alice = repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "SAMSUNG", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert len(repo._portfolios) == 1
|
||||
portfolio = repo._portfolios[0]
|
||||
assert portfolio.name == "SAMSUNG"
|
||||
assert portfolio.owner_user_id == alice.id
|
||||
try:
|
||||
uuid.UUID(portfolio.portfolio_id)
|
||||
except ValueError:
|
||||
raise AssertionError(f"portfolio_id is not a valid UUID: {portfolio.portfolio_id!r}")
|
||||
|
||||
|
||||
def test_create_returns_uuid_in_response() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "create", "MY_PORTFOLIO", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
payload = json.loads(result.stdout)
|
||||
portfolio_id = payload["portfolio"]["portfolio_id"]
|
||||
uuid.UUID(portfolio_id) # must not raise
|
||||
|
||||
|
||||
# --- REQ-017: CLI name+user identification ---
|
||||
|
||||
def test_status_resolves_by_name_and_user() -> None:
|
||||
repo = FakeRepo()
|
||||
alice = repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
create_result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "create", "MY_PORT", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
assert create_result.exit_code == 0, create_result.output
|
||||
|
||||
status_result = runner.invoke(
|
||||
runner.app,
|
||||
["--format", "json", "portfolio", "status", "MY_PORT", "--user", "alice"],
|
||||
)
|
||||
|
||||
assert status_result.exit_code == 0, status_result.output
|
||||
payload = json.loads(status_result.stdout)
|
||||
assert payload["portfolio"]["name"] == "MY_PORT"
|
||||
assert payload["portfolio"]["owner_user_id"] == alice.id
|
||||
|
||||
|
||||
def test_status_fails_when_user_not_found() -> None:
|
||||
repo = FakeRepo()
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "status", "MY_PORT", "--user", "ghost"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "ghost" in result.output
|
||||
|
||||
|
||||
def test_status_fails_when_portfolio_name_not_found_for_user() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "status", "NONEXISTENT", "--user", "alice"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "NONEXISTENT" in result.output
|
||||
|
||||
|
||||
def test_create_without_user_fails() -> None:
|
||||
repo = FakeRepo()
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "MY_PORT", "--symbol", "005930", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "--user" in result.output
|
||||
|
||||
|
||||
def test_create_with_unknown_user_fails() -> None:
|
||||
repo = FakeRepo()
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "MY_PORT", "--symbol", "005930", "--user", "ghost", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "ghost" in result.output
|
||||
|
||||
|
||||
def test_name_validation_rejects_invalid_characters() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "invalid name!", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "invalid name!" in result.output or "유효하지 않은" in result.output
|
||||
|
||||
|
||||
def test_name_validation_rejects_empty_name() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
def test_alice_and_bob_can_have_same_portfolio_name() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
repo.create_user("bob", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
alice_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "SAMSUNG", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
bob_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "SAMSUNG", "--symbol", "005930", "--user", "bob", "--force"],
|
||||
)
|
||||
|
||||
assert alice_result.exit_code == 0, alice_result.output
|
||||
assert bob_result.exit_code == 0, bob_result.output
|
||||
assert len(repo._portfolios) == 2
|
||||
portfolio_ids = {p.portfolio_id for p in repo._portfolios}
|
||||
assert len(portfolio_ids) == 2, "Alice and Bob must have distinct portfolio UUIDs"
|
||||
|
||||
|
||||
def test_buy_resolves_portfolio_by_name_and_user() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "MY_PORT", "--symbol", "005930", "--cash-balance", "1000000", "--user", "alice", "--force"],
|
||||
)
|
||||
buy_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "buy", "MY_PORT", "--user", "alice", "--quantity", "1", "--price", "70000", "--force"],
|
||||
)
|
||||
|
||||
assert buy_result.exit_code == 0, buy_result.output
|
||||
assert len(repo._entries) == 1
|
||||
assert repo._entries[0].action == "BUY"
|
||||
|
||||
|
||||
def test_delete_resolves_portfolio_by_name_and_user() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "MY_PORT", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
delete_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "delete", "MY_PORT", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert delete_result.exit_code == 0, delete_result.output
|
||||
assert "MY_PORT" in delete_result.output
|
||||
assert len(repo._portfolios) == 0
|
||||
|
||||
|
||||
def test_alice_cannot_delete_bobs_portfolio() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
repo.create_user("bob", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "SAMSUNG", "--symbol", "005930", "--user", "bob", "--force"],
|
||||
)
|
||||
delete_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "delete", "SAMSUNG", "--user", "alice", "--force"],
|
||||
)
|
||||
|
||||
assert delete_result.exit_code == 1
|
||||
assert len(repo._portfolios) == 1, "Bob's portfolio must not be deleted"
|
||||
|
||||
|
||||
def test_update_new_name_renames_portfolio() -> None:
|
||||
repo = FakeRepo()
|
||||
repo.create_user("alice", permission="read")
|
||||
runner = build_runner(repo)
|
||||
|
||||
runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "create", "OLD_NAME", "--symbol", "005930", "--user", "alice", "--force"],
|
||||
)
|
||||
rename_result = runner.invoke(
|
||||
runner.app,
|
||||
["portfolio", "update", "OLD_NAME", "--user", "alice", "--new-name", "NEW_NAME", "--force"],
|
||||
)
|
||||
|
||||
assert rename_result.exit_code == 0, rename_result.output
|
||||
portfolio = repo._portfolios[0]
|
||||
assert portfolio.name == "NEW_NAME"
|
||||
Reference in New Issue
Block a user