test(req-001): add missing dividend verification regressions and tighten catalog mappings

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-09 12:08:31 +00:00
committed by GitHub
co-authored by genichin
parent f193aa6da3
commit 1015b465bb
3 changed files with 221 additions and 22 deletions
+179
View File
@@ -2,11 +2,14 @@
from __future__ import annotations
import os
import time
from contextlib import ExitStack
from datetime import UTC, datetime
from typing import Any
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from app.core.config import Settings, get_settings
@@ -506,6 +509,54 @@ def test_dividend_updates_only_profit_buffer_and_records_executed_at_and_tax() -
assert body["source"] == "manual"
def test_dividend_tax_defaults_to_zero_when_omitted() -> None:
repo = _FakeRepo(_portfolio(cash=1000, quantity=10, price=10, buffer=500))
client = _make_client(repo)
resp = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 120, "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert resp.status_code == 201
body = resp.json()
assert body["tax"] == 0.0
assert body["cash_balance_after"] == 1000.0
assert body["asset_quantity_after"] == 10.0
assert body["profit_buffer_after"] == 620.0
def test_dividend_executed_at_uses_utc_midnight_independent_of_server_timezone(monkeypatch: pytest.MonkeyPatch) -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
original_tz = None
if hasattr(time, "tzset"):
original_tz = os.environ.get("TZ")
monkeypatch.setenv("TZ", "Asia/Seoul")
time.tzset()
try:
resp = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 1, "executed_at": "2026-09-01"},
headers=_auth(),
)
finally:
if hasattr(time, "tzset"):
if original_tz is None:
monkeypatch.delenv("TZ", raising=False)
else:
monkeypatch.setenv("TZ", original_tz)
time.tzset()
assert resp.status_code == 201
body = resp.json()
assert body["executed_at"] == "2026-09-01T00:00:00Z"
assert body["recorded_at"] != body["executed_at"]
def test_dividend_requires_executed_at_and_rejects_fee_field() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
@@ -583,6 +634,51 @@ def test_dividend_duplicate_requires_confirmation() -> None:
assert confirmed.status_code == 201
def test_dividend_with_different_decimal_amount_is_not_treated_as_duplicate() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
first = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": "1.00000000", "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert first.status_code == 201
second = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": "1.00000001", "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert second.status_code == 201
def test_reversed_dividend_is_excluded_from_future_duplicate_candidate_matching() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
first = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 50, "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert first.status_code == 201
reversal = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "REVERSAL", "target_ledger_entry_id": first.json()["ledger_entry_id"]},
headers=_auth(),
)
assert reversal.status_code == 201
replay = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 50, "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert replay.status_code == 201
def test_confirm_duplicate_false_is_accepted_for_non_dividend_actions() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
@@ -635,6 +731,89 @@ def test_dividend_reversal_decreases_profit_buffer() -> None:
assert reversal.json()["profit_buffer_after"] == 200.0
def test_duplicate_reversal_is_rejected_without_state_change() -> None:
repo = _FakeRepo(_portfolio(buffer=200))
client = _make_client(repo)
dividend = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 80, "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert dividend.status_code == 201
first_reversal = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "REVERSAL", "target_ledger_entry_id": dividend.json()["ledger_entry_id"]},
headers=_auth(),
)
assert first_reversal.status_code == 201
before_count = len(repo.entries)
second_reversal = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "REVERSAL", "target_ledger_entry_id": dividend.json()["ledger_entry_id"]},
headers=_auth(),
)
assert second_reversal.status_code == 409
assert second_reversal.json()["detail"]["code"] == "LEDGER_ENTRY_ALREADY_REVERSED"
assert len(repo.entries) == before_count
assert repo.portfolio.profit_buffer == 200.0
def test_dividend_reversal_insufficient_buffer_is_rejected_without_state_change() -> None:
repo = _FakeRepo(_portfolio(buffer=0, cash=1000, quantity=10, price=10))
client = _make_client(repo)
dividend = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 100, "executed_at": "2026-09-01", "note": "Q3"},
headers=_auth(),
)
assert dividend.status_code == 201
withdraw = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "WITHDRAW", "amount": 100},
headers=_auth(),
)
assert withdraw.status_code == 201
assert repo.portfolio.profit_buffer == 0.0
before_count = len(repo.entries)
reversal = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "REVERSAL", "target_ledger_entry_id": dividend.json()["ledger_entry_id"]},
headers=_auth(),
)
assert reversal.status_code == 409
assert reversal.json()["detail"]["code"] == "INSUFFICIENT_BUFFER"
assert len(repo.entries) == before_count
assert repo.portfolio.profit_buffer == 0.0
def test_manual_ledger_conflict_secret_is_not_rendered_in_logs(caplog: pytest.LogCaptureFixture) -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)
secret = "SENTINEL_SECRET_TOKEN_67890"
def _raise_secret(*args: Any, **kwargs: Any) -> LedgerEntry:
raise ValueError(f"synthetic persistence {secret}")
repo.record_manual_ledger = _raise_secret # type: ignore[method-assign]
caplog.clear()
caplog.set_level("INFO")
resp = client.post(
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
json={"action": "DIVIDEND", "amount": 1, "executed_at": "2026-09-01"},
headers=_auth(),
)
assert resp.status_code == 422
rendered = "\n".join(f"{record.name}:{record.getMessage()}" for record in caplog.records)
assert secret not in rendered
def test_unified_ledger_includes_dividend_fields() -> None:
repo = _FakeRepo(_portfolio())
client = _make_client(repo)