1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
"""Tests for save_snapshot in portfolio-manager."""
import pytest
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock
from shared.models import Position
class TestSaveSnapshot:
@pytest.mark.asyncio
async def test_save_snapshot_saves_to_db_and_notifies(self):
from portfolio_manager.main import save_snapshot
pos = Position(
symbol="BTCUSDT",
quantity=Decimal("0.5"),
avg_entry_price=Decimal("50000"),
current_price=Decimal("52000"),
)
tracker = MagicMock()
tracker.get_all_positions.return_value = [pos]
tracker.realized_pnl = Decimal("500")
db = AsyncMock()
notifier = AsyncMock()
log = MagicMock()
await save_snapshot(db, tracker, notifier, log)
expected_total = Decimal("0.5") * Decimal("52000") # 26000
expected_unrealized = Decimal("0.5") * (Decimal("52000") - Decimal("50000")) # 1000
db.insert_portfolio_snapshot.assert_awaited_once_with(
total_value=expected_total,
realized_pnl=Decimal("500"),
unrealized_pnl=expected_unrealized,
)
notifier.send_daily_summary.assert_awaited_once_with(
[pos], expected_total, expected_unrealized
)
log.info.assert_called_once_with(
"snapshot_saved",
total_value=str(expected_total),
positions=1,
)
@pytest.mark.asyncio
async def test_save_snapshot_empty_positions(self):
from portfolio_manager.main import save_snapshot
tracker = MagicMock()
tracker.get_all_positions.return_value = []
tracker.realized_pnl = Decimal("0")
db = AsyncMock()
notifier = AsyncMock()
log = MagicMock()
await save_snapshot(db, tracker, notifier, log)
db.insert_portfolio_snapshot.assert_awaited_once_with(
total_value=Decimal("0"),
realized_pnl=Decimal("0"),
unrealized_pnl=Decimal("0"),
)
notifier.send_daily_summary.assert_awaited_once()
log.info.assert_called_once()
|
