mirror of
https://github.com/elisiariocouto/leggen.git
synced 2025-12-13 11:22:21 +00:00
fix: Resolve all CI failures - linting, typing, and test issues
Co-authored-by: elisiariocouto <818914+elisiariocouto@users.noreply.github.com>
This commit is contained in:
committed by
Elisiário Couto
parent
5987a759b8
commit
c8f0a103c6
@@ -48,7 +48,7 @@ export default function MonthlyTrends({ className, days = 365 }: MonthlyTrendsPr
|
|||||||
const monthlyMap: { [key: string]: MonthlyData } = {};
|
const monthlyMap: { [key: string]: MonthlyData } = {};
|
||||||
|
|
||||||
transactions.forEach((transaction) => {
|
transactions.forEach((transaction) => {
|
||||||
const date = new Date(transaction.date);
|
const date = new Date(transaction.transaction_date);
|
||||||
const monthKey = `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}`;
|
const monthKey = `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}`;
|
||||||
|
|
||||||
if (!monthlyMap[monthKey]) {
|
if (!monthlyMap[monthKey]) {
|
||||||
@@ -63,10 +63,10 @@ export default function MonthlyTrends({ className, days = 365 }: MonthlyTrendsPr
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
if (transaction.amount > 0) {
|
if (transaction.transaction_value > 0) {
|
||||||
monthlyMap[monthKey].income += transaction.amount;
|
monthlyMap[monthKey].income += transaction.transaction_value;
|
||||||
} else {
|
} else {
|
||||||
monthlyMap[monthKey].expenses += Math.abs(transaction.amount);
|
monthlyMap[monthKey].expenses += Math.abs(transaction.transaction_value);
|
||||||
}
|
}
|
||||||
|
|
||||||
monthlyMap[monthKey].net = monthlyMap[monthKey].income - monthlyMap[monthKey].expenses;
|
monthlyMap[monthKey].net = monthlyMap[monthKey].income - monthlyMap[monthKey].expenses;
|
||||||
|
|||||||
@@ -3,7 +3,6 @@
|
|||||||
import click
|
import click
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from leggen.utils.paths import path_manager
|
|
||||||
|
|
||||||
|
|
||||||
@click.command()
|
@click.command()
|
||||||
|
|||||||
@@ -581,7 +581,7 @@ def get_historical_balances(account_id=None, days=365):
|
|||||||
|
|
||||||
# Calculate historical balances by working backwards from current balance
|
# Calculate historical balances by working backwards from current balance
|
||||||
historical_balances = []
|
historical_balances = []
|
||||||
account_running_balances = {}
|
account_running_balances: dict[str, dict[str, float]] = {}
|
||||||
|
|
||||||
# Initialize running balances with current balances
|
# Initialize running balances with current balances
|
||||||
for (acc_id, balance_type), balance_info in current_balances.items():
|
for (acc_id, balance_type), balance_info in current_balances.items():
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
"""Centralized path management for Leggen."""
|
"""Centralized path management for Leggen."""
|
||||||
|
|
||||||
|
import contextlib
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
@@ -47,12 +48,8 @@ class PathManager:
|
|||||||
db_path = self.get_config_dir() / "leggen.db"
|
db_path = self.get_config_dir() / "leggen.db"
|
||||||
|
|
||||||
# Try to ensure the directory exists, but handle permission errors gracefully
|
# Try to ensure the directory exists, but handle permission errors gracefully
|
||||||
try:
|
with contextlib.suppress(PermissionError, OSError):
|
||||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
except (PermissionError, OSError):
|
|
||||||
# If we can't create the directory, continue anyway
|
|
||||||
# This allows tests and error cases to work as expected
|
|
||||||
pass
|
|
||||||
|
|
||||||
return db_path
|
return db_path
|
||||||
|
|
||||||
|
|||||||
@@ -224,7 +224,7 @@ async def get_historical_balances(
|
|||||||
try:
|
try:
|
||||||
# Get historical balances from database
|
# Get historical balances from database
|
||||||
historical_balances = await database_service.get_historical_balances_from_db(
|
historical_balances = await database_service.get_historical_balances_from_db(
|
||||||
account_id=account_id, days=days
|
account_id=account_id, days=days or 365
|
||||||
)
|
)
|
||||||
|
|
||||||
return APIResponse(
|
return APIResponse(
|
||||||
|
|||||||
@@ -1,22 +1,22 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""Sample database generator for Leggen testing and development."""
|
"""Sample database generator for Leggen testing and development."""
|
||||||
|
|
||||||
import argparse
|
|
||||||
import json
|
import json
|
||||||
import random
|
import random
|
||||||
import sqlite3
|
import sqlite3
|
||||||
import sys
|
import sys
|
||||||
import os
|
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import List, Dict, Any
|
from typing import List, Dict, Any
|
||||||
|
|
||||||
|
import click
|
||||||
|
|
||||||
# Add the project root to the Python path
|
# Add the project root to the Python path
|
||||||
project_root = Path(__file__).parent.parent
|
project_root = Path(__file__).parent.parent
|
||||||
sys.path.insert(0, str(project_root))
|
sys.path.insert(0, str(project_root))
|
||||||
|
|
||||||
import click
|
# Import after path setup - this is necessary for the script to work
|
||||||
from leggen.utils.paths import path_manager
|
from leggen.utils.paths import path_manager # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
class SampleDataGenerator:
|
class SampleDataGenerator:
|
||||||
@@ -464,14 +464,14 @@ class SampleDataGenerator:
|
|||||||
|
|
||||||
# Print summary
|
# Print summary
|
||||||
click.echo("\n✅ Sample database created successfully!")
|
click.echo("\n✅ Sample database created successfully!")
|
||||||
click.echo(f"📊 Summary:")
|
click.echo("📊 Summary:")
|
||||||
click.echo(f" - Accounts: {len(accounts)}")
|
click.echo(f" - Accounts: {len(accounts)}")
|
||||||
click.echo(f" - Transactions: {len(transactions)}")
|
click.echo(f" - Transactions: {len(transactions)}")
|
||||||
click.echo(f" - Balances: {len(balances)}")
|
click.echo(f" - Balances: {len(balances)}")
|
||||||
click.echo(f" - Database: {self.db_path}")
|
click.echo(f" - Database: {self.db_path}")
|
||||||
|
|
||||||
# Show account details
|
# Show account details
|
||||||
click.echo(f"\n📋 Sample accounts:")
|
click.echo("\n📋 Sample accounts:")
|
||||||
for account in accounts:
|
for account in accounts:
|
||||||
institution_name = next(
|
institution_name = next(
|
||||||
inst["name"]
|
inst["name"]
|
||||||
@@ -533,12 +533,12 @@ def main(database: Path, accounts: int, transactions: int, force: bool):
|
|||||||
generator.generate_sample_database(accounts, transactions)
|
generator.generate_sample_database(accounts, transactions)
|
||||||
|
|
||||||
# Show usage instructions
|
# Show usage instructions
|
||||||
click.echo(f"\n🚀 Usage instructions:")
|
click.echo("\n🚀 Usage instructions:")
|
||||||
click.echo(f"To use this sample database with leggen commands:")
|
click.echo("To use this sample database with leggen commands:")
|
||||||
click.echo(f" export LEGGEN_DATABASE_PATH={db_path}")
|
click.echo(f" export LEGGEN_DATABASE_PATH={db_path}")
|
||||||
click.echo(f" leggen transactions")
|
click.echo(" leggen transactions")
|
||||||
click.echo(f"")
|
click.echo("")
|
||||||
click.echo(f"To use this sample database with leggend API:")
|
click.echo("To use this sample database with leggend API:")
|
||||||
click.echo(f" leggend --database {db_path}")
|
click.echo(f" leggend --database {db_path}")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from unittest.mock import Mock, AsyncMock
|
from unittest.mock import Mock, AsyncMock, patch
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
from leggend.main import create_app
|
from leggend.main import create_app
|
||||||
@@ -22,7 +22,7 @@ class TestAnalyticsFix:
|
|||||||
return Mock(spec=DatabaseService)
|
return Mock(spec=DatabaseService)
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_transaction_stats_uses_all_transactions(self, client, mock_database_service):
|
async def test_transaction_stats_uses_all_transactions(self, mock_database_service):
|
||||||
"""Test that transaction stats endpoint uses all transactions (not limited to 100)"""
|
"""Test that transaction stats endpoint uses all transactions (not limited to 100)"""
|
||||||
# Mock data for 600 transactions (simulating the issue)
|
# Mock data for 600 transactions (simulating the issue)
|
||||||
mock_transactions = []
|
mock_transactions = []
|
||||||
@@ -40,41 +40,34 @@ class TestAnalyticsFix:
|
|||||||
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
||||||
|
|
||||||
# Test that the endpoint calls get_transactions_from_db with limit=None
|
# Test that the endpoint calls get_transactions_from_db with limit=None
|
||||||
with client as test_client:
|
with patch('leggend.api.routes.transactions.database_service', mock_database_service):
|
||||||
# Replace the database service in the route handler
|
app = create_app()
|
||||||
from leggend.api.routes import transactions
|
client = TestClient(app)
|
||||||
original_service = transactions.database_service
|
|
||||||
transactions.database_service = mock_database_service
|
response = client.get("/api/v1/transactions/stats?days=365")
|
||||||
|
|
||||||
try:
|
assert response.status_code == 200
|
||||||
response = test_client.get("/api/v1/transactions/stats?days=365")
|
data = response.json()
|
||||||
|
|
||||||
assert response.status_code == 200
|
# Verify that limit=None was passed to get all transactions
|
||||||
data = response.json()
|
mock_database_service.get_transactions_from_db.assert_called_once()
|
||||||
|
call_args = mock_database_service.get_transactions_from_db.call_args
|
||||||
# Verify that limit=None was passed to get all transactions
|
assert call_args.kwargs.get("limit") is None, "Stats endpoint should pass limit=None 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
|
# Verify that the response contains stats for all 600 transactions
|
||||||
assert call_args.kwargs.get("limit") is None, "Stats endpoint should pass limit=None to get all transactions"
|
assert data["success"] is True
|
||||||
|
stats = data["data"]
|
||||||
# Verify that the response contains stats for all 600 transactions
|
assert stats["total_transactions"] == 600, "Should process all 600 transactions, not just 100"
|
||||||
assert data["success"] is True
|
|
||||||
stats = data["data"]
|
# Verify calculations are correct for all transactions
|
||||||
assert stats["total_transactions"] == 600, "Should process all 600 transactions, not just 100"
|
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)
|
||||||
# Verify calculations are correct for all transactions
|
|
||||||
expected_income = sum(txn["transactionValue"] for txn in mock_transactions if txn["transactionValue"] > 0)
|
assert stats["total_income"] == expected_income
|
||||||
expected_expenses = sum(abs(txn["transactionValue"]) for txn in mock_transactions if txn["transactionValue"] < 0)
|
assert stats["total_expenses"] == expected_expenses
|
||||||
|
|
||||||
assert stats["total_income"] == expected_income
|
|
||||||
assert stats["total_expenses"] == expected_expenses
|
|
||||||
|
|
||||||
finally:
|
|
||||||
# Restore original service
|
|
||||||
transactions.database_service = original_service
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_analytics_endpoint_returns_all_transactions(self, client, mock_database_service):
|
async def test_analytics_endpoint_returns_all_transactions(self, mock_database_service):
|
||||||
"""Test that the new analytics endpoint returns all transactions without pagination"""
|
"""Test that the new analytics endpoint returns all transactions without pagination"""
|
||||||
# Mock data for 600 transactions
|
# Mock data for 600 transactions
|
||||||
mock_transactions = []
|
mock_transactions = []
|
||||||
@@ -91,28 +84,21 @@ class TestAnalyticsFix:
|
|||||||
|
|
||||||
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
mock_database_service.get_transactions_from_db = AsyncMock(return_value=mock_transactions)
|
||||||
|
|
||||||
with client as test_client:
|
with patch('leggend.api.routes.transactions.database_service', mock_database_service):
|
||||||
# Replace the database service in the route handler
|
app = create_app()
|
||||||
from leggend.api.routes import transactions
|
client = TestClient(app)
|
||||||
original_service = transactions.database_service
|
|
||||||
transactions.database_service = mock_database_service
|
response = client.get("/api/v1/transactions/analytics?days=365")
|
||||||
|
|
||||||
try:
|
assert response.status_code == 200
|
||||||
response = test_client.get("/api/v1/transactions/analytics?days=365")
|
data = response.json()
|
||||||
|
|
||||||
assert response.status_code == 200
|
# Verify that limit=None was passed to get all transactions
|
||||||
data = response.json()
|
mock_database_service.get_transactions_from_db.assert_called_once()
|
||||||
|
call_args = mock_database_service.get_transactions_from_db.call_args
|
||||||
# Verify that limit=None was passed to get all transactions
|
assert call_args.kwargs.get("limit") is None, "Analytics endpoint should pass limit=None"
|
||||||
mock_database_service.get_transactions_from_db.assert_called_once()
|
|
||||||
call_args = mock_database_service.get_transactions_from_db.call_args
|
# Verify that all 600 transactions are returned
|
||||||
assert call_args.kwargs.get("limit") is None, "Analytics endpoint should pass limit=None"
|
assert data["success"] is True
|
||||||
|
transactions_data = data["data"]
|
||||||
# Verify that all 600 transactions are returned
|
assert len(transactions_data) == 600, "Analytics endpoint should return all 600 transactions"
|
||||||
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
|
|
||||||
@@ -13,7 +13,6 @@ from leggen.database.sqlite import persist_balances, get_balances
|
|||||||
class MockContext:
|
class MockContext:
|
||||||
"""Mock context for testing."""
|
"""Mock context for testing."""
|
||||||
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
|
|||||||
Reference in New Issue
Block a user