diff --git a/src/dualentry_cli/commands/__init__.py b/src/dualentry_cli/commands/__init__.py index dbf2615..47ebd4a 100644 --- a/src/dualentry_cli/commands/__init__.py +++ b/src/dualentry_cli/commands/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from collections.abc import Callable from pathlib import Path import typer @@ -97,6 +98,19 @@ def _load_json_file(file: Path) -> dict: raise typer.Exit(code=1) from None +# ── Post command helpers ─────────────────────────────────────────── + +_WRITABLE_FIELDS = {"date", "transaction_date", "memo", "currency_iso_4217_code", "exchange_rate", "record_status", "items", "attachments"} +_WRITABLE_ITEM_FIELDS = {"id", "company_id", "account_number", "debit", "credit", "memo", "position", "classifications", "customer_id", "vendor_id", "currency", "eliminate"} + + +def _strip_to_writable(data: dict) -> dict: + payload = {k: v for k, v in data.items() if k in _WRITABLE_FIELDS} + if "items" in payload: + payload["items"] = [{k: v for k, v in item.items() if k in _WRITABLE_ITEM_FIELDS} for item in payload["items"]] + return payload + + # ── Factory ───────────────────────────────────────────────────────── @@ -108,6 +122,9 @@ def make_resource_app( has_update: bool = True, has_delete: bool = False, has_number: bool = False, + has_post: bool = False, + template: dict | None = None, + validate_fn: Callable[[Path], None] | None = None, ) -> typer.Typer: """Create a Typer app for a standard CRUD resource.""" app = typer.Typer(help=f"Manage {name}", no_args_is_help=True, cls=HelpfulGroup) @@ -246,4 +263,54 @@ def delete_cmd( delete_cmd.__doc__ = f"Delete a {resource}." + if validate_fn is not None: + + @app.command("validate") + def validate_cmd( + file: Path = typer.Option(..., "--file", "-f", help="JSON file to validate"), + ): + validate_fn(file) + + validate_cmd.__doc__ = f"Validate a {resource} payload (client-side)." + + if has_post: + + @app.command("post") + def post_cmd( + number: str = typer.Argument(help="Record number of the draft to post"), + output: str = Format, + ): + from dualentry_cli.main import get_client + + client = get_client() + stripped = _strip_record_prefix(number) + data = client.get(f"/{path}/{stripped}/") + + current_status = data.get("record_status", "") + if current_status != "draft": + typer.secho(f" \u2717 Cannot post: record is '{current_status}', only draft records can be posted.", fg=typer.colors.RED, err=True) + raise typer.Exit(code=1) + + payload = _strip_to_writable(data) + payload["record_status"] = "posted" + result = client.put(f"/{path}/{stripped}/", json=payload) + format_output(result, resource=resource, fmt=output) + + post_cmd.__doc__ = f"Post a draft {resource}." + + if template is not None: + + @app.command("template") + def template_cmd( + output_file: Path | None = typer.Option(None, "--output", "-o", help="Write template to file instead of stdout"), + ): + content = json.dumps(template, indent=2) + if output_file: + output_file.write_text(content + "\n") + typer.secho(f"Template written to {output_file}", fg=typer.colors.GREEN) + else: + typer.echo(content) + + template_cmd.__doc__ = f"Output a sample {resource} JSON template." + return app diff --git a/src/dualentry_cli/commands/ije_extras.py b/src/dualentry_cli/commands/ije_extras.py new file mode 100644 index 0000000..623aa90 --- /dev/null +++ b/src/dualentry_cli/commands/ije_extras.py @@ -0,0 +1,101 @@ +"""Intercompany journal entry template and validation logic.""" + +from __future__ import annotations + +from pathlib import Path + +import typer + +from dualentry_cli.commands import _load_json_file + +IJE_TEMPLATE = { + "date": "2026-01-01", + "memo": "Intercompany transfer", + "currency_iso_4217_code": "USD", + "exchange_rate": "1.00000000", + "record_status": "draft", + "items": [ + { + "company_id": 1, + "account_number": 1000, + "debit": "1000.00", + "credit": "0.00", + "memo": "", + "position": 0, + "eliminate": True, + }, + { + "company_id": 1, + "account_number": 2000, + "debit": "0.00", + "credit": "1000.00", + "memo": "", + "position": 1, + "eliminate": True, + }, + { + "company_id": 2, + "account_number": 1000, + "debit": "1000.00", + "credit": "0.00", + "memo": "", + "position": 2, + "eliminate": True, + }, + { + "company_id": 2, + "account_number": 2000, + "debit": "0.00", + "credit": "1000.00", + "memo": "", + "position": 3, + "eliminate": True, + }, + ], +} + + +def validate_ije(file: Path) -> None: + from decimal import Decimal, InvalidOperation + + payload = _load_json_file(file) + errors: list[str] = [] + + items = payload.get("items") + if not items or not isinstance(items, list): + errors.append("Payload must contain a non-empty 'items' array.") + else: + company_ids = set() + total_debits = Decimal(0) + total_credits = Decimal(0) + + for i, item in enumerate(items): + cid = item.get("company_id") + if cid is not None: + company_ids.add(cid) + + try: + debit = Decimal(str(item.get("debit", "0"))) + credit = Decimal(str(item.get("credit", "0"))) + except (InvalidOperation, TypeError): + errors.append(f"Item {i}: invalid debit/credit value.") + continue + + total_debits += debit + total_credits += credit + + if not errors: + if len(company_ids) < 2: + errors.append("Intercompany journal entries require lines across at least two distinct companies.") + + total_debits = total_debits.quantize(Decimal("0.01")) + total_credits = total_credits.quantize(Decimal("0.01")) + if total_debits != total_credits: + errors.append(f"Total debits ({total_debits}) must equal total credits ({total_credits}).") + + if errors: + for err in errors: + typer.secho(f" \u2717 {err}", fg=typer.colors.RED, err=True) + raise typer.Exit(code=1) + + typer.secho(" \u2713 Valid", fg=typer.colors.GREEN) diff --git a/src/dualentry_cli/main.py b/src/dualentry_cli/main.py index 612095b..e8bdc0d 100644 --- a/src/dualentry_cli/main.py +++ b/src/dualentry_cli/main.py @@ -8,6 +8,7 @@ from dualentry_cli.cli import HelpfulGroup from dualentry_cli.commands import make_resource_app from dualentry_cli.commands.accounts import app as accounts_app +from dualentry_cli.commands.ije_extras import IJE_TEMPLATE, validate_ije from dualentry_cli.config import Config app = typer.Typer(name="dualentry", help="DualEntry accounting CLI", no_args_is_help=True, cls=HelpfulGroup) @@ -69,7 +70,18 @@ app.add_typer(make_resource_app("contracts", "contract", "contracts"), name="contracts") app.add_typer(make_resource_app("budgets", "budget", "budgets"), name="budgets") app.add_typer(make_resource_app("workflows", "workflow", "workflows", has_create=False, has_update=False), name="workflows") -app.add_typer(make_resource_app("intercompany journal entries", "intercompany-journal-entry", "intercompany-journal-entries", has_number=True), name="intercompany-journal-entries") +app.add_typer( + make_resource_app( + "intercompany journal entries", + "intercompany-journal-entry", + "intercompany-journal-entries", + has_number=True, + has_post=True, + template=IJE_TEMPLATE, + validate_fn=validate_ije, + ), + name="intercompany-journal-entries", +) app.add_typer(make_resource_app("paper checks", "paper-check", "paper-checks", has_number=True), name="paper-checks") app.add_typer(make_resource_app("inbox items", "inbox-item", "inbox", has_create=False, has_update=False), name="inbox") diff --git a/src/dualentry_cli/output.py b/src/dualentry_cli/output.py index e94722b..8203ae8 100644 --- a/src/dualentry_cli/output.py +++ b/src/dualentry_cli/output.py @@ -31,6 +31,7 @@ "vendor-prepayment-application": "VPA", "vendor-refund": "VR", "journal-entry": "JE", + "intercompany-journal-entry": "IJE", "bank-transfer": "BT", "fixed-asset": "FA", } @@ -545,6 +546,102 @@ def _journal_entry_detail(r): _register("journal-entry", _journal_entry_list, _journal_entry_detail) +# ── Intercompany Journal Entry ────────────────────────────────────── + + +def _ije_list(items): + table = Table(title="Intercompany Journal Entries", show_lines=False) + table.add_column("ID", style="dim", justify="right") + table.add_column("#", style="bold", justify="right") + table.add_column("Date", justify="center") + table.add_column("Companies", min_width=20) + table.add_column("Memo", max_width=20) + table.add_column("Currency", justify="center") + table.add_column("Amount", justify="right", style="bold") + table.add_column("Status") + + for r in items: + currency = r.get("currency_iso_4217_code", "") + companies = r.get("companies", []) + company_names = ", ".join(c.get("name", "") for c in companies) if companies else r.get("company_name", "-") + memo = r.get("memo", "") or "" + amount = sum(float(item.get("debit") or 0) for item in r.get("items", [])) + table.add_row( + _fmt_id(r.get("internal_id"), "intercompany-journal-entry"), + str(r.get("record_number", "")), + r.get("date", "-"), + company_names, + memo[:20] + ("..." if len(memo) > 20 else ""), + currency, + _money(amount or r.get("amount"), currency), + _status_badge(r.get("record_status", "")), + ) + + console.print(table) + + +def _ije_detail(r): + currency = r.get("currency_iso_4217_code", "") + + header = Text() + header.append("INTERCOMPANY JOURNAL ENTRY", style="bold") + header.append(f" {_fmt_id(r.get('internal_id'), 'intercompany-journal-entry')}", style="bold cyan") + status = r.get("record_status", "") + if status: + header.append(f" {status.upper()}", style=_status_color(status)) + console.print(Panel(header, expand=False)) + + details = Table.grid(padding=(0, 2)) + details.add_column(style="dim", min_width=20) + details.add_column() + details.add_row("Number:", str(r.get("record_number", "-"))) + details.add_row("Date:", r.get("date", "-")) + tx_date = r.get("transaction_date") + if tx_date and tx_date != r.get("date"): + details.add_row("Transaction Date:", tx_date) + companies = r.get("companies", []) + if companies: + details.add_row("Companies:", ", ".join(c.get("name", "") for c in companies)) + details.add_row("Currency:", currency or "-") + exchange_rate = r.get("exchange_rate", "") + if exchange_rate and exchange_rate not in ("1", "1.00", "1.00000000"): + details.add_row("Exchange Rate:", exchange_rate) + if r.get("memo"): + details.add_row("Memo:", r["memo"]) + console.print(details) + console.print() + + items = r.get("items", []) + if items: + items_sorted = sorted(items, key=lambda x: x.get("position", 0)) + je_table = Table(show_lines=True, title="Entries") + je_table.add_column("#", justify="right", style="dim", width=4) + je_table.add_column("Company", min_width=14) + je_table.add_column("Account", min_width=20) + je_table.add_column("Memo", min_width=16) + je_table.add_column("Debit", justify="right", width=14) + je_table.add_column("Credit", justify="right", width=14) + je_table.add_column("Elim", justify="center", width=5) + + for i, item in enumerate(items_sorted, 1): + account = item.get("account_name") or str(item.get("account_number", "")) + elim = "\u2713" if item.get("eliminate") else "" + je_table.add_row( + str(i), + item.get("company_name", "-"), + account, + item.get("memo", ""), + _money(item.get("debit"), currency) if item.get("debit") else "", + _money(item.get("credit"), currency) if item.get("credit") else "", + elim, + ) + + console.print(je_table) + + +_register("intercompany-journal-entry", _ije_list, _ije_detail) + + # ── Bank Transfer ──────────────────────────────────────────────────── diff --git a/tests/test_intercompany_journal_entries.py b/tests/test_intercompany_journal_entries.py new file mode 100644 index 0000000..f8ecd9c --- /dev/null +++ b/tests/test_intercompany_journal_entries.py @@ -0,0 +1,328 @@ +import json +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from dualentry_cli.main import app +from dualentry_cli.output import format_output + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def mock_get_client(): + mock_client = MagicMock() + with patch("dualentry_cli.main.get_client", return_value=mock_client): + yield mock_client + + +class TestIJEFormatting: + def test_ije_list_shows_prefix(self, capsys): + data = { + "items": [ + { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "record_status": "posted", + "companies": [ + {"id": 1, "name": "Company A"}, + {"id": 2, "name": "Company B"}, + ], + "items": [ + {"debit": "1000.00", "credit": "0.00"}, + {"debit": "0.00", "credit": "1000.00"}, + ], + } + ], + "count": 1, + } + format_output(data, resource="intercompany-journal-entry", fmt="human") + captured = capsys.readouterr() + assert "IJE-100" in captured.out + assert "Company A" in captured.out + assert "Company B" in captured.out + assert "IC" in captured.out + assert "transfer" in captured.out + + def test_ije_detail_shows_lines_with_company(self, capsys): + data = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "transaction_date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "exchange_rate": "1.00000000", + "record_status": "posted", + "companies": [ + {"id": 1, "name": "Company A"}, + {"id": 2, "name": "Company B"}, + ], + "items": [ + { + "id": 1, + "company_id": 1, + "company_name": "Company A", + "account_number": 1000, + "account_name": "Cash", + "debit": "1000.00", + "credit": "0.00", + "memo": "Debit line", + "position": 0, + "eliminate": True, + }, + { + "id": 2, + "company_id": 2, + "company_name": "Company B", + "account_number": 2000, + "account_name": "Payable", + "debit": "0.00", + "credit": "1000.00", + "memo": "Credit line", + "position": 1, + "eliminate": False, + }, + ], + } + format_output(data, resource="intercompany-journal-entry", fmt="human") + captured = capsys.readouterr() + assert "INTERCOMPANY JOURNAL ENTRY" in captured.out + assert "IJE-100" in captured.out + assert "Company A" in captured.out + assert "Company B" in captured.out + assert "Cash" in captured.out + assert "Payable" in captured.out + + +class TestIJECommands: + def test_list(self, mock_get_client): + mock_get_client.get.return_value = { + "items": [ + { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "record_status": "posted", + "companies": [{"id": 1, "name": "Co A"}, {"id": 2, "name": "Co B"}], + "items": [{"debit": "1000.00", "credit": "0.00"}, {"debit": "0.00", "credit": "1000.00"}], + } + ], + "count": 1, + } + result = runner.invoke(app, ["intercompany-journal-entries", "list"]) + assert result.exit_code == 0 + assert "IJE-100" in result.output + mock_get_client.get.assert_called_once_with("/intercompany-journal-entries/", params={"limit": 20, "offset": 0}) + + def test_get_by_number(self, mock_get_client): + mock_get_client.get.return_value = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "record_status": "posted", + "companies": [], + "items": [], + } + result = runner.invoke(app, ["intercompany-journal-entries", "get", "42"]) + assert result.exit_code == 0 + assert "IJE-100" in result.output + mock_get_client.get.assert_called_once_with("/intercompany-journal-entries/42/") + + def test_create(self, mock_get_client, tmp_path): + payload = {"date": "2026-04-20", "memo": "test", "items": []} + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + mock_get_client.post.return_value = { + "internal_id": 101, + "record_number": 43, + "date": "2026-04-20", + "memo": "test", + "currency_iso_4217_code": "USD", + "record_status": "draft", + "companies": [], + "items": [], + } + result = runner.invoke(app, ["intercompany-journal-entries", "create", "--file", str(data_file)]) + assert result.exit_code == 0 + mock_get_client.post.assert_called_once_with("/intercompany-journal-entries/", json=payload) + + def test_update(self, mock_get_client, tmp_path): + payload = {"memo": "updated"} + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + mock_get_client.put.return_value = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "updated", + "currency_iso_4217_code": "USD", + "record_status": "draft", + "companies": [], + "items": [], + } + result = runner.invoke(app, ["intercompany-journal-entries", "update", "42", "--file", str(data_file)]) + assert result.exit_code == 0 + mock_get_client.put.assert_called_once_with("/intercompany-journal-entries/42/", json=payload) + + +class TestIJEValidate: + def test_validate_valid_payload(self, tmp_path): + payload = { + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "items": [ + {"company_id": 1, "account_number": 1000, "debit": "1000.00", "credit": "0.00"}, + {"company_id": 2, "account_number": 2000, "debit": "0.00", "credit": "1000.00"}, + ], + } + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + result = runner.invoke(app, ["intercompany-journal-entries", "validate", "--file", str(data_file)]) + assert result.exit_code == 0 + assert "Valid" in result.output + + def test_validate_unbalanced(self, tmp_path): + payload = { + "items": [ + {"company_id": 1, "account_number": 1000, "debit": "1000.00", "credit": "0.00"}, + {"company_id": 2, "account_number": 2000, "debit": "0.00", "credit": "500.00"}, + ], + } + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + result = runner.invoke(app, ["intercompany-journal-entries", "validate", "--file", str(data_file)]) + assert result.exit_code == 1 + assert "debit" in result.output.lower() or "balance" in result.output.lower() + + def test_validate_single_company(self, tmp_path): + payload = { + "items": [ + {"company_id": 1, "account_number": 1000, "debit": "1000.00", "credit": "0.00"}, + {"company_id": 1, "account_number": 2000, "debit": "0.00", "credit": "1000.00"}, + ], + } + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + result = runner.invoke(app, ["intercompany-journal-entries", "validate", "--file", str(data_file)]) + assert result.exit_code == 1 + assert "two" in result.output.lower() or "companies" in result.output.lower() + + def test_validate_missing_items(self, tmp_path): + payload = {"date": "2026-04-20", "memo": "IC transfer"} + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + result = runner.invoke(app, ["intercompany-journal-entries", "validate", "--file", str(data_file)]) + assert result.exit_code == 1 + assert "items" in result.output.lower() + + def test_validate_empty_items(self, tmp_path): + payload = {"items": []} + data_file = tmp_path / "ije.json" + data_file.write_text(json.dumps(payload)) + result = runner.invoke(app, ["intercompany-journal-entries", "validate", "--file", str(data_file)]) + assert result.exit_code == 1 + + +class TestIJEPost: + def test_post_draft_to_posted(self, mock_get_client): + draft_response = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "transaction_date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "exchange_rate": "1.00000000", + "record_status": "draft", + "companies": [{"id": 1, "name": "Co A"}, {"id": 2, "name": "Co B"}, {"id": 3, "name": "Elim Co"}], + "company_ids": [1, 2, 3], + "items": [ + {"id": 1, "company_id": 1, "company_name": "Co A", "account_number": 1000, "debit": "1000.00", "credit": "0.00", "memo": "", "position": 0, "eliminate": True}, + {"id": 2, "company_id": 2, "company_name": "Co B", "account_number": 2000, "debit": "0.00", "credit": "1000.00", "memo": "", "position": 1, "eliminate": True}, + {"id": 3, "company_id": 3, "company_name": "Elim Co", "account_number": 1000, "debit": "0.00", "credit": "1000.00", "memo": "", "position": 2, "eliminate": True}, + {"id": 4, "company_id": 3, "company_name": "Elim Co", "account_number": 2000, "debit": "1000.00", "credit": "0.00", "memo": "", "position": 3, "eliminate": True}, + ], + } + posted_response = {**draft_response, "record_status": "posted"} + mock_get_client.get.return_value = draft_response + mock_get_client.put.return_value = posted_response + result = runner.invoke(app, ["intercompany-journal-entries", "post", "42"]) + assert result.exit_code == 0 + assert "POSTED" in result.output + put_call = mock_get_client.put.call_args + assert put_call[0][0] == "/intercompany-journal-entries/42/" + put_payload = put_call[1]["json"] + assert put_payload["record_status"] == "posted" + assert "companies" not in put_payload + assert "company_ids" not in put_payload + assert "record_number" not in put_payload + assert "internal_id" not in put_payload + assert "company_name" not in put_payload["items"][0] + + def test_post_already_posted_fails(self, mock_get_client): + mock_get_client.get.return_value = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "record_status": "posted", + "companies": [], + "items": [], + } + result = runner.invoke(app, ["intercompany-journal-entries", "post", "42"]) + assert result.exit_code == 1 + assert "already" in result.output.lower() or "draft" in result.output.lower() + + def test_post_archived_fails(self, mock_get_client): + mock_get_client.get.return_value = { + "internal_id": 100, + "record_number": 42, + "date": "2026-04-20", + "memo": "IC transfer", + "currency_iso_4217_code": "USD", + "record_status": "archived", + "companies": [], + "items": [], + } + result = runner.invoke(app, ["intercompany-journal-entries", "post", "42"]) + assert result.exit_code == 1 + + +class TestIJETemplate: + def test_template_stdout(self): + result = runner.invoke(app, ["intercompany-journal-entries", "template"]) + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert "date" in parsed + assert "items" in parsed + assert len(parsed["items"]) == 4 + company_ids = {item["company_id"] for item in parsed["items"]} + assert len(company_ids) >= 2 + assert parsed["record_status"] == "draft" + + def test_template_to_file(self, tmp_path): + out_file = tmp_path / "template.json" + result = runner.invoke(app, ["intercompany-journal-entries", "template", "--output", str(out_file)]) + assert result.exit_code == 0 + assert out_file.exists() + parsed = json.loads(out_file.read_text()) + assert "items" in parsed + assert len(parsed["items"]) == 4 + + def test_template_is_balanced(self): + result = runner.invoke(app, ["intercompany-journal-entries", "template"]) + parsed = json.loads(result.output) + total_debits = sum(float(item["debit"]) for item in parsed["items"]) + total_credits = sum(float(item["credit"]) for item in parsed["items"]) + assert total_debits == total_credits