mirror of
https://github.com/elisiariocouto/leggen.git
synced 2025-12-13 13:42:19 +00:00
Compare commits
13 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b3eab6ae26 | ||
|
|
a5d10b3539 | ||
|
|
1c901a9dda | ||
|
|
1e94333d8f | ||
|
|
4006dd128e | ||
|
|
7d9744a40e | ||
|
|
8654471042 | ||
|
|
e9711339bd | ||
|
|
0c030efef2 | ||
|
|
e4e04ea34e | ||
|
|
f4bf549b99 | ||
|
|
8cc4f567f8 | ||
|
|
a939b841f3 |
55
.github/workflows/ci.yml
vendored
Normal file
55
.github/workflows/ci.yml
vendored
Normal file
@@ -0,0 +1,55 @@
|
||||
name: CI
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ "main", "dev" ]
|
||||
pull_request:
|
||||
branches: [ "main", "dev" ]
|
||||
|
||||
jobs:
|
||||
test-python:
|
||||
name: Test Python
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Install uv
|
||||
uses: astral-sh/setup-uv@v5
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version-file: "pyproject.toml"
|
||||
|
||||
- name: Create config directory for tests
|
||||
run: |
|
||||
mkdir -p ~/.config/leggen
|
||||
cp config.example.toml ~/.config/leggen/config.toml
|
||||
|
||||
- name: Run Python tests
|
||||
run: uv run pytest
|
||||
|
||||
test-frontend:
|
||||
name: Test Frontend
|
||||
runs-on: ubuntu-latest
|
||||
defaults:
|
||||
run:
|
||||
working-directory: ./frontend
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Setup Node.js
|
||||
uses: actions/setup-node@v4
|
||||
with:
|
||||
node-version: '20'
|
||||
cache: 'npm'
|
||||
cache-dependency-path: frontend/package-lock.json
|
||||
|
||||
- name: Install dependencies
|
||||
run: npm install
|
||||
|
||||
- name: Run lint
|
||||
run: npm run lint
|
||||
|
||||
- name: Run build
|
||||
run: npm run build
|
||||
31
.github/workflows/release.yml
vendored
31
.github/workflows/release.yml
vendored
@@ -133,3 +133,34 @@ jobs:
|
||||
push: true
|
||||
tags: ${{ steps.meta-frontend.outputs.tags }}
|
||||
labels: ${{ steps.meta-frontend.outputs.labels }}
|
||||
|
||||
create-github-release:
|
||||
name: Create GitHub Release
|
||||
runs-on: ubuntu-latest
|
||||
needs: [build, publish-to-pypi, push-docker-backend, push-docker-frontend]
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
fetch-depth: 0
|
||||
|
||||
- name: Install git-cliff
|
||||
run: |
|
||||
wget -qO- https://github.com/orhun/git-cliff/releases/latest/download/git-cliff-2.10.0-x86_64-unknown-linux-gnu.tar.gz | tar xz
|
||||
sudo mv git-cliff-*/git-cliff /usr/local/bin/
|
||||
|
||||
- name: Generate release notes
|
||||
id: release_notes
|
||||
run: |
|
||||
echo "notes<<EOF" >> $GITHUB_OUTPUT
|
||||
git-cliff --current >> $GITHUB_OUTPUT
|
||||
echo "EOF" >> $GITHUB_OUTPUT
|
||||
|
||||
- name: Create Release
|
||||
uses: softprops/action-gh-release@v2
|
||||
with:
|
||||
tag_name: ${{ github.ref_name }}
|
||||
name: Release ${{ github.ref_name }}
|
||||
body: ${{ steps.release_notes.outputs.notes }}
|
||||
draft: false
|
||||
prerelease: false
|
||||
|
||||
66
CHANGELOG.md
66
CHANGELOG.md
@@ -1,4 +1,70 @@
|
||||
|
||||
## 2025.9.9 (2025/09/11)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- **core:** Handle permission errors gracefully in database path creation. ([4006dd12](https://github.com/elisiariocouto/leggen/commit/4006dd128e0896b338cb93fad60a1eca90c1873d))
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
- **frontend:** Improve transactions table mobile UX with responsive card layout ([1e94333d](https://github.com/elisiariocouto/leggen/commit/1e94333d8f0275542ae7fd6e49fb8b7f03ad3d11))
|
||||
- **frontend:** Improve transactions table mobile UX with responsive card layout ([1c901a9d](https://github.com/elisiariocouto/leggen/commit/1c901a9ddab0f6515dce56df8cce74518805a6bb))
|
||||
- Remove config.toml file - should be created when needed ([a5d10b35](https://github.com/elisiariocouto/leggen/commit/a5d10b3539e7cfc649b0fee05b12c4a03681e135))
|
||||
|
||||
|
||||
### Refactor
|
||||
|
||||
- **core:** Integrate directory creation with database path retrieval and remove backup file. ([7d9744a4](https://github.com/elisiariocouto/leggen/commit/7d9744a40e7898e5bbe52e2e9f54317aa5c1cdd6))
|
||||
|
||||
|
||||
|
||||
## 2025.9.9 (2025/09/11)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- **core:** Handle permission errors gracefully in database path creation. ([4006dd12](https://github.com/elisiariocouto/leggen/commit/4006dd128e0896b338cb93fad60a1eca90c1873d))
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
- **frontend:** Improve transactions table mobile UX with responsive card layout ([1e94333d](https://github.com/elisiariocouto/leggen/commit/1e94333d8f0275542ae7fd6e49fb8b7f03ad3d11))
|
||||
- **frontend:** Improve transactions table mobile UX with responsive card layout ([1c901a9d](https://github.com/elisiariocouto/leggen/commit/1c901a9ddab0f6515dce56df8cce74518805a6bb))
|
||||
- Remove config.toml file - should be created when needed ([a5d10b35](https://github.com/elisiariocouto/leggen/commit/a5d10b3539e7cfc649b0fee05b12c4a03681e135))
|
||||
|
||||
|
||||
### Refactor
|
||||
|
||||
- **core:** Integrate directory creation with database path retrieval and remove backup file. ([7d9744a4](https://github.com/elisiariocouto/leggen/commit/7d9744a40e7898e5bbe52e2e9f54317aa5c1cdd6))
|
||||
|
||||
|
||||
|
||||
## 2025.9.8 (2025/09/11)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Change branch name from develop to dev in CI workflow ([f4bf549b](https://github.com/elisiariocouto/leggen/commit/f4bf549b99197d70104abf5731ab1ccb67cc9a69))
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
- Update CI workflow to use Node.js 20 instead of 18 ([e4e04ea3](https://github.com/elisiariocouto/leggen/commit/e4e04ea34ea568c08292562243b6e6c08234d918))
|
||||
|
||||
|
||||
|
||||
## 2025.9.8 (2025/09/11)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Change branch name from develop to dev in CI workflow ([f4bf549b](https://github.com/elisiariocouto/leggen/commit/f4bf549b99197d70104abf5731ab1ccb67cc9a69))
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
- Update CI workflow to use Node.js 20 instead of 18 ([e4e04ea3](https://github.com/elisiariocouto/leggen/commit/e4e04ea34ea568c08292562243b6e6c08234d918))
|
||||
|
||||
|
||||
|
||||
## 2025.9.7 (2025/09/11)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
@@ -357,6 +357,10 @@ tests/ # Test suite
|
||||
3. Make your changes with tests
|
||||
4. Submit a pull request
|
||||
|
||||
The repository uses GitHub Actions for CI/CD:
|
||||
- **CI**: Runs Python tests (`uv run pytest`) and frontend linting/build on every push
|
||||
- **Release**: Creates GitHub releases with changelog when tags are pushed
|
||||
|
||||
## ⚠️ Notes
|
||||
- This project is in active development
|
||||
- GoCardless API rate limits apply
|
||||
|
||||
@@ -412,9 +412,9 @@ export default function TransactionsTable() {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="grid grid-cols-1 md:grid-cols-4 gap-4">
|
||||
<div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
|
||||
{/* Search */}
|
||||
<div>
|
||||
<div className="sm:col-span-2 lg:col-span-1">
|
||||
<label className="block text-sm font-medium text-gray-700 mb-1">
|
||||
Search
|
||||
</label>
|
||||
@@ -448,8 +448,7 @@ export default function TransactionsTable() {
|
||||
<option value="">All accounts</option>
|
||||
{accounts?.map((account) => (
|
||||
<option key={account.id} value={account.id}>
|
||||
{account.name || "Unnamed Account"} (
|
||||
{account.institution_id})
|
||||
{account.name || "Unnamed Account"} ({account.institution_id})
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
@@ -489,7 +488,7 @@ export default function TransactionsTable() {
|
||||
</div>
|
||||
|
||||
{/* Amount Range Filters */}
|
||||
<div className="grid grid-cols-1 md:grid-cols-2 gap-4 mt-4">
|
||||
<div className="grid grid-cols-1 sm:grid-cols-2 gap-4 mt-4">
|
||||
<div>
|
||||
<label className="block text-sm font-medium text-gray-700 mb-1">
|
||||
Min Amount
|
||||
@@ -547,94 +546,206 @@ export default function TransactionsTable() {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Table */}
|
||||
{/* Responsive Table/Cards */}
|
||||
<div className="bg-white rounded-lg shadow overflow-hidden">
|
||||
<div className="overflow-x-auto">
|
||||
<table className="min-w-full divide-y divide-gray-200">
|
||||
<thead className="bg-gray-50">
|
||||
{table.getHeaderGroups().map((headerGroup) => (
|
||||
<tr key={headerGroup.id}>
|
||||
{headerGroup.headers.map((header) => (
|
||||
<th
|
||||
key={header.id}
|
||||
className="px-6 py-3 text-left text-xs font-medium text-gray-500 uppercase tracking-wider cursor-pointer hover:bg-gray-100"
|
||||
onClick={header.column.getToggleSortingHandler()}
|
||||
>
|
||||
<div className="flex items-center space-x-1">
|
||||
<span>
|
||||
{header.isPlaceholder
|
||||
? null
|
||||
: flexRender(
|
||||
header.column.columnDef.header,
|
||||
header.getContext(),
|
||||
)}
|
||||
</span>
|
||||
{header.column.getCanSort() && (
|
||||
<div className="flex flex-col">
|
||||
<ChevronUp
|
||||
className={`h-3 w-3 ${
|
||||
header.column.getIsSorted() === "asc"
|
||||
? "text-blue-600"
|
||||
: "text-gray-400"
|
||||
}`}
|
||||
/>
|
||||
<ChevronDown
|
||||
className={`h-3 w-3 -mt-1 ${
|
||||
header.column.getIsSorted() === "desc"
|
||||
? "text-blue-600"
|
||||
: "text-gray-400"
|
||||
}`}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</th>
|
||||
))}
|
||||
</tr>
|
||||
))}
|
||||
</thead>
|
||||
<tbody className="bg-white divide-y divide-gray-200">
|
||||
{table.getRowModel().rows.length === 0 ? (
|
||||
<tr>
|
||||
<td
|
||||
colSpan={columns.length}
|
||||
className="px-6 py-12 text-center"
|
||||
>
|
||||
<div className="text-gray-400 mb-4">
|
||||
<TrendingUp className="h-12 w-12 mx-auto" />
|
||||
</div>
|
||||
<h3 className="text-lg font-medium text-gray-900 mb-2">
|
||||
No transactions found
|
||||
</h3>
|
||||
<p className="text-gray-600">
|
||||
{hasActiveFilters
|
||||
? "Try adjusting your filters to see more results."
|
||||
: "No transactions are available for the selected criteria."}
|
||||
</p>
|
||||
</td>
|
||||
</tr>
|
||||
) : (
|
||||
table.getRowModel().rows.map((row) => (
|
||||
<tr key={row.id} className="hover:bg-gray-50">
|
||||
{row.getVisibleCells().map((cell) => (
|
||||
<td key={cell.id} className="px-6 py-4 whitespace-nowrap">
|
||||
{flexRender(
|
||||
cell.column.columnDef.cell,
|
||||
cell.getContext(),
|
||||
)}
|
||||
</td>
|
||||
{/* Desktop Table View (hidden on mobile) */}
|
||||
<div className="hidden md:block">
|
||||
<div className="overflow-x-auto">
|
||||
<table className="min-w-full divide-y divide-gray-200">
|
||||
<thead className="bg-gray-50">
|
||||
{table.getHeaderGroups().map((headerGroup) => (
|
||||
<tr key={headerGroup.id}>
|
||||
{headerGroup.headers.map((header) => (
|
||||
<th
|
||||
key={header.id}
|
||||
className="px-6 py-3 text-left text-xs font-medium text-gray-500 uppercase tracking-wider cursor-pointer hover:bg-gray-100"
|
||||
onClick={header.column.getToggleSortingHandler()}
|
||||
>
|
||||
<div className="flex items-center space-x-1">
|
||||
<span>
|
||||
{header.isPlaceholder
|
||||
? null
|
||||
: flexRender(
|
||||
header.column.columnDef.header,
|
||||
header.getContext(),
|
||||
)}
|
||||
</span>
|
||||
{header.column.getCanSort() && (
|
||||
<div className="flex flex-col">
|
||||
<ChevronUp
|
||||
className={`h-3 w-3 ${
|
||||
header.column.getIsSorted() === "asc"
|
||||
? "text-blue-600"
|
||||
: "text-gray-400"
|
||||
}`}
|
||||
/>
|
||||
<ChevronDown
|
||||
className={`h-3 w-3 -mt-1 ${
|
||||
header.column.getIsSorted() === "desc"
|
||||
? "text-blue-600"
|
||||
: "text-gray-400"
|
||||
}`}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</th>
|
||||
))}
|
||||
</tr>
|
||||
))
|
||||
)}
|
||||
</tbody>
|
||||
</table>
|
||||
))}
|
||||
</thead>
|
||||
<tbody className="bg-white divide-y divide-gray-200">
|
||||
{table.getRowModel().rows.length === 0 ? (
|
||||
<tr>
|
||||
<td
|
||||
colSpan={columns.length}
|
||||
className="px-6 py-12 text-center"
|
||||
>
|
||||
<div className="text-gray-400 mb-4">
|
||||
<TrendingUp className="h-12 w-12 mx-auto" />
|
||||
</div>
|
||||
<h3 className="text-lg font-medium text-gray-900 mb-2">
|
||||
No transactions found
|
||||
</h3>
|
||||
<p className="text-gray-600">
|
||||
{hasActiveFilters
|
||||
? "Try adjusting your filters to see more results."
|
||||
: "No transactions are available for the selected criteria."}
|
||||
</p>
|
||||
</td>
|
||||
</tr>
|
||||
) : (
|
||||
table.getRowModel().rows.map((row) => (
|
||||
<tr key={row.id} className="hover:bg-gray-50">
|
||||
{row.getVisibleCells().map((cell) => (
|
||||
<td key={cell.id} className="px-6 py-4 whitespace-nowrap">
|
||||
{flexRender(
|
||||
cell.column.columnDef.cell,
|
||||
cell.getContext(),
|
||||
)}
|
||||
</td>
|
||||
))}
|
||||
</tr>
|
||||
))
|
||||
)}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Mobile Card View (visible only on mobile) */}
|
||||
<div className="md:hidden">
|
||||
{table.getRowModel().rows.length === 0 ? (
|
||||
<div className="px-6 py-12 text-center">
|
||||
<div className="text-gray-400 mb-4">
|
||||
<TrendingUp className="h-12 w-12 mx-auto" />
|
||||
</div>
|
||||
<h3 className="text-lg font-medium text-gray-900 mb-2">
|
||||
No transactions found
|
||||
</h3>
|
||||
<p className="text-gray-600">
|
||||
{hasActiveFilters
|
||||
? "Try adjusting your filters to see more results."
|
||||
: "No transactions are available for the selected criteria."}
|
||||
</p>
|
||||
</div>
|
||||
) : (
|
||||
<div className="divide-y divide-gray-200">
|
||||
{table.getRowModel().rows.map((row) => {
|
||||
const transaction = row.original;
|
||||
const account = accounts?.find(
|
||||
(acc) => acc.id === transaction.account_id,
|
||||
);
|
||||
const isPositive = transaction.transaction_value > 0;
|
||||
|
||||
return (
|
||||
<div
|
||||
key={row.id}
|
||||
className="p-4 hover:bg-gray-50 transition-colors"
|
||||
>
|
||||
<div className="flex items-start justify-between">
|
||||
<div className="flex-1 min-w-0">
|
||||
<div className="flex items-start space-x-3">
|
||||
<div
|
||||
className={`p-2 rounded-full flex-shrink-0 ${
|
||||
isPositive ? "bg-green-100" : "bg-red-100"
|
||||
}`}
|
||||
>
|
||||
{isPositive ? (
|
||||
<TrendingUp className="h-4 w-4 text-green-600" />
|
||||
) : (
|
||||
<TrendingDown className="h-4 w-4 text-red-600" />
|
||||
)}
|
||||
</div>
|
||||
<div className="flex-1 min-w-0">
|
||||
<h4 className="text-sm font-medium text-gray-900 break-words">
|
||||
{transaction.description}
|
||||
</h4>
|
||||
<div className="text-xs text-gray-500 space-y-1 mt-1">
|
||||
{account && (
|
||||
<p className="break-words">
|
||||
{account.name || "Unnamed Account"} •{" "}
|
||||
{account.institution_id}
|
||||
</p>
|
||||
)}
|
||||
{(transaction.creditor_name || transaction.debtor_name) && (
|
||||
<p className="break-words">
|
||||
{isPositive ? "From: " : "To: "}
|
||||
{transaction.creditor_name || transaction.debtor_name}
|
||||
</p>
|
||||
)}
|
||||
{transaction.reference && (
|
||||
<p className="break-words">Ref: {transaction.reference}</p>
|
||||
)}
|
||||
<p className="text-gray-400">
|
||||
{transaction.transaction_date
|
||||
? formatDate(transaction.transaction_date)
|
||||
: "No date"}
|
||||
{transaction.booking_date &&
|
||||
transaction.booking_date !== transaction.transaction_date && (
|
||||
<span className="ml-2">
|
||||
(Booked: {formatDate(transaction.booking_date)})
|
||||
</span>
|
||||
)}
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
<div className="text-right ml-3 flex-shrink-0">
|
||||
<p
|
||||
className={`text-lg font-semibold mb-1 ${
|
||||
isPositive ? "text-green-600" : "text-red-600"
|
||||
}`}
|
||||
>
|
||||
{isPositive ? "+" : ""}
|
||||
{formatCurrency(
|
||||
transaction.transaction_value,
|
||||
transaction.transaction_currency,
|
||||
)}
|
||||
</p>
|
||||
<button
|
||||
onClick={() => handleViewRaw(transaction)}
|
||||
className="inline-flex items-center px-2 py-1 text-xs bg-gray-100 text-gray-700 rounded hover:bg-gray-200 transition-colors"
|
||||
title="View raw transaction data"
|
||||
>
|
||||
<Eye className="h-3 w-3 mr-1" />
|
||||
Raw
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
|
||||
{/* Pagination */}
|
||||
{pagination && (
|
||||
<div className="bg-white px-4 py-3 flex items-center justify-between border-t border-gray-200 sm:px-6">
|
||||
<div className="flex-1 flex justify-between sm:hidden">
|
||||
<div className="bg-white px-4 py-3 flex flex-col sm:flex-row items-center justify-between border-t border-gray-200 space-y-3 sm:space-y-0">
|
||||
{/* Mobile pagination controls */}
|
||||
<div className="flex justify-between w-full sm:hidden">
|
||||
<div className="flex space-x-2">
|
||||
<button
|
||||
onClick={() => setCurrentPage(1)}
|
||||
@@ -670,7 +781,22 @@ export default function TransactionsTable() {
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
<div className="hidden sm:flex-1 sm:flex sm:items-center sm:justify-between">
|
||||
|
||||
{/* Mobile pagination info */}
|
||||
<div className="text-center w-full sm:hidden">
|
||||
<p className="text-sm text-gray-700">
|
||||
Page <span className="font-medium">{pagination.page}</span> of{" "}
|
||||
<span className="font-medium">{pagination.total_pages}</span>
|
||||
<br />
|
||||
<span className="text-xs text-gray-500">
|
||||
Showing {(pagination.page - 1) * pagination.per_page + 1}-
|
||||
{Math.min(pagination.page * pagination.per_page, pagination.total)} of {pagination.total}
|
||||
</span>
|
||||
</p>
|
||||
</div>
|
||||
|
||||
{/* Desktop pagination */}
|
||||
<div className="hidden sm:flex sm:flex-1 sm:items-center sm:justify-between">
|
||||
<div className="flex items-center space-x-2">
|
||||
<p className="text-sm text-gray-700">
|
||||
Showing{" "}
|
||||
|
||||
65
leggen/commands/generate_sample_db.py
Normal file
65
leggen/commands/generate_sample_db.py
Normal file
@@ -0,0 +1,65 @@
|
||||
"""Generate sample database command."""
|
||||
|
||||
import click
|
||||
from pathlib import Path
|
||||
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--database",
|
||||
type=click.Path(path_type=Path),
|
||||
help="Path to database file (default: uses LEGGEN_DATABASE_PATH or ~/.config/leggen/leggen-dev.db)",
|
||||
)
|
||||
@click.option(
|
||||
"--accounts",
|
||||
type=int,
|
||||
default=3,
|
||||
help="Number of sample accounts to generate (default: 3)",
|
||||
)
|
||||
@click.option(
|
||||
"--transactions",
|
||||
type=int,
|
||||
default=50,
|
||||
help="Number of transactions per account (default: 50)",
|
||||
)
|
||||
@click.option(
|
||||
"--force",
|
||||
is_flag=True,
|
||||
help="Overwrite existing database without confirmation",
|
||||
)
|
||||
@click.pass_context
|
||||
def generate_sample_db(ctx: click.Context, database: Path, accounts: int, transactions: int, force: bool):
|
||||
"""Generate a sample database with realistic financial data for testing."""
|
||||
|
||||
# Import here to avoid circular imports
|
||||
import sys
|
||||
import subprocess
|
||||
from pathlib import Path as PathlibPath
|
||||
|
||||
# Get the script path
|
||||
script_path = PathlibPath(__file__).parent.parent.parent / "scripts" / "generate_sample_db.py"
|
||||
|
||||
# Build command arguments
|
||||
cmd = [sys.executable, str(script_path)]
|
||||
|
||||
if database:
|
||||
cmd.extend(["--database", str(database)])
|
||||
|
||||
cmd.extend(["--accounts", str(accounts)])
|
||||
cmd.extend(["--transactions", str(transactions)])
|
||||
|
||||
if force:
|
||||
cmd.append("--force")
|
||||
|
||||
# Execute the script
|
||||
try:
|
||||
subprocess.run(cmd, check=True)
|
||||
except subprocess.CalledProcessError as e:
|
||||
click.echo(f"Error generating sample database: {e}")
|
||||
ctx.exit(1)
|
||||
|
||||
|
||||
# Export the command
|
||||
generate_sample_db = generate_sample_db
|
||||
@@ -5,14 +5,12 @@ from sqlite3 import IntegrityError
|
||||
import click
|
||||
|
||||
from leggen.utils.text import success, warning
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
def persist_balances(ctx: click.Context, balance: dict):
|
||||
# Connect to SQLite database
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
db_path = path_manager.get_database_path()
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -108,10 +106,7 @@ def persist_balances(ctx: click.Context, balance: dict):
|
||||
|
||||
def persist_transactions(ctx: click.Context, account: str, transactions: list) -> list:
|
||||
# Connect to SQLite database
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
db_path = path_manager.get_database_path()
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -216,9 +211,7 @@ def get_transactions(
|
||||
search=None,
|
||||
):
|
||||
"""Get transactions from SQLite database with optional filtering"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return []
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
@@ -288,9 +281,7 @@ def get_transactions(
|
||||
|
||||
def get_balances(account_id=None):
|
||||
"""Get latest balances from SQLite database"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return []
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
@@ -329,9 +320,7 @@ def get_balances(account_id=None):
|
||||
|
||||
def get_account_summary(account_id):
|
||||
"""Get basic account info from transactions table (avoids GoCardless API call)"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return None
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
@@ -365,9 +354,7 @@ def get_account_summary(account_id):
|
||||
|
||||
def get_transaction_count(account_id=None, **filters):
|
||||
"""Get total count of transactions matching filters"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return 0
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
@@ -414,10 +401,7 @@ def get_transaction_count(account_id=None, **filters):
|
||||
|
||||
def persist_account(account_data: dict):
|
||||
"""Persist account details to SQLite database"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
db_path = path_manager.get_database_path()
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -485,9 +469,7 @@ def persist_account(account_data: dict):
|
||||
|
||||
def get_accounts(account_ids=None):
|
||||
"""Get account details from SQLite database"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return []
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
@@ -519,9 +501,7 @@ def get_accounts(account_ids=None):
|
||||
|
||||
def get_account(account_id: str):
|
||||
"""Get specific account details from SQLite database"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return None
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
|
||||
@@ -7,6 +7,7 @@ import click
|
||||
|
||||
from leggen.utils.config import load_config
|
||||
from leggen.utils.text import error
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
cmd_folder = os.path.abspath(os.path.join(os.path.dirname(__file__), "commands"))
|
||||
|
||||
@@ -77,7 +78,7 @@ class Group(click.Group):
|
||||
"-c",
|
||||
"--config",
|
||||
type=click.Path(dir_okay=False),
|
||||
default=Path.home() / ".config" / "leggen" / "config.toml",
|
||||
default=lambda: str(path_manager.get_config_file_path()),
|
||||
show_default=True,
|
||||
callback=load_config,
|
||||
is_eager=True,
|
||||
@@ -86,6 +87,20 @@ class Group(click.Group):
|
||||
show_envvar=True,
|
||||
help="Path to TOML configuration file",
|
||||
)
|
||||
@click.option(
|
||||
"--config-dir",
|
||||
type=click.Path(exists=False, file_okay=False, path_type=Path),
|
||||
envvar="LEGGEN_CONFIG_DIR",
|
||||
show_envvar=True,
|
||||
help="Directory containing configuration files (default: ~/.config/leggen)",
|
||||
)
|
||||
@click.option(
|
||||
"--database",
|
||||
type=click.Path(dir_okay=False, path_type=Path),
|
||||
envvar="LEGGEN_DATABASE_PATH",
|
||||
show_envvar=True,
|
||||
help="Path to SQLite database file (default: <config-dir>/leggen.db)",
|
||||
)
|
||||
@click.option(
|
||||
"--api-url",
|
||||
type=str,
|
||||
@@ -100,7 +115,7 @@ class Group(click.Group):
|
||||
)
|
||||
@click.version_option(package_name="leggen")
|
||||
@click.pass_context
|
||||
def cli(ctx: click.Context, api_url: str):
|
||||
def cli(ctx: click.Context, config_dir: Path, database: Path, api_url: str):
|
||||
"""
|
||||
Leggen: An Open Banking CLI
|
||||
"""
|
||||
@@ -109,5 +124,11 @@ def cli(ctx: click.Context, api_url: str):
|
||||
if "--help" in sys.argv[1:] or "-h" in sys.argv[1:]:
|
||||
return
|
||||
|
||||
# Set up path manager with user-provided paths
|
||||
if config_dir:
|
||||
path_manager.set_config_dir(config_dir)
|
||||
if database:
|
||||
path_manager.set_database_path(database)
|
||||
|
||||
# Store API URL in context for commands to use
|
||||
ctx.obj["api_url"] = api_url
|
||||
|
||||
81
leggen/utils/paths.py
Normal file
81
leggen/utils/paths.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""Centralized path management for Leggen."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class PathManager:
|
||||
"""Manages configurable paths for config and database files."""
|
||||
|
||||
def __init__(self):
|
||||
self._config_dir: Optional[Path] = None
|
||||
self._database_path: Optional[Path] = None
|
||||
|
||||
def get_config_dir(self) -> Path:
|
||||
"""Get the configuration directory."""
|
||||
if self._config_dir is not None:
|
||||
return self._config_dir
|
||||
|
||||
# Check environment variable first
|
||||
config_dir = os.environ.get("LEGGEN_CONFIG_DIR")
|
||||
if config_dir:
|
||||
return Path(config_dir)
|
||||
|
||||
# Default to ~/.config/leggen
|
||||
return Path.home() / ".config" / "leggen"
|
||||
|
||||
def set_config_dir(self, path: Path) -> None:
|
||||
"""Set the configuration directory."""
|
||||
self._config_dir = Path(path)
|
||||
|
||||
def get_config_file_path(self) -> Path:
|
||||
"""Get the configuration file path."""
|
||||
return self.get_config_dir() / "config.toml"
|
||||
|
||||
def get_database_path(self) -> Path:
|
||||
"""Get the database file path and ensure the directory exists."""
|
||||
if self._database_path is not None:
|
||||
db_path = self._database_path
|
||||
else:
|
||||
# Check environment variable first
|
||||
database_path = os.environ.get("LEGGEN_DATABASE_PATH")
|
||||
if database_path:
|
||||
db_path = Path(database_path)
|
||||
else:
|
||||
# Default to config_dir/leggen.db
|
||||
db_path = self.get_config_dir() / "leggen.db"
|
||||
|
||||
# Try to ensure the directory exists, but handle permission errors gracefully
|
||||
try:
|
||||
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
|
||||
|
||||
def set_database_path(self, path: Path) -> None:
|
||||
"""Set the database file path."""
|
||||
self._database_path = Path(path)
|
||||
|
||||
def get_auth_file_path(self) -> Path:
|
||||
"""Get the authentication file path."""
|
||||
return self.get_config_dir() / "auth.json"
|
||||
|
||||
def ensure_config_dir_exists(self) -> None:
|
||||
"""Ensure the configuration directory exists."""
|
||||
self.get_config_dir().mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def ensure_database_dir_exists(self) -> None:
|
||||
"""Ensure the database directory exists.
|
||||
|
||||
Note: get_database_path() now automatically ensures the directory exists,
|
||||
so this method is mainly for explicit directory creation in tests.
|
||||
"""
|
||||
self.get_database_path().parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
# Global instance for the application
|
||||
path_manager = PathManager()
|
||||
@@ -5,6 +5,7 @@ from pathlib import Path
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
from loguru import logger
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
class Config:
|
||||
@@ -23,9 +24,10 @@ class Config:
|
||||
|
||||
if config_path is None:
|
||||
config_path = os.environ.get(
|
||||
"LEGGEN_CONFIG_FILE",
|
||||
str(Path.home() / ".config" / "leggen" / "config.toml"),
|
||||
"LEGGEN_CONFIG_FILE"
|
||||
)
|
||||
if not config_path:
|
||||
config_path = str(path_manager.get_config_file_path())
|
||||
|
||||
self._config_path = config_path
|
||||
|
||||
@@ -53,9 +55,10 @@ class Config:
|
||||
|
||||
if config_path is None:
|
||||
config_path = self._config_path or os.environ.get(
|
||||
"LEGGEN_CONFIG_FILE",
|
||||
str(Path.home() / ".config" / "leggen" / "config.toml"),
|
||||
"LEGGEN_CONFIG_FILE"
|
||||
)
|
||||
if not config_path:
|
||||
config_path = str(path_manager.get_config_file_path())
|
||||
|
||||
if config_path is None:
|
||||
raise ValueError("No config path specified")
|
||||
|
||||
@@ -121,6 +121,8 @@ def create_app() -> FastAPI:
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
parser = argparse.ArgumentParser(description="Start the Leggend API service")
|
||||
parser.add_argument(
|
||||
@@ -132,8 +134,24 @@ def main():
|
||||
parser.add_argument(
|
||||
"--port", type=int, default=8000, help="Port to bind to (default: 8000)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--config-dir",
|
||||
type=Path,
|
||||
help="Directory containing configuration files (default: ~/.config/leggen)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--database",
|
||||
type=Path,
|
||||
help="Path to SQLite database file (default: <config-dir>/leggen.db)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
# Set up path manager with user-provided paths
|
||||
if args.config_dir:
|
||||
path_manager.set_config_dir(args.config_dir)
|
||||
if args.database:
|
||||
path_manager.set_database_path(args.database)
|
||||
|
||||
if args.reload:
|
||||
# Use string import for reload to work properly
|
||||
uvicorn.run(
|
||||
|
||||
@@ -6,6 +6,7 @@ from loguru import logger
|
||||
|
||||
from leggend.config import config
|
||||
import leggen.database.sqlite as sqlite_db
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
class DatabaseService:
|
||||
@@ -280,9 +281,7 @@ class DatabaseService:
|
||||
|
||||
async def _check_balance_timestamp_migration_needed(self) -> bool:
|
||||
"""Check if balance timestamps need migration"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return False
|
||||
|
||||
@@ -310,9 +309,7 @@ class DatabaseService:
|
||||
|
||||
async def _migrate_balance_timestamps(self):
|
||||
"""Convert all Unix timestamps to datetime strings"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
logger.warning("Database file not found, skipping migration")
|
||||
return
|
||||
@@ -399,9 +396,7 @@ class DatabaseService:
|
||||
|
||||
async def _check_null_transaction_ids_migration_needed(self) -> bool:
|
||||
"""Check if null transaction IDs need migration"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return False
|
||||
|
||||
@@ -429,9 +424,8 @@ class DatabaseService:
|
||||
async def _migrate_null_transaction_ids(self):
|
||||
"""Populate null internalTransactionId fields using transactionId from raw data"""
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
logger.warning("Database file not found, skipping migration")
|
||||
return
|
||||
@@ -538,9 +532,7 @@ class DatabaseService:
|
||||
|
||||
async def _check_composite_key_migration_needed(self) -> bool:
|
||||
"""Check if composite key migration is needed"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
return False
|
||||
|
||||
@@ -586,9 +578,7 @@ class DatabaseService:
|
||||
|
||||
async def _migrate_to_composite_key(self):
|
||||
"""Migrate transactions table to use composite primary key (accountId, transactionId)"""
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path = path_manager.get_database_path()
|
||||
if not db_path.exists():
|
||||
logger.warning("Database file not found, skipping migration")
|
||||
return
|
||||
@@ -704,10 +694,7 @@ class DatabaseService:
|
||||
try:
|
||||
import sqlite3
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
db_path = path_manager.get_database_path()
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -786,10 +773,7 @@ class DatabaseService:
|
||||
import sqlite3
|
||||
import json
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
db_path = path_manager.get_database_path()
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -888,11 +872,6 @@ class DatabaseService:
|
||||
) -> None:
|
||||
"""Persist account details to SQLite"""
|
||||
try:
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path.home() / ".config" / "leggen" / "leggen.db"
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Use the sqlite_db module function
|
||||
sqlite_db.persist_account(account_data)
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ from typing import Dict, Any, List
|
||||
from loguru import logger
|
||||
|
||||
from leggend.config import config
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
def _log_rate_limits(response):
|
||||
@@ -39,8 +40,8 @@ class GoCardlessService:
|
||||
if self._token:
|
||||
return self._token
|
||||
|
||||
# Use ~/.config/leggen for consistency with main config
|
||||
auth_file = Path.home() / ".config" / "leggen" / "auth.json"
|
||||
# Use path manager for auth file
|
||||
auth_file = path_manager.get_auth_file_path()
|
||||
|
||||
if auth_file.exists():
|
||||
try:
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "leggen"
|
||||
version = "2025.9.7"
|
||||
version = "2025.9.9"
|
||||
description = "An Open Banking CLI"
|
||||
authors = [{ name = "Elisiário Couto", email = "elisiario@couto.io" }]
|
||||
requires-python = "~=3.13.0"
|
||||
|
||||
426
scripts/generate_sample_db.py
Executable file
426
scripts/generate_sample_db.py
Executable file
@@ -0,0 +1,426 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Sample database generator for Leggen testing and development."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import random
|
||||
import sqlite3
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any
|
||||
|
||||
# Add the project root to the Python path
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
import click
|
||||
from leggen.utils.paths import path_manager
|
||||
|
||||
|
||||
class SampleDataGenerator:
|
||||
"""Generates realistic sample data for testing Leggen."""
|
||||
|
||||
def __init__(self, db_path: Path):
|
||||
self.db_path = db_path
|
||||
self.institutions = [
|
||||
{
|
||||
"id": "REVOLUT_REVOLT21",
|
||||
"name": "Revolut",
|
||||
"bic": "REVOLT21",
|
||||
"country": "LT",
|
||||
},
|
||||
{
|
||||
"id": "BANCOBPI_BBPIPTPL",
|
||||
"name": "Banco BPI",
|
||||
"bic": "BBPIPTPL",
|
||||
"country": "PT",
|
||||
},
|
||||
{
|
||||
"id": "MONZO_MONZGB2L",
|
||||
"name": "Monzo Bank",
|
||||
"bic": "MONZGB2L",
|
||||
"country": "GB",
|
||||
},
|
||||
{
|
||||
"id": "NUBANK_NUPBBR25",
|
||||
"name": "Nu Pagamentos",
|
||||
"bic": "NUPBBR25",
|
||||
"country": "BR",
|
||||
},
|
||||
]
|
||||
|
||||
self.transaction_types = [
|
||||
{"description": "Grocery Store", "amount_range": (-150, -20), "frequency": 0.3},
|
||||
{"description": "Coffee Shop", "amount_range": (-15, -3), "frequency": 0.2},
|
||||
{"description": "Gas Station", "amount_range": (-80, -30), "frequency": 0.1},
|
||||
{"description": "Online Shopping", "amount_range": (-200, -25), "frequency": 0.15},
|
||||
{"description": "Restaurant", "amount_range": (-60, -15), "frequency": 0.15},
|
||||
{"description": "Salary", "amount_range": (2500, 5000), "frequency": 0.02},
|
||||
{"description": "ATM Withdrawal", "amount_range": (-200, -20), "frequency": 0.05},
|
||||
{"description": "Transfer to Savings", "amount_range": (-1000, -100), "frequency": 0.03},
|
||||
]
|
||||
|
||||
def ensure_database_dir(self):
|
||||
"""Ensure database directory exists."""
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def create_tables(self):
|
||||
"""Create database tables."""
|
||||
conn = sqlite3.connect(str(self.db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Create accounts table
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS accounts (
|
||||
id TEXT PRIMARY KEY,
|
||||
institution_id TEXT,
|
||||
status TEXT,
|
||||
iban TEXT,
|
||||
name TEXT,
|
||||
currency TEXT,
|
||||
created DATETIME,
|
||||
last_accessed DATETIME,
|
||||
last_updated DATETIME
|
||||
)
|
||||
""")
|
||||
|
||||
# Create transactions table with composite primary key
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS transactions (
|
||||
accountId TEXT NOT NULL,
|
||||
transactionId TEXT NOT NULL,
|
||||
internalTransactionId TEXT,
|
||||
institutionId TEXT,
|
||||
iban TEXT,
|
||||
transactionDate DATETIME,
|
||||
description TEXT,
|
||||
transactionValue REAL,
|
||||
transactionCurrency TEXT,
|
||||
transactionStatus TEXT,
|
||||
rawTransaction JSON,
|
||||
PRIMARY KEY (accountId, transactionId)
|
||||
)
|
||||
""")
|
||||
|
||||
# Create balances table
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS balances (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
account_id TEXT,
|
||||
bank TEXT,
|
||||
status TEXT,
|
||||
iban TEXT,
|
||||
amount REAL,
|
||||
currency TEXT,
|
||||
type TEXT,
|
||||
timestamp DATETIME
|
||||
)
|
||||
""")
|
||||
|
||||
# Create indexes
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_transactions_internal_id ON transactions(internalTransactionId)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_transactions_date ON transactions(transactionDate)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_transactions_account_date ON transactions(accountId, transactionDate)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_transactions_amount ON transactions(transactionValue)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_balances_account_id ON balances(account_id)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_balances_timestamp ON balances(timestamp)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_balances_account_type_timestamp ON balances(account_id, type, timestamp)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_accounts_institution_id ON accounts(institution_id)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_accounts_status ON accounts(status)")
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def generate_iban(self, country_code: str) -> str:
|
||||
"""Generate a realistic IBAN for the given country."""
|
||||
ibans = {
|
||||
"LT": lambda: f"LT{random.randint(10, 99)}{random.randint(10000, 99999)}{random.randint(10000000, 99999999)}",
|
||||
"PT": lambda: f"PT{random.randint(10, 99)}{random.randint(1000, 9999)}{random.randint(1000, 9999)}{random.randint(10000000000, 99999999999)}",
|
||||
"GB": lambda: f"GB{random.randint(10, 99)}MONZ{random.randint(100000, 999999)}{random.randint(100000, 999999)}",
|
||||
"BR": lambda: f"BR{random.randint(10, 99)}{random.randint(10000000, 99999999)}{random.randint(1000, 9999)}{random.randint(10000000, 99999999)}",
|
||||
}
|
||||
return ibans.get(country_code, lambda: f"{country_code}{random.randint(1000000000000000, 9999999999999999)}")()
|
||||
|
||||
def generate_accounts(self, num_accounts: int = 3) -> List[Dict[str, Any]]:
|
||||
"""Generate sample accounts."""
|
||||
accounts = []
|
||||
base_date = datetime.now() - timedelta(days=90)
|
||||
|
||||
for i in range(num_accounts):
|
||||
institution = random.choice(self.institutions)
|
||||
account_id = f"account-{i+1:03d}-{random.randint(1000, 9999)}"
|
||||
|
||||
account = {
|
||||
"id": account_id,
|
||||
"institution_id": institution["id"],
|
||||
"status": "READY",
|
||||
"iban": self.generate_iban(institution["country"]),
|
||||
"name": f"Personal Account {i+1}",
|
||||
"currency": "EUR",
|
||||
"created": (base_date + timedelta(days=random.randint(0, 30))).isoformat(),
|
||||
"last_accessed": (datetime.now() - timedelta(hours=random.randint(1, 48))).isoformat(),
|
||||
"last_updated": datetime.now().isoformat(),
|
||||
}
|
||||
accounts.append(account)
|
||||
|
||||
return accounts
|
||||
|
||||
def generate_transactions(self, accounts: List[Dict[str, Any]], num_transactions_per_account: int = 50) -> List[Dict[str, Any]]:
|
||||
"""Generate sample transactions for accounts."""
|
||||
transactions = []
|
||||
base_date = datetime.now() - timedelta(days=60)
|
||||
|
||||
for account in accounts:
|
||||
account_transactions = []
|
||||
current_balance = random.uniform(500, 3000)
|
||||
|
||||
for i in range(num_transactions_per_account):
|
||||
# Choose transaction type based on frequency weights
|
||||
transaction_type = random.choices(
|
||||
self.transaction_types,
|
||||
weights=[t["frequency"] for t in self.transaction_types]
|
||||
)[0]
|
||||
|
||||
# Generate transaction amount
|
||||
min_amount, max_amount = transaction_type["amount_range"]
|
||||
amount = round(random.uniform(min_amount, max_amount), 2)
|
||||
|
||||
# Generate transaction date (more recent transactions are more likely)
|
||||
days_ago = random.choices(
|
||||
range(60),
|
||||
weights=[1.5 ** (60 - d) for d in range(60)]
|
||||
)[0]
|
||||
transaction_date = base_date + timedelta(days=days_ago, hours=random.randint(6, 22), minutes=random.randint(0, 59))
|
||||
|
||||
# Generate transaction IDs
|
||||
transaction_id = f"bank-txn-{account['id']}-{i+1:04d}"
|
||||
internal_transaction_id = f"int-txn-{random.randint(100000, 999999)}"
|
||||
|
||||
# Create realistic descriptions
|
||||
descriptions = {
|
||||
"Grocery Store": ["TESCO", "SAINSBURY'S", "LIDL", "ALDI", "WALMART", "CARREFOUR"],
|
||||
"Coffee Shop": ["STARBUCKS", "COSTA COFFEE", "PRET A MANGER", "LOCAL CAFE"],
|
||||
"Gas Station": ["BP", "SHELL", "ESSO", "GALP", "PETROBRAS"],
|
||||
"Online Shopping": ["AMAZON", "EBAY", "ZALANDO", "ASOS", "APPLE"],
|
||||
"Restaurant": ["PIZZA HUT", "MCDONALD'S", "BURGER KING", "LOCAL RESTAURANT"],
|
||||
"Salary": ["MONTHLY SALARY", "PAYROLL DEPOSIT", "SALARY PAYMENT"],
|
||||
"ATM Withdrawal": ["ATM WITHDRAWAL", "CASH WITHDRAWAL"],
|
||||
"Transfer to Savings": ["SAVINGS TRANSFER", "INVESTMENT TRANSFER"],
|
||||
}
|
||||
|
||||
specific_descriptions = descriptions.get(transaction_type["description"], [transaction_type["description"]])
|
||||
description = random.choice(specific_descriptions)
|
||||
|
||||
# Create raw transaction (simplified GoCardless format)
|
||||
raw_transaction = {
|
||||
"transactionId": transaction_id,
|
||||
"bookingDate": transaction_date.strftime("%Y-%m-%d"),
|
||||
"valueDate": transaction_date.strftime("%Y-%m-%d"),
|
||||
"transactionAmount": {
|
||||
"amount": str(amount),
|
||||
"currency": account["currency"]
|
||||
},
|
||||
"remittanceInformationUnstructured": description,
|
||||
"bankTransactionCode": "PMNT" if amount < 0 else "RCDT",
|
||||
}
|
||||
|
||||
# Determine status (most are booked, some recent ones might be pending)
|
||||
status = "pending" if days_ago < 2 and random.random() < 0.1 else "booked"
|
||||
|
||||
transaction = {
|
||||
"accountId": account["id"],
|
||||
"transactionId": transaction_id,
|
||||
"internalTransactionId": internal_transaction_id,
|
||||
"institutionId": account["institution_id"],
|
||||
"iban": account["iban"],
|
||||
"transactionDate": transaction_date.isoformat(),
|
||||
"description": description,
|
||||
"transactionValue": amount,
|
||||
"transactionCurrency": account["currency"],
|
||||
"transactionStatus": status,
|
||||
"rawTransaction": raw_transaction,
|
||||
}
|
||||
|
||||
account_transactions.append(transaction)
|
||||
current_balance += amount
|
||||
|
||||
# Sort transactions by date for realistic ordering
|
||||
account_transactions.sort(key=lambda x: x["transactionDate"])
|
||||
transactions.extend(account_transactions)
|
||||
|
||||
return transactions
|
||||
|
||||
def generate_balances(self, accounts: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""Generate sample balances for accounts."""
|
||||
balances = []
|
||||
|
||||
for account in accounts:
|
||||
# Calculate balance from transactions (simplified)
|
||||
base_balance = random.uniform(500, 2000)
|
||||
|
||||
balance_types = ["interimAvailable", "closingBooked", "authorised"]
|
||||
|
||||
for balance_type in balance_types:
|
||||
# Add some variation to balance types
|
||||
variation = random.uniform(-50, 50) if balance_type != "interimAvailable" else 0
|
||||
balance_amount = base_balance + variation
|
||||
|
||||
balance = {
|
||||
"account_id": account["id"],
|
||||
"bank": account["institution_id"],
|
||||
"status": account["status"],
|
||||
"iban": account["iban"],
|
||||
"amount": round(balance_amount, 2),
|
||||
"currency": account["currency"],
|
||||
"type": balance_type,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
}
|
||||
balances.append(balance)
|
||||
|
||||
return balances
|
||||
|
||||
def insert_data(self, accounts: List[Dict[str, Any]], transactions: List[Dict[str, Any]], balances: List[Dict[str, Any]]):
|
||||
"""Insert generated data into the database."""
|
||||
conn = sqlite3.connect(str(self.db_path))
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Insert accounts
|
||||
for account in accounts:
|
||||
cursor.execute("""
|
||||
INSERT OR REPLACE INTO accounts
|
||||
(id, institution_id, status, iban, name, currency, created, last_accessed, last_updated)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
account["id"], account["institution_id"], account["status"], account["iban"],
|
||||
account["name"], account["currency"], account["created"],
|
||||
account["last_accessed"], account["last_updated"]
|
||||
))
|
||||
|
||||
# Insert transactions
|
||||
for transaction in transactions:
|
||||
cursor.execute("""
|
||||
INSERT OR REPLACE INTO transactions
|
||||
(accountId, transactionId, internalTransactionId, institutionId, iban,
|
||||
transactionDate, description, transactionValue, transactionCurrency,
|
||||
transactionStatus, rawTransaction)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
transaction["accountId"], transaction["transactionId"],
|
||||
transaction["internalTransactionId"], transaction["institutionId"],
|
||||
transaction["iban"], transaction["transactionDate"], transaction["description"],
|
||||
transaction["transactionValue"], transaction["transactionCurrency"],
|
||||
transaction["transactionStatus"], json.dumps(transaction["rawTransaction"])
|
||||
))
|
||||
|
||||
# Insert balances
|
||||
for balance in balances:
|
||||
cursor.execute("""
|
||||
INSERT INTO balances
|
||||
(account_id, bank, status, iban, amount, currency, type, timestamp)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
balance["account_id"], balance["bank"], balance["status"], balance["iban"],
|
||||
balance["amount"], balance["currency"], balance["type"], balance["timestamp"]
|
||||
))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def generate_sample_database(self, num_accounts: int = 3, num_transactions_per_account: int = 50):
|
||||
"""Generate complete sample database."""
|
||||
click.echo(f"🗄️ Creating sample database at: {self.db_path}")
|
||||
|
||||
self.ensure_database_dir()
|
||||
self.create_tables()
|
||||
|
||||
click.echo(f"👥 Generating {num_accounts} sample accounts...")
|
||||
accounts = self.generate_accounts(num_accounts)
|
||||
|
||||
click.echo(f"💳 Generating {num_transactions_per_account} transactions per account...")
|
||||
transactions = self.generate_transactions(accounts, num_transactions_per_account)
|
||||
|
||||
click.echo("💰 Generating account balances...")
|
||||
balances = self.generate_balances(accounts)
|
||||
|
||||
click.echo("💾 Inserting data into database...")
|
||||
self.insert_data(accounts, transactions, balances)
|
||||
|
||||
# Print summary
|
||||
click.echo("\n✅ Sample database created successfully!")
|
||||
click.echo(f"📊 Summary:")
|
||||
click.echo(f" - Accounts: {len(accounts)}")
|
||||
click.echo(f" - Transactions: {len(transactions)}")
|
||||
click.echo(f" - Balances: {len(balances)}")
|
||||
click.echo(f" - Database: {self.db_path}")
|
||||
|
||||
# Show account details
|
||||
click.echo(f"\n📋 Sample accounts:")
|
||||
for account in accounts:
|
||||
institution_name = next(inst["name"] for inst in self.institutions if inst["id"] == account["institution_id"])
|
||||
click.echo(f" - {account['id']} ({institution_name}) - {account['iban']}")
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--database",
|
||||
type=click.Path(path_type=Path),
|
||||
help="Path to database file (default: uses LEGGEN_DATABASE_PATH or ~/.config/leggen/leggen-dev.db)",
|
||||
)
|
||||
@click.option(
|
||||
"--accounts",
|
||||
type=int,
|
||||
default=3,
|
||||
help="Number of sample accounts to generate (default: 3)",
|
||||
)
|
||||
@click.option(
|
||||
"--transactions",
|
||||
type=int,
|
||||
default=50,
|
||||
help="Number of transactions per account (default: 50)",
|
||||
)
|
||||
@click.option(
|
||||
"--force",
|
||||
is_flag=True,
|
||||
help="Overwrite existing database without confirmation",
|
||||
)
|
||||
def main(database: Path, accounts: int, transactions: int, force: bool):
|
||||
"""Generate a sample database with realistic financial data for testing Leggen."""
|
||||
|
||||
# Determine database path
|
||||
if database:
|
||||
db_path = database
|
||||
else:
|
||||
# Use development database by default to avoid overwriting production data
|
||||
import os
|
||||
env_path = os.environ.get("LEGGEN_DATABASE_PATH")
|
||||
if env_path:
|
||||
db_path = Path(env_path)
|
||||
else:
|
||||
# Default to development database in config directory
|
||||
db_path = path_manager.get_config_dir() / "leggen-dev.db"
|
||||
|
||||
# Check if database exists and ask for confirmation
|
||||
if db_path.exists() and not force:
|
||||
click.echo(f"⚠️ Database already exists: {db_path}")
|
||||
if not click.confirm("Do you want to overwrite it?"):
|
||||
click.echo("Aborted.")
|
||||
return
|
||||
|
||||
# Generate the sample database
|
||||
generator = SampleDataGenerator(db_path)
|
||||
generator.generate_sample_database(accounts, transactions)
|
||||
|
||||
# Show usage instructions
|
||||
click.echo(f"\n🚀 Usage instructions:")
|
||||
click.echo(f"To use this sample database with leggen commands:")
|
||||
click.echo(f" export LEGGEN_DATABASE_PATH={db_path}")
|
||||
click.echo(f" leggen transactions")
|
||||
click.echo(f"")
|
||||
click.echo(f"To use this sample database with leggend API:")
|
||||
click.echo(f" leggend --database {db_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -86,23 +86,17 @@ def api_client(fastapi_app):
|
||||
@pytest.fixture
|
||||
def mock_db_path(temp_db_path):
|
||||
"""Mock the database path to use temporary database for testing."""
|
||||
from pathlib import Path
|
||||
|
||||
# Create the expected directory structure
|
||||
temp_home = temp_db_path.parent
|
||||
config_dir = temp_home / ".config" / "leggen"
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create the expected database path
|
||||
expected_db_path = config_dir / "leggen.db"
|
||||
|
||||
# Mock Path.home to return our temp directory
|
||||
def mock_home():
|
||||
return temp_home
|
||||
|
||||
# Patch Path.home in the main pathlib module
|
||||
with patch.object(Path, "home", staticmethod(mock_home)):
|
||||
yield expected_db_path
|
||||
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:
|
||||
# Restore original path
|
||||
path_manager._database_path = original_database_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
162
tests/unit/test_configurable_paths.py
Normal file
162
tests/unit/test_configurable_paths.py
Normal file
@@ -0,0 +1,162 @@
|
||||
"""Integration tests for configurable paths."""
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from leggen.utils.paths import path_manager
|
||||
from leggen.database.sqlite import persist_balances, get_balances
|
||||
|
||||
|
||||
class MockContext:
|
||||
"""Mock context for testing."""
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestConfigurablePaths:
|
||||
"""Test configurable path management."""
|
||||
|
||||
def test_default_paths(self):
|
||||
"""Test that default paths are correctly set."""
|
||||
# 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:
|
||||
path_manager._config_dir = original_config
|
||||
path_manager._database_path = original_db
|
||||
|
||||
def test_environment_variables(self):
|
||||
"""Test that environment variables override defaults."""
|
||||
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)
|
||||
}):
|
||||
# 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:
|
||||
path_manager._config_dir = original_config
|
||||
path_manager._database_path = original_db
|
||||
|
||||
def test_explicit_path_setting(self):
|
||||
"""Test explicitly setting paths."""
|
||||
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"
|
||||
finally:
|
||||
# Restore original paths
|
||||
path_manager._config_dir = original_config
|
||||
path_manager._database_path = original_db
|
||||
|
||||
def test_database_operations_with_custom_path(self):
|
||||
"""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 = {
|
||||
"account_id": "test-account",
|
||||
"bank": "TEST_BANK",
|
||||
"status": "active",
|
||||
"iban": "TEST_IBAN",
|
||||
"amount": 1000.0,
|
||||
"currency": "EUR",
|
||||
"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
|
||||
if test_db_path.exists():
|
||||
test_db_path.unlink()
|
||||
|
||||
def test_directory_creation(self):
|
||||
"""Test that directories are created as needed."""
|
||||
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
|
||||
@@ -21,14 +21,18 @@ def temp_db_path():
|
||||
|
||||
@pytest.fixture
|
||||
def mock_home_db_path(temp_db_path):
|
||||
"""Mock the home database path to use temp file."""
|
||||
config_dir = temp_db_path.parent / ".config" / "leggen"
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
db_file = config_dir / "leggen.db"
|
||||
|
||||
with patch("pathlib.Path.home") as mock_home:
|
||||
mock_home.return_value = temp_db_path.parent
|
||||
yield db_file
|
||||
"""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:
|
||||
# Restore original path
|
||||
path_manager._database_path = original_database_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -90,18 +94,14 @@ class TestSQLiteDatabase:
|
||||
"""Test persisting transactions to database."""
|
||||
ctx = MockContext()
|
||||
|
||||
# Mock the database path
|
||||
with patch("pathlib.Path.home") as mock_home:
|
||||
mock_home.return_value = mock_home_db_path.parent / ".."
|
||||
# Persist transactions
|
||||
new_transactions = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
|
||||
# Persist transactions
|
||||
new_transactions = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
|
||||
# Should return all transactions as new
|
||||
assert len(new_transactions) == 2
|
||||
assert new_transactions[0]["internalTransactionId"] == "txn-001"
|
||||
# Should return all transactions as new
|
||||
assert len(new_transactions) == 2
|
||||
assert new_transactions[0]["internalTransactionId"] == "txn-001"
|
||||
|
||||
def test_persist_transactions_duplicates(
|
||||
self, mock_home_db_path, sample_transactions
|
||||
@@ -109,40 +109,34 @@ class TestSQLiteDatabase:
|
||||
"""Test handling duplicate transactions."""
|
||||
ctx = MockContext()
|
||||
|
||||
with patch("pathlib.Path.home") as mock_home:
|
||||
mock_home.return_value = mock_home_db_path.parent / ".."
|
||||
# Insert transactions twice
|
||||
new_transactions_1 = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
new_transactions_2 = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
|
||||
# Insert transactions twice
|
||||
new_transactions_1 = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
new_transactions_2 = sqlite_db.persist_transactions(
|
||||
ctx, "test-account-123", sample_transactions
|
||||
)
|
||||
|
||||
# First time should return all as new
|
||||
assert len(new_transactions_1) == 2
|
||||
# Second time should also return all (INSERT OR REPLACE behavior with composite key)
|
||||
assert len(new_transactions_2) == 2
|
||||
# First time should return all as new
|
||||
assert len(new_transactions_1) == 2
|
||||
# Second time should also return all (INSERT OR REPLACE behavior with composite key)
|
||||
assert len(new_transactions_2) == 2
|
||||
|
||||
def test_get_transactions_all(self, mock_home_db_path, sample_transactions):
|
||||
"""Test retrieving all transactions."""
|
||||
ctx = MockContext()
|
||||
|
||||
with patch("pathlib.Path.home") as mock_home:
|
||||
mock_home.return_value = mock_home_db_path.parent / ".."
|
||||
# Insert test data
|
||||
sqlite_db.persist_transactions(ctx, "test-account-123", sample_transactions)
|
||||
|
||||
# Insert test data
|
||||
sqlite_db.persist_transactions(ctx, "test-account-123", sample_transactions)
|
||||
# Get all transactions
|
||||
transactions = sqlite_db.get_transactions()
|
||||
|
||||
# Get all transactions
|
||||
transactions = sqlite_db.get_transactions()
|
||||
|
||||
assert len(transactions) == 2
|
||||
assert (
|
||||
transactions[0]["internalTransactionId"] == "txn-002"
|
||||
) # Ordered by date DESC
|
||||
assert transactions[1]["internalTransactionId"] == "txn-001"
|
||||
assert len(transactions) == 2
|
||||
assert (
|
||||
transactions[0]["internalTransactionId"] == "txn-002"
|
||||
) # Ordered by date DESC
|
||||
assert transactions[1]["internalTransactionId"] == "txn-001"
|
||||
|
||||
def test_get_transactions_filtered_by_account(
|
||||
self, mock_home_db_path, sample_transactions
|
||||
|
||||
Reference in New Issue
Block a user