mirror of
https://github.com/elisiariocouto/leggen.git
synced 2025-12-13 11:22:21 +00:00
feat(analytics): Fix transaction limits and improve chart legends
Co-authored-by: elisiariocouto <818914+elisiariocouto@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
committed by
Elisiário Couto
Elisiário Couto
parent
692bee574e
commit
e136fc4b75
118
tests/unit/test_analytics_fix.py
Normal file
118
tests/unit/test_analytics_fix.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""Tests for analytics fixes to ensure all transactions are used in statistics."""
|
||||
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import Mock, AsyncMock
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from leggend.main import create_app
|
||||
from leggend.services.database_service import DatabaseService
|
||||
|
||||
|
||||
class TestAnalyticsFix:
|
||||
"""Test analytics fixes for transaction limits"""
|
||||
|
||||
@pytest.fixture
|
||||
def client(self):
|
||||
app = create_app()
|
||||
return TestClient(app)
|
||||
|
||||
@pytest.fixture
|
||||
def mock_database_service(self):
|
||||
return Mock(spec=DatabaseService)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_stats_uses_all_transactions(self, client, mock_database_service):
|
||||
"""Test that transaction stats endpoint uses all transactions (not limited to 100)"""
|
||||
# Mock data for 600 transactions (simulating the issue)
|
||||
mock_transactions = []
|
||||
for i in range(600):
|
||||
mock_transactions.append({
|
||||
"transactionId": f"txn-{i}",
|
||||
"transactionDate": (datetime.now() - timedelta(days=i % 365)).isoformat(),
|
||||
"description": f"Transaction {i}",
|
||||
"transactionValue": 10.0 if i % 2 == 0 else -5.0,
|
||||
"transactionCurrency": "EUR",
|
||||
"transactionStatus": "booked",
|
||||
"accountId": f"account-{i % 3}",
|
||||
})
|
||||
|
||||
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
||||
|
||||
# Test that the endpoint calls get_transactions_from_db with limit=None
|
||||
with client as test_client:
|
||||
# Replace the database service in the route handler
|
||||
from leggend.api.routes import transactions
|
||||
original_service = transactions.database_service
|
||||
transactions.database_service = mock_database_service
|
||||
|
||||
try:
|
||||
response = test_client.get("/api/v1/transactions/stats?days=365")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Verify that limit=None was passed to get all transactions
|
||||
mock_database_service.get_transactions_from_db.assert_called_once()
|
||||
call_args = mock_database_service.get_transactions_from_db.call_args
|
||||
assert call_args.kwargs.get("limit") is None, "Stats endpoint should pass limit=None to get all transactions"
|
||||
|
||||
# Verify that the response contains stats for all 600 transactions
|
||||
assert data["success"] is True
|
||||
stats = data["data"]
|
||||
assert stats["total_transactions"] == 600, "Should process all 600 transactions, not just 100"
|
||||
|
||||
# Verify calculations are correct for all transactions
|
||||
expected_income = sum(txn["transactionValue"] for txn in mock_transactions if txn["transactionValue"] > 0)
|
||||
expected_expenses = sum(abs(txn["transactionValue"]) for txn in mock_transactions if txn["transactionValue"] < 0)
|
||||
|
||||
assert stats["total_income"] == expected_income
|
||||
assert stats["total_expenses"] == expected_expenses
|
||||
|
||||
finally:
|
||||
# Restore original service
|
||||
transactions.database_service = original_service
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analytics_endpoint_returns_all_transactions(self, client, mock_database_service):
|
||||
"""Test that the new analytics endpoint returns all transactions without pagination"""
|
||||
# Mock data for 600 transactions
|
||||
mock_transactions = []
|
||||
for i in range(600):
|
||||
mock_transactions.append({
|
||||
"transactionId": f"txn-{i}",
|
||||
"transactionDate": (datetime.now() - timedelta(days=i % 365)).isoformat(),
|
||||
"description": f"Transaction {i}",
|
||||
"transactionValue": 10.0 if i % 2 == 0 else -5.0,
|
||||
"transactionCurrency": "EUR",
|
||||
"transactionStatus": "booked",
|
||||
"accountId": f"account-{i % 3}",
|
||||
})
|
||||
|
||||
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
||||
|
||||
with client as test_client:
|
||||
# Replace the database service in the route handler
|
||||
from leggend.api.routes import transactions
|
||||
original_service = transactions.database_service
|
||||
transactions.database_service = mock_database_service
|
||||
|
||||
try:
|
||||
response = test_client.get("/api/v1/transactions/analytics?days=365")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Verify that limit=None was passed to get all transactions
|
||||
mock_database_service.get_transactions_from_db.assert_called_once()
|
||||
call_args = mock_database_service.get_transactions_from_db.call_args
|
||||
assert call_args.kwargs.get("limit") is None, "Analytics endpoint should pass limit=None"
|
||||
|
||||
# Verify that all 600 transactions are returned
|
||||
assert data["success"] is True
|
||||
transactions_data = data["data"]
|
||||
assert len(transactions_data) == 600, "Analytics endpoint should return all 600 transactions"
|
||||
|
||||
finally:
|
||||
# Restore original service
|
||||
transactions.database_service = original_service
|
||||
@@ -12,6 +12,7 @@ from leggen.database.sqlite import persist_balances, get_balances
|
||||
|
||||
class MockContext:
|
||||
"""Mock context for testing."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
@@ -24,15 +25,15 @@ class TestConfigurablePaths:
|
||||
# Reset path manager
|
||||
original_config = path_manager._config_dir
|
||||
original_db = path_manager._database_path
|
||||
|
||||
|
||||
try:
|
||||
path_manager._config_dir = None
|
||||
path_manager._database_path = None
|
||||
|
||||
|
||||
# Test defaults
|
||||
config_dir = path_manager.get_config_dir()
|
||||
db_path = path_manager.get_database_path()
|
||||
|
||||
|
||||
assert config_dir == Path.home() / ".config" / "leggen"
|
||||
assert db_path == Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
finally:
|
||||
@@ -44,22 +45,25 @@ class TestConfigurablePaths:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
test_config_dir = Path(tmpdir) / "test-config"
|
||||
test_db_path = Path(tmpdir) / "test.db"
|
||||
|
||||
with patch.dict(os.environ, {
|
||||
'LEGGEN_CONFIG_DIR': str(test_config_dir),
|
||||
'LEGGEN_DATABASE_PATH': str(test_db_path)
|
||||
}):
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"LEGGEN_CONFIG_DIR": str(test_config_dir),
|
||||
"LEGGEN_DATABASE_PATH": str(test_db_path),
|
||||
},
|
||||
):
|
||||
# Reset path manager to pick up environment variables
|
||||
original_config = path_manager._config_dir
|
||||
original_db = path_manager._database_path
|
||||
|
||||
|
||||
try:
|
||||
path_manager._config_dir = None
|
||||
path_manager._database_path = None
|
||||
|
||||
|
||||
config_dir = path_manager.get_config_dir()
|
||||
db_path = path_manager.get_database_path()
|
||||
|
||||
|
||||
assert config_dir == test_config_dir
|
||||
assert db_path == test_db_path
|
||||
finally:
|
||||
@@ -71,20 +75,25 @@ class TestConfigurablePaths:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
test_config_dir = Path(tmpdir) / "explicit-config"
|
||||
test_db_path = Path(tmpdir) / "explicit.db"
|
||||
|
||||
|
||||
# Save original paths
|
||||
original_config = path_manager._config_dir
|
||||
original_db = path_manager._database_path
|
||||
|
||||
|
||||
try:
|
||||
# Set explicit paths
|
||||
path_manager.set_config_dir(test_config_dir)
|
||||
path_manager.set_database_path(test_db_path)
|
||||
|
||||
|
||||
assert path_manager.get_config_dir() == test_config_dir
|
||||
assert path_manager.get_database_path() == test_db_path
|
||||
assert path_manager.get_config_file_path() == test_config_dir / "config.toml"
|
||||
assert path_manager.get_auth_file_path() == test_config_dir / "auth.json"
|
||||
assert (
|
||||
path_manager.get_config_file_path()
|
||||
== test_config_dir / "config.toml"
|
||||
)
|
||||
assert (
|
||||
path_manager.get_auth_file_path() == test_config_dir / "auth.json"
|
||||
)
|
||||
finally:
|
||||
# Restore original paths
|
||||
path_manager._config_dir = original_config
|
||||
@@ -94,14 +103,14 @@ class TestConfigurablePaths:
|
||||
"""Test that database operations work with custom paths."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp_file:
|
||||
test_db_path = Path(tmp_file.name)
|
||||
|
||||
|
||||
# Save original database path
|
||||
original_db = path_manager._database_path
|
||||
|
||||
|
||||
try:
|
||||
# Set custom database path
|
||||
path_manager.set_database_path(test_db_path)
|
||||
|
||||
|
||||
# Test database operations
|
||||
ctx = MockContext()
|
||||
balance = {
|
||||
@@ -114,20 +123,20 @@ class TestConfigurablePaths:
|
||||
"type": "available",
|
||||
"timestamp": "2023-01-01T00:00:00",
|
||||
}
|
||||
|
||||
|
||||
# Persist balance
|
||||
persist_balances(ctx, balance)
|
||||
|
||||
|
||||
# Retrieve balances
|
||||
balances = get_balances()
|
||||
|
||||
|
||||
assert len(balances) == 1
|
||||
assert balances[0]["account_id"] == "test-account"
|
||||
assert balances[0]["amount"] == 1000.0
|
||||
|
||||
|
||||
# Verify database file exists at custom location
|
||||
assert test_db_path.exists()
|
||||
|
||||
|
||||
finally:
|
||||
# Restore original path and cleanup
|
||||
path_manager._database_path = original_db
|
||||
@@ -139,24 +148,24 @@ class TestConfigurablePaths:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
test_config_dir = Path(tmpdir) / "new" / "config" / "dir"
|
||||
test_db_path = Path(tmpdir) / "new" / "db" / "dir" / "test.db"
|
||||
|
||||
|
||||
# Save original paths
|
||||
original_config = path_manager._config_dir
|
||||
original_db = path_manager._database_path
|
||||
|
||||
|
||||
try:
|
||||
# Set paths to non-existent directories
|
||||
path_manager.set_config_dir(test_config_dir)
|
||||
path_manager.set_database_path(test_db_path)
|
||||
|
||||
|
||||
# Ensure directories are created
|
||||
path_manager.ensure_config_dir_exists()
|
||||
path_manager.ensure_database_dir_exists()
|
||||
|
||||
|
||||
assert test_config_dir.exists()
|
||||
assert test_db_path.parent.exists()
|
||||
|
||||
|
||||
finally:
|
||||
# Restore original paths
|
||||
path_manager._config_dir = original_config
|
||||
path_manager._database_path = original_db
|
||||
path_manager._database_path = original_db
|
||||
|
||||
@@ -23,11 +23,11 @@ def temp_db_path():
|
||||
def mock_home_db_path(temp_db_path):
|
||||
"""Mock the database path to use temp file."""
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
# Set the path manager to use the temporary database
|
||||
original_database_path = path_manager._database_path
|
||||
path_manager.set_database_path(temp_db_path)
|
||||
|
||||
|
||||
try:
|
||||
yield temp_db_path
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user