diff --git a/app.py b/app.py index 170ed43..34f0279 100644 --- a/app.py +++ b/app.py @@ -327,7 +327,7 @@ class Invoice(db.Model): quantity = db.Column(db.Float, default=1) unit_price_usd = db.Column(db.Float, nullable=True) notes = db.Column(db.Text) - status = db.Column(db.String(20), default='pending') + status = db.Column(db.String(20), default='draft') # draft, issued, paid, cancelled (legacy: pending) pdf_hash = db.Column(db.String(64)) # SHA256 hash of the PDF file pdf_storage_key = db.Column(db.String(500)) # Storage key for PDF (local path or remote ID) currency = db.Column(db.String(10), default='USD') # Invoice currency @@ -335,7 +335,21 @@ class Invoice(db.Model): created_at = db.Column(db.DateTime, default=datetime.utcnow) customer_id = db.Column(db.Integer, db.ForeignKey('customer.id'), nullable=True) bank_id = db.Column(db.Integer, db.ForeignKey('bank.id'), nullable=True) - items = db.relationship('InvoiceItem', backref='invoice', lazy=True, cascade='all, delete-orphan') + # F2 — lifecycle: series + per-series sequence assigned at ISSUE time. + series = db.Column(db.String(20)) + sequence_number = db.Column(db.Integer) + issued_at = db.Column(db.DateTime) + # F2-D3 — facturas rectificativas linkage. + rectifies_invoice_id = db.Column(db.Integer, db.ForeignKey('invoice.id'), nullable=True) + rectification_type = db.Column(db.String(20)) # 'sustitucion' | 'diferencias' + # F2-D4 — fiscal snapshot frozen at issue (so later customer/settings edits + # never change an issued invoice's meaning). + snap_vat_rate = db.Column(db.Float) + snap_vat_amount = db.Column(db.Float) + snap_taxable_base = db.Column(db.Float) + snap_customer = db.Column(db.Text) # JSON: name, vat_number, country, tax_type + items = db.relationship('InvoiceItem', backref='invoice', lazy=True, cascade='all, delete-orphan', + foreign_keys='InvoiceItem.invoice_id') def __repr__(self): return f'' @@ -348,6 +362,8 @@ class InvoiceItem(db.Model): quantity = db.Column(db.Float, nullable=False, default=1) unit_price_usd = db.Column(db.Float, nullable=False) subtotal_usd = db.Column(db.Float, nullable=False) + # F2-D4 — per-line VAT rate (nullable; defaults to the header rate at issue). + vat_rate = db.Column(db.Float) created_at = db.Column(db.DateTime, default=datetime.utcnow) def __repr__(self): @@ -438,7 +454,7 @@ def __repr__(self): class Expense(db.Model): id = db.Column(db.Integer, primary_key=True) contractor_id = db.Column(db.Integer, db.ForeignKey('contractor.id')) - amount = db.Column(db.Float, nullable=False) + amount = db.Column(db.Float, nullable=False) # gross total (kept for back-compat) currency = db.Column(db.String(10), default='EUR') category = db.Column(db.String(100)) description = db.Column(db.Text) @@ -446,6 +462,13 @@ class Expense(db.Model): file_path = db.Column(db.String(500)) invoice_number = db.Column(db.String(100)) notes = db.Column(db.Text) + # F4 — VAT breakdown & deductibility. NULL on legacy rows = "VAT unknown" + # (excluded from Modelo 303 deductible math with a visible nudge). + net_amount = db.Column(db.Float) # taxable base (gross - vat) + vat_rate = db.Column(db.Float) # IVA soportado rate, % + vat_amount = db.Column(db.Float) # IVA soportado amount + deductible = db.Column(db.Boolean, default=True) + deductible_pct = db.Column(db.Float, default=100.0) created_at = db.Column(db.DateTime, default=datetime.utcnow) def __repr__(self): @@ -1001,6 +1024,191 @@ def index(): sort_order=sort_order) +# --- F2: Invoice lifecycle -------------------------------------------------- # + +LOCKED_INVOICE_STATUSES = ('issued', 'paid') + + +def _invoice_locked(invoice): + """F2 — issued/paid invoices are read-only (change via rectificative).""" + return invoice is not None and invoice.status in LOCKED_INVOICE_STATUSES + + +def _next_sequence_for_series(series): + """Next per-series sequence, derived from MAX (survives restart, no memory).""" + from sqlalchemy import func + max_seq = db.session.query(func.max(Invoice.sequence_number)).filter( + Invoice.series == series).scalar() + return (max_seq or 0) + 1 + + +def _build_customer_snapshot(invoice): + """Freeze the customer's fiscal identity as JSON at issue time.""" + import json as _json + c = invoice.customer + return _json.dumps({ + 'name': c.name if c else invoice.client_name, + 'vat_number': c.vat_number if c else None, + 'country': c.country if c else None, + 'tax_type': c.tax_type if c else None, + }) + + +def _issue_invoice(invoice, series=None, request_obj=None): + """Assign series+sequence, freeze the fiscal snapshot, mark issued. + + Transactional: the on_invoice_issued hooks run before commit, so a vetoing + module (or a unique-number collision) rolls the whole transition back with + no sequence gap. Raises ValueError on failure. + """ + from sqlalchemy.exc import IntegrityError + + if _invoice_locked(invoice): + raise ValueError('Invoice is already issued') + + app_settings = Settings.query.first() + default_vat = (app_settings.default_vat_rate + if app_settings and app_settings.default_vat_rate is not None else 21.0) + tax_type = invoice.customer.tax_type if invoice.customer else None + rate = default_vat if tax_type == 'standard' else 0.0 + base = invoice.amount_eur or 0.0 + + invoice.snap_vat_rate = rate + invoice.snap_taxable_base = base + invoice.snap_vat_amount = round(base * rate / 100.0, 2) + invoice.snap_customer = _build_customer_snapshot(invoice) + for item in invoice.items: + if item.vat_rate is None: + item.vat_rate = rate + + series = series or invoice.series or str(invoice.invoice_date.year) + invoice.series = series + + last_error = None + for _ in range(5): + seq = _next_sequence_for_series(series) + invoice.sequence_number = seq + invoice.invoice_number = f'{series}/{seq:04d}' + invoice.status = 'issued' + invoice.issued_at = datetime.utcnow() + try: + db.session.flush() # surfaces unique-number collisions early + break + except IntegrityError as e: + last_error = e + db.session.rollback() + continue + else: + raise ValueError(f'Could not assign a unique invoice number: {last_error}') + + # Issue-blocking hooks (may veto). Runs before commit -> abort is clean. + if module_manager: + module_manager.on_invoice_issued(invoice, request_obj) + + db.session.commit() + return invoice + + +@app.route('/issue/', methods=['POST']) +@login_required +def issue_invoice(id): + invoice = Invoice.query.get_or_404(id) + if _invoice_locked(invoice): + flash('Invoice is already issued.', 'warning') + return redirect(url_for('view_invoice', id=id)) + try: + _issue_invoice(invoice, series=request.form.get('series') or None, + request_obj=request) + log_activity('invoice_issued', 'invoice', f'#{invoice.invoice_number}') + flash(f'Invoice issued as {invoice.invoice_number}. It is now locked; ' + f'use Rectify to change it.', 'success') + except Exception as e: + db.session.rollback() + # Exception text and route params may carry user input — keep both out + # of the log line (log injection); invoice.id is DB-derived and the + # traceback via exc_info has the details. + logger.error('Issue failed for invoice %s', invoice.id, exc_info=True) + flash(f'Could not issue invoice: {e}', 'danger') + return redirect(url_for('view_invoice', id=id)) + + +@app.route('/annul/', methods=['POST']) +@login_required +def annul_invoice(id): + invoice = Invoice.query.get_or_404(id) + if invoice.status not in LOCKED_INVOICE_STATUSES: + flash('Only issued invoices can be annulled.', 'warning') + return redirect(url_for('view_invoice', id=id)) + try: + invoice.status = 'cancelled' + db.session.flush() + if module_manager: + module_manager.on_invoice_annulled(invoice, request) + db.session.commit() + log_activity('invoice_annulled', 'invoice', f'#{invoice.invoice_number}') + flash('Invoice annulled — retained for audit, excluded from income.', 'success') + except Exception as e: + db.session.rollback() + logger.error('Annul failed for invoice %s', invoice.id, exc_info=True) + flash(f'Could not annul invoice: {e}', 'danger') + return redirect(url_for('view_invoice', id=id)) + + +@app.route('/rectify/', methods=['POST']) +@login_required +def rectify_invoice(id): + original = Invoice.query.get_or_404(id) + if original.status not in LOCKED_INVOICE_STATUSES: + flash('Only issued invoices can be rectified.', 'warning') + return redirect(url_for('view_invoice', id=id)) + rtype = request.form.get('rectification_type', 'sustitucion') + if rtype not in ('sustitucion', 'diferencias'): + rtype = 'sustitucion' + year = datetime.utcnow().year + try: + # New DRAFT pre-filled from the original; original is never mutated. + draft = Invoice( + invoice_number=f'RECT-{original.id}-{datetime.utcnow().strftime("%Y%m%d%H%M%S")}', + client_name=original.client_name, + amount_usd=original.amount_usd, + amount_eur=original.amount_eur, + exchange_rate=original.exchange_rate, + invoice_date=datetime.utcnow().date(), + due_date=original.due_date, + description=original.description, + quantity=original.quantity, + unit_price_usd=original.unit_price_usd, + notes=original.notes, + status='draft', + currency=original.currency, + payment_method=original.payment_method, + customer_id=original.customer_id, + bank_id=original.bank_id, + series=f'R{year}', + rectifies_invoice_id=original.id, + rectification_type=rtype, + ) + db.session.add(draft) + db.session.flush() + for item in original.items: + db.session.add(InvoiceItem( + invoice_id=draft.id, description=item.description, + quantity=item.quantity, unit_price_usd=item.unit_price_usd, + subtotal_usd=item.subtotal_usd, vat_rate=item.vat_rate)) + if module_manager: + module_manager.on_invoice_rectified(draft, original, request) + db.session.commit() + log_activity('invoice_rectified', 'invoice', + f'#{original.invoice_number} -> rectificative draft #{draft.id}') + flash('Rectificative draft created. Edit it, then issue.', 'success') + return redirect(url_for('edit_invoice', id=draft.id)) + except Exception as e: + db.session.rollback() + logger.error('Rectify failed for invoice %s', original.id, exc_info=True) + flash(f'Could not create rectificative: {e}', 'danger') + return redirect(url_for('view_invoice', id=id)) + + @app.route('/create', methods=['GET', 'POST']) @login_required def create_invoice(): @@ -1011,7 +1219,9 @@ def create_invoice(): invoice_date_str = request.form['invoice_date'] due_date_str = request.form.get('due_date') - status = request.form['status'] + # F2 — new invoices start as DRAFT (freely editable) unless the form + # explicitly sets a status. Issuing is a separate, one-way action. + status = request.form.get('status') or 'draft' notes = request.form.get('notes', '') bank_id = request.form.get('bank_id') @@ -1276,9 +1486,9 @@ def create_invoice(): def edit_invoice(id): invoice = Invoice.query.get_or_404(id) - # Prevent editing paid invoices - if invoice.status == 'paid': - flash('Cannot edit invoice with status PAID. Please change status first if needed.', 'danger') + # F2 — issued/paid invoices are immutable; changes go via a rectificative. + if _invoice_locked(invoice): + flash('This invoice is issued and locked. Use "Rectify" to correct it.', 'danger') return redirect(url_for('view_invoice', id=id)) if request.method == 'POST': @@ -1288,7 +1498,11 @@ def edit_invoice(id): invoice_date_str = request.form['invoice_date'] due_date_str = request.form.get('due_date') - invoice.status = request.form['status'] + # F2 — issuing must go through the Issue action (assigns series, + # sequence and the fiscal snapshot). Ignore a direct 'issued' here. + _new_status = request.form['status'] + if _new_status != 'issued': + invoice.status = _new_status invoice.notes = request.form.get('notes', '') invoice.payment_method = request.form.get('payment_method', 'Bank Transfer') bank_id = request.form.get('bank_id') @@ -1468,9 +1682,9 @@ def delete_invoice(id): # and is not covered by CSRF protection. invoice = Invoice.query.get_or_404(id) - # Prevent deleting paid invoices - if invoice.status == 'paid': - flash('Cannot delete invoice with status PAID. Please change status first if needed.', 'danger') + # F2 — issued/paid invoices cannot be deleted; annul them instead. + if _invoice_locked(invoice): + flash('Issued invoices cannot be deleted. Use "Annul" to cancel it.', 'danger') return redirect(url_for('index')) # Delete PDF from storage @@ -2289,6 +2503,29 @@ def _apply_currency_provider(s, mgr): conn.execute(text("ALTER TABLE settings ADD COLUMN payment_methods TEXT DEFAULT 'Bank Transfer,PayPal,Credit Card,Cash,Crypto'")) conn.commit() + # F2 — invoice lifecycle: series/sequence, rectificative links, fiscal snapshot + inv_columns = [c['name'] for c in inspector.get_columns('invoice')] + for col, typedef in [('series', 'VARCHAR(20)'), + ('sequence_number', 'INTEGER'), + ('issued_at', 'DATETIME'), + ('rectifies_invoice_id', 'INTEGER'), + ('rectification_type', 'VARCHAR(20)'), + ('snap_vat_rate', 'FLOAT'), + ('snap_vat_amount', 'FLOAT'), + ('snap_taxable_base', 'FLOAT'), + ('snap_customer', 'TEXT')]: + if col not in inv_columns: + with db.engine.connect() as conn: + conn.execute(text(f'ALTER TABLE invoice ADD COLUMN {col} {typedef}')) + conn.commit() + + # F2-D4 — per-line VAT rate on invoice items + item_columns = [c['name'] for c in inspector.get_columns('invoice_item')] + if 'vat_rate' not in item_columns: + with db.engine.connect() as conn: + conn.execute(text('ALTER TABLE invoice_item ADD COLUMN vat_rate FLOAT')) + conn.commit() + # Initialize module system mgr = init_module_manager() diff --git a/module_manager.py b/module_manager.py index 9ba24ad..77bd301 100644 --- a/module_manager.py +++ b/module_manager.py @@ -349,6 +349,29 @@ def on_invoice_updated(self, invoice, request): """ pass + def on_invoice_issued(self, invoice, request): + """ + F2 — called when a draft invoice becomes ISSUED, before the transition + commits. Unlike the swallowed panel hooks, a raise here ABORTS the issue + (transactional) so compliance modules (e.g. verifactu) can veto. + """ + pass + + def on_invoice_rectified(self, new_invoice, original, request): + """ + F2 — called when a rectificative draft is created from an issued invoice, + before commit. A raise aborts the rectification. `original` is never + mutated. + """ + pass + + def on_invoice_annulled(self, invoice, request): + """ + F2 — called when an issued invoice is annulled (marked cancelled), before + commit. A raise aborts the annulment. + """ + pass + def get_invoice_templates(self): """ Return invoice PDF templates provided by this module. @@ -1421,8 +1444,13 @@ def get_by_number(self, invoice_number): return Invoice.query.filter_by(invoice_number=invoice_number).first() def is_locked(self, invoice): - """Check if invoice is locked (PAID status = read-only).""" - return invoice and invoice.status == 'paid' + """Check if invoice is locked (read-only). + + F2 — generalized from "PAID only" to "issued or later": both `issued` + and `paid` invoices are immutable; changes go through rectificatives. + Legacy `pending` and new `draft` invoices stay freely editable. + """ + return invoice is not None and invoice.status in ('issued', 'paid') def update(self, invoice_id, **fields): """Update invoice fields. Raises ValueError if invoice is PAID. @@ -2132,6 +2160,22 @@ def on_invoice_updated(self, invoice, request): logger.error("Module '%s' on_invoice_updated error: %s", _sanitize_log(mod.module_id), e) + def on_invoice_issued(self, invoice, request): + """F2 — fan out the ISSUE transition. Issue-blocking: a raising module + propagates so the caller can roll back the transition.""" + for mod in self.modules.values(): + mod.on_invoice_issued(invoice, request) + + def on_invoice_rectified(self, new_invoice, original, request): + """F2 — fan out rectification. Issue-blocking (propagates).""" + for mod in self.modules.values(): + mod.on_invoice_rectified(new_invoice, original, request) + + def on_invoice_annulled(self, invoice, request): + """F2 — fan out annulment. Issue-blocking (propagates).""" + for mod in self.modules.values(): + mod.on_invoice_annulled(invoice, request) + def get_invoice_templates(self): """Collect invoice PDF templates from core + all active modules. diff --git a/modules/expenses/index.py b/modules/expenses/index.py index afe7ec2..2c0d751 100644 --- a/modules/expenses/index.py +++ b/modules/expenses/index.py @@ -57,6 +57,12 @@ class Expense(db.Model): file_path = db.Column(db.String(500)) invoice_number = db.Column(db.String(100)) notes = db.Column(db.Text) + # F4 — must stay in sync with the core Expense model in app.py. + net_amount = db.Column(db.Float) + vat_rate = db.Column(db.Float) + vat_amount = db.Column(db.Float) + deductible = db.Column(db.Boolean, default=True) + deductible_pct = db.Column(db.Float, default=100.0) created_at = db.Column(db.DateTime, default=datetime.utcnow) class Contractor(db.Model): @@ -70,10 +76,109 @@ class Settings(db.Model): __table_args__ = {'extend_existing': True} id = db.Column(db.Integer, primary_key=True) + class ExpensesConfig(db.Model): + __tablename__ = 'expenses_config' + __table_args__ = {'extend_existing': True} + id = db.Column(db.Integer, primary_key=True) + key = db.Column(db.String(100), unique=True, nullable=False) + value = db.Column(db.Text) + self.Expense = Expense self.Contractor = Contractor self.Settings = Settings - return {} # No new tables to create — all exist in core + self.ExpensesConfig = ExpensesConfig + # expenses_config is module-owned -> let the manager create it. + return {'ExpensesConfig': ExpensesConfig} + + # F4 — seed category defaults (VAT rate % + deductible %). Shipped values; + # user override stored as JSON in expenses_config['category_defaults']. + SEED_CATEGORY_DEFAULTS = { + 'Office Supplies': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Software': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Equipment': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Services': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Professional Services': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Telecommunications': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Utilities': {'vat_rate': 21.0, 'deductible_pct': 100.0}, + 'Travel': {'vat_rate': 10.0, 'deductible_pct': 100.0}, + 'Insurance': {'vat_rate': 0.0, 'deductible_pct': 100.0}, # exempt + 'Social Security': {'vat_rate': 0.0, 'deductible_pct': 100.0}, + } + + def on_enable(self): + """F4 — add VAT/deductibility columns to the expense table (idempotent).""" + from sqlalchemy import inspect as sa_inspect, text + migrations = [ + ('net_amount', 'FLOAT'), + ('vat_rate', 'FLOAT'), + ('vat_amount', 'FLOAT'), + ('deductible', 'BOOLEAN'), + ('deductible_pct', 'FLOAT'), + ] + try: + inspector = sa_inspect(self._db.engine) + cols = [c['name'] for c in inspector.get_columns('expense')] + with self._db.engine.connect() as conn: + for name, coltype in migrations: + if name not in cols: + conn.execute(text( + f'ALTER TABLE expense ADD COLUMN {name} {coltype}')) + conn.commit() + except Exception as e: # pragma: no cover - defensive + self.logger.error('Expense VAT migration failed: %s', e) + + def get_category_defaults(self): + """Merged category defaults: seed values overlaid with user overrides.""" + import json + defaults = {k: dict(v) for k, v in self.SEED_CATEGORY_DEFAULTS.items()} + try: + row = self.ExpensesConfig.query.filter_by(key='category_defaults').first() + if row and row.value: + defaults.update(json.loads(row.value)) + except Exception as e: # pragma: no cover - defensive + self.logger.warning('Reading category defaults failed: %s', e) + return defaults + + # --- Settings panel (F4-D3): edit category VAT/deductible defaults ---- # + + @property + def settings_tab(self): + return {'id': 'expenses', 'label': 'Expenses'} + + def get_settings_html(self, settings): + import json + current = json.dumps(self.get_category_defaults(), indent=2, ensure_ascii=False) + return f''' +

Expense Category VAT Defaults

+

+ JSON map of category → default VAT rate (%) and deductible (%). Applied + when picking a category on the expense form (never overwrites a field + you already touched). +

+
+ +
+ ''' + + def save_settings(self, settings, form): + import json + if 'expense_category_defaults' not in form: + return # guard: unrelated tab save + raw = form.get('expense_category_defaults', '').strip() + try: + parsed = json.loads(raw) if raw else {} + if not isinstance(parsed, dict): + raise ValueError('must be a JSON object') + except (ValueError, TypeError) as e: + self.core.flash(f'Category defaults not saved (invalid JSON): {e}', 'danger') + return + row = self.ExpensesConfig.query.filter_by(key='category_defaults').first() + if not row: + row = self.ExpensesConfig(key='category_defaults') + self._db.session.add(row) + row.value = json.dumps(parsed, ensure_ascii=False) + self._db.session.commit() def register_routes(self, app): """Register expense routes""" @@ -156,6 +261,12 @@ def _api_serialize_expense(self, e): 'description': e.description, 'expense_date': e.expense_date.isoformat() if e.expense_date else None, 'invoice_number': e.invoice_number, + # F4 — VAT breakdown & deductibility. + 'net_amount': e.net_amount, + 'vat_rate': e.vat_rate, + 'vat_amount': e.vat_amount, + 'deductible': e.deductible, + 'deductible_pct': e.deductible_pct, } def _api_expenses(self, request): @@ -179,6 +290,14 @@ def _api_expenses(self, request): contractor_id = int(contractor_id) except (TypeError, ValueError): raise ApiError(400, 'bad_request', 'contractor_id must be an integer') + def _num(name): + v = body.get(name) + if v is None: + return None + try: + return float(v) + except (TypeError, ValueError): + raise ApiError(400, 'bad_request', f'{name} must be a number') exp = repo.create( amount=amount, currency=body.get('currency', 'EUR'), @@ -188,6 +307,12 @@ def _api_expenses(self, request): contractor_id=contractor_id, invoice_number=body.get('invoice_number'), notes=body.get('notes'), + # F4 fields (all optional). + net_amount=_num('net_amount'), + vat_rate=_num('vat_rate'), + vat_amount=_num('vat_amount'), + deductible=bool(body.get('deductible', True)), + deductible_pct=_num('deductible_pct') if body.get('deductible_pct') is not None else 100.0, ) self.core.log_activity('expense_created', 'expense', {'id': exp.id, 'amount': amount, @@ -248,6 +373,48 @@ def _list_expenses(self): contractors_map=contractors_map, categories=categories) + @staticmethod + def _parse_vat_fields(form, gross): + """Build the VAT/deductibility kwargs from the submitted form. + + Gross is authoritative. If net/vat not supplied but a rate is, split + gross by the rate (gross = net * (1 + rate)). Everything overridable. + Returns a dict of expense fields (values may be None = unknown). + """ + def _f(name): + raw = form.get(name) + if raw in (None, ''): + return None + try: + return float(raw) + except (TypeError, ValueError): + return None + + net = _f('net_amount') + vat_rate = _f('vat_rate') + vat_amount = _f('vat_amount') + + if gross is not None and vat_rate is not None and net is None and vat_amount is None: + # Derive net + VAT from gross + rate. + net = round(gross / (1 + vat_rate / 100.0), 2) if vat_rate else gross + vat_amount = round(gross - net, 2) + elif gross is not None and net is not None and vat_amount is None: + vat_amount = round(gross - net, 2) + + # Unchecked HTML checkboxes are omitted from the form, so absence = False. + deductible = form.get('deductible') not in (None, '', 'false', '0', 'off') + deductible_pct = _f('deductible_pct') + if deductible_pct is None: + deductible_pct = 100.0 + + return { + 'net_amount': net, + 'vat_rate': vat_rate, + 'vat_amount': vat_amount, + 'deductible': deductible, + 'deductible_pct': deductible_pct, + } + def _create_expense(self): """Create a new expense""" repo = self._get_repo() @@ -256,18 +423,21 @@ def _create_expense(self): if request.method == 'POST': try: file = request.files.get('file') + gross = float(request.form['amount']) + vat_fields = self._parse_vat_fields(request.form, gross) repo.create_with_file( app=self.core.app, file=file, storage=self.core, contractor_id=request.form.get('contractor_id') or None, - amount=float(request.form['amount']), + amount=gross, currency=request.form.get('currency', 'EUR'), category=request.form.get('category'), description=request.form.get('description'), expense_date=datetime.strptime(request.form['expense_date'], '%Y-%m-%d').date(), invoice_number=request.form.get('invoice_number'), - notes=request.form.get('notes') + notes=request.form.get('notes'), + **vat_fields ) flash('Expense created successfully!', 'success') return redirect(url_for('expenses.expenses_index')) @@ -279,7 +449,8 @@ def _create_expense(self): contractors = repo.get_all_contractors() tracked_currencies = settings_repo.get_tracked_currencies() return render_template('expense_form.html', expense=None, - contractors=contractors, tracked_currencies=tracked_currencies) + contractors=contractors, tracked_currencies=tracked_currencies, + category_defaults=self.get_category_defaults()) def _edit_expense(self, id): """Edit an existing expense""" @@ -290,19 +461,22 @@ def _edit_expense(self, id): if request.method == 'POST': try: file = request.files.get('file') + gross = float(request.form['amount']) + vat_fields = self._parse_vat_fields(request.form, gross) repo.update_with_file( app=self.core.app, expense=expense, file=file, storage=self.core, contractor_id=request.form.get('contractor_id') or None, - amount=float(request.form['amount']), + amount=gross, currency=request.form.get('currency', 'EUR'), category=request.form.get('category'), description=request.form.get('description'), expense_date=datetime.strptime(request.form['expense_date'], '%Y-%m-%d').date(), invoice_number=request.form.get('invoice_number'), - notes=request.form.get('notes') + notes=request.form.get('notes'), + **vat_fields ) flash('Expense updated successfully!', 'success') return redirect(url_for('expenses.expenses_index')) @@ -314,7 +488,8 @@ def _edit_expense(self, id): contractors = repo.get_all_contractors() tracked_currencies = settings_repo.get_tracked_currencies() return render_template('expense_form.html', expense=expense, - contractors=contractors, tracked_currencies=tracked_currencies) + contractors=contractors, tracked_currencies=tracked_currencies, + category_defaults=self.get_category_defaults()) def _delete_expense(self, id): """Delete an expense""" @@ -420,13 +595,20 @@ def _get_expenses_for_report(self, start_date, end_date): for expense in expenses: amount_eur = self._convert_to(expense.amount, expense.currency, 'EUR', expense.expense_date) + vat_eur = None + if expense.vat_amount is not None: + vat_eur = self._convert_to(expense.vat_amount, expense.currency, 'EUR', + expense.expense_date) result.append({ 'expense_date': expense.expense_date.strftime('%d/%m/%Y'), 'invoice_number': expense.invoice_number or '', 'contractor_name': contractors_map.get(expense.contractor_id, 'N/A'), 'category': expense.category or 'N/A', 'description': expense.description or '', - 'amount_eur': amount_eur + 'amount_eur': amount_eur, + # F4 — VAT split (None on legacy rows). + 'vat_eur': vat_eur, + 'deductible_pct': expense.deductible_pct, }) return result @@ -441,19 +623,42 @@ def get_tax_obligations(self, context): self._db.extract('year', self.Expense.expense_date) == current_year ).all() - # Convert expenses to base currency + default_vat_rate = ((context.get('settings').default_vat_rate or 21.0) / 100.0 + if context.get('settings') and hasattr(context['settings'], 'default_vat_rate') + else 0.21) + + # Convert expenses to base currency. VAT paid (IVA soportado) now uses the + # real per-expense vat_amount × deductible_pct when F4 data is present; + # legacy rows (no vat_amount) fall back to the derived estimate and are + # counted so the user knows the deducible figure is approximate. total_expenses = 0 vat_paid = 0 + missing_vat_count = 0 for expense in expenses_query: amount_base = self._convert_to(expense.amount, expense.currency, base_currency, expense.expense_date) total_expenses += amount_base - vat_rate = (context.get('settings').default_vat_rate or 21.0) / 100.0 if context.get('settings') and hasattr(context['settings'], 'default_vat_rate') else 0.21 - vat_paid += amount_base * vat_rate + if expense.vat_amount is not None: + if getattr(expense, 'deductible', True): + vat_base = self._convert_to(expense.vat_amount, expense.currency, + base_currency, expense.expense_date) + pct = (expense.deductible_pct if expense.deductible_pct is not None else 100.0) + vat_paid += vat_base * (pct / 100.0) + else: + missing_vat_count += 1 + vat_paid += amount_base * default_vat_rate # legacy estimate vat_collected = context.get('vat_collected', 0) vat_to_pay = vat_collected - vat_paid + notes = [ + f"VAT: Collected {context['currency_symbol']}{vat_collected:.2f} - Paid {context['currency_symbol']}{vat_paid:.2f}" + ] + if missing_vat_count: + notes.append( + f"{missing_vat_count} expense(s) lack VAT data — IVA soportado " + f"estimated for those.") + return { 'summary_columns': [ {'label': 'Expenses', 'value': total_expenses} @@ -461,9 +666,7 @@ def get_tax_obligations(self, context): 'breakdown_rows': [ {'label': 'VAT to Pay (IVA)', 'amount': vat_to_pay} ], - 'notes': [ - f"VAT: Collected {context['currency_symbol']}{vat_collected:.2f} - Paid {context['currency_symbol']}{vat_paid:.2f}" - ], + 'notes': notes, 'deductions': total_expenses, 'tax_total': vat_to_pay } diff --git a/modules/expenses/templates/expense_form.html b/modules/expenses/templates/expense_form.html index 71badd5..f443737 100644 --- a/modules/expenses/templates/expense_form.html +++ b/modules/expenses/templates/expense_form.html @@ -87,6 +87,42 @@

{{ 'Edit' if expense else 'Create' }} Expense

+ +
+ VAT breakdown (IVA soportado) +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ +
+
@@ -112,6 +148,8 @@

{{ 'Edit' if expense else 'Create' }} Expense

+{% endblock %} diff --git a/modules/tax_es_forms/templates/tax_forms_draft_303.html b/modules/tax_es_forms/templates/tax_forms_draft_303.html new file mode 100644 index 0000000..67e92a8 --- /dev/null +++ b/modules/tax_es_forms/templates/tax_forms_draft_303.html @@ -0,0 +1,77 @@ +{% extends "base.html" %} + +{% block title %}Modelo 303 · Q{{ draft.quarter }} {{ draft.year }}{% endblock %} + +{% block content %} +
+

Modelo 303 (IVA) — Q{{ draft.quarter }} {{ draft.year }}

+
+ ← Back + +
+
+ + +
+ Estimate, not tax advice. Devengo (accrual) basis; régimen + general only. Verify every box against the current official AEAT Modelo 303 + before filing. Box table version: {{ draft.table_version }}. +
+ +{% if draft.meta.missing_expense_vat_count %} +
+ ⚠ {{ draft.meta.missing_expense_vat_count }} expense(s) this quarter lack VAT + breakdown — deductible IVA (boxes 28/29/45) understates your real deductions. +
+{% endif %} + + + + + + + {% for box in ['01', '03', '27', '28', '29', '45', '46', '71'] %} + + + + + + {% endfor %} + +
BoxConceptAmount (EUR)
{{ box }}{{ draft.labels[box] }} + {{ "%.2f"|format(draft.boxes[box]) }} + +
+ +

+ Box 71 (resultado de la liquidación) = box 27 (IVA devengado) − box 45 (IVA + deducible). A positive figure is payable to AEAT; a negative figure is to + compensate or refund. +

+ + + + +{% endblock %} diff --git a/modules/tax_es_forms/templates/tax_forms_draft_index.html b/modules/tax_es_forms/templates/tax_forms_draft_index.html new file mode 100644 index 0000000..e5184ae --- /dev/null +++ b/modules/tax_es_forms/templates/tax_forms_draft_index.html @@ -0,0 +1,51 @@ +{% extends "base.html" %} + +{% block title %}Tax Drafts{% endblock %} + +{% block content %} +
+

Modelo 303 / 130 Drafts

+ Box table version: {{ table_version }} +
+ +
+ Estimate, not tax advice. These drafts add up the data already + in the app so you can transcribe the boxes into AEAT's portal. Always check the + figures and box numbers against the current official model before filing. + Régimen general / estimación directa only — módulos, recargo de equivalencia, + prorrata and criterio de caja are out of scope. +
+ + + + + + + + + + + + + {% for year in years %} + + + {% for q in [1, 2, 3, 4] %} + + {% endfor %} + + {% endfor %} + +
YearQ1Q2Q3Q4
{{ year }} + 303 + 130 +
+ + +{% endblock %} diff --git a/templates/create.html b/templates/create.html index a5244f1..7e1f2b1 100644 --- a/templates/create.html +++ b/templates/create.html @@ -29,10 +29,14 @@

Invoice Details + + Draft invoices stay editable. Use Issue on the invoice to make it permanent and get its sequential number. +

diff --git a/templates/edit.html b/templates/edit.html index 679858a..2bcf253 100644 --- a/templates/edit.html +++ b/templates/edit.html @@ -29,7 +29,8 @@

Invoice Details diff --git a/templates/view.html b/templates/view.html index 0be15f5..ff6ec0a 100644 --- a/templates/view.html +++ b/templates/view.html @@ -144,6 +144,24 @@

Service DetailsBack to List Preview PDF Download PDF + + {# F2 — lifecycle actions #} + {% if invoice.status in ('draft', 'pending') %} +
+ +
+ {% elif invoice.status in ('issued', 'paid') %} +
+ + +
+
+ +
+ {% endif %} + {% if module_manager %} {% for action_html in module_manager.get_invoice_actions(invoice) %} {{ action_html|safe }} diff --git a/tests/conftest.py b/tests/conftest.py index 1a4cb34..e702fc6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,6 +11,7 @@ import pytest from app import app as flask_app, db as _db # noqa: E402 +from module_manager import ModuleManager # noqa: E402 @pytest.fixture @@ -27,3 +28,42 @@ def app(): @pytest.fixture def client(app): return app.test_client() + + +@pytest.fixture(scope='session') +def loaded_modules(): + """One shared ModuleManager for the whole session. + + Building more than one ModuleManager redefines the singleton + `module_enabled` table in the shared metadata and errors, so every test that + needs loaded modules goes through this single instance. Enables the modules + the suite exercises (expenses, tax_es_forms). + """ + import sys + from app import Settings + appmod = sys.modules['app'] + + ctx = flask_app.app_context() + ctx.push() + _db.create_all() + + mm = ModuleManager(flask_app, _db) + mm.core._settings_model = Settings + mm.init_db() + mm.discover_modules() + + enabled_model = mm._get_module_enabled_model() + for module_id in ('expenses', 'tax_es_forms', 'fiscal_calendar'): + if not enabled_model.query.filter_by(module_id=module_id).first(): + _db.session.add(enabled_model(module_id=module_id, enabled=True)) + _db.session.commit() + mm.load_enabled_modules() + + appmod.module_manager = mm + flask_app.jinja_env.globals['module_manager'] = mm + flask_app.jinja_env.globals.setdefault('app_version', 'test') + + yield mm + + _db.session.remove() + ctx.pop() diff --git a/tests/test_expense_vat.py b/tests/test_expense_vat.py new file mode 100644 index 0000000..4b8f4cc --- /dev/null +++ b/tests/test_expense_vat.py @@ -0,0 +1,85 @@ +"""F4 — Expense VAT breakdown & deductibility tests.""" +from datetime import date + +import pytest + +from modules.expenses.index import ExpensesModule + + +@pytest.fixture +def expenses_module(loaded_modules): + return loaded_modules.modules['expenses'] + + +# --- F4-D1: both Expense declarations stay in sync ------------------------- + +def test_expense_model_column_parity(): + """Core app.py Expense and the module mirror must expose the same columns.""" + from app import Expense as CoreExpense + from sqlalchemy import inspect as sa_inspect + + core_cols = {c.key for c in sa_inspect(CoreExpense).columns} + for field in ('net_amount', 'vat_rate', 'vat_amount', 'deductible', 'deductible_pct'): + assert field in core_cols, f'core Expense missing {field}' + + +def test_module_expense_has_vat_columns(expenses_module): + from sqlalchemy import inspect as sa_inspect + mod_cols = {c.key for c in sa_inspect(expenses_module.Expense).columns} + from app import Expense as CoreExpense + core_cols = {c.key for c in sa_inspect(CoreExpense).columns} + # Parity for the F4 fields across both declaration sites. + for field in ('net_amount', 'vat_rate', 'vat_amount', 'deductible', 'deductible_pct'): + assert field in mod_cols, f'module Expense missing {field}' + assert field in core_cols + + +# --- F4-D2: gross+rate splitting ------------------------------------------- + +def test_parse_vat_from_gross_and_rate(): + fields = ExpensesModule._parse_vat_fields({'vat_rate': '21'}, gross=121.0) + assert fields['net_amount'] == 100.0 + assert fields['vat_amount'] == 21.0 + assert fields['vat_rate'] == 21.0 + assert fields['deductible'] is False # checkbox absent -> unchecked + assert fields['deductible_pct'] == 100.0 + + +def test_parse_vat_explicit_net_overrides_split(): + fields = ExpensesModule._parse_vat_fields( + {'vat_rate': '10', 'net_amount': '90', 'deductible': 'on', + 'deductible_pct': '50'}, gross=100.0) + assert fields['net_amount'] == 90.0 + assert fields['vat_amount'] == 10.0 # gross - net + assert fields['deductible'] is True + assert fields['deductible_pct'] == 50.0 + + +def test_parse_vat_no_rate_leaves_unknown(): + fields = ExpensesModule._parse_vat_fields({}, gross=50.0) + assert fields['net_amount'] is None + assert fields['vat_amount'] is None + + +# --- F4-D4: tax obligations use real VAT, count missing -------------------- + +def test_tax_obligations_uses_real_vat_and_counts_missing(expenses_module): + from app import db + + em = expenses_module + # One expense with VAT data, one legacy (no VAT). + db.session.add(em.Expense(amount=121.0, currency='EUR', category='Software', + expense_date=date(2026, 3, 1), net_amount=100.0, + vat_amount=21.0, vat_rate=21.0, deductible=True, + deductible_pct=100.0)) + db.session.add(em.Expense(amount=50.0, currency='EUR', category='Legacy', + expense_date=date(2026, 3, 2))) + db.session.commit() + + settings = db.session.query(em.Settings).first() + ctx = {'current_year': 2026, 'base_currency': 'EUR', 'currency_symbol': '€', + 'settings': settings, 'vat_collected': 0} + result = em.get_tax_obligations(ctx) + # vat_paid = 21 (real) + 50*0.21 (legacy estimate) = 31.5 -> vat_to_pay = -31.5 + assert round(result['tax_total'], 2) == -31.5 + assert any('lack VAT data' in n for n in result['notes']) diff --git a/tests/test_expenses_list.py b/tests/test_expenses_list.py index 6f71d3b..75e4b2b 100644 --- a/tests/test_expenses_list.py +++ b/tests/test_expenses_list.py @@ -10,42 +10,10 @@ import pytest -from module_manager import ModuleManager - -@pytest.fixture(scope='module') -def expenses_module(): - """Load the Expenses module once against the app + a fresh DB. - - Module-scoped so the ModuleManager (which registers the singleton - `module_enabled` table) is built only once — re-building it per test would - redefine that table in the shared metadata and error. - """ - import app as appmod - from app import app as flask_app, db, Settings - - with flask_app.app_context(): - db.create_all() - - mm = ModuleManager(flask_app, db) - mm.core._settings_model = Settings - mm.init_db() - mm.discover_modules() - - enabled_model = mm._get_module_enabled_model() - db.session.add(enabled_model(module_id='expenses', enabled=True)) - db.session.commit() - mm.load_enabled_modules() - - # Templates reference these Jinja globals. - appmod.module_manager = mm - flask_app.jinja_env.globals.setdefault('module_manager', mm) - flask_app.jinja_env.globals.setdefault('app_version', 'test') - - yield mm.modules['expenses'] - - db.session.remove() - db.drop_all() +@pytest.fixture +def expenses_module(loaded_modules): + return loaded_modules.modules['expenses'] def test_list_view_renders_contractor_name(expenses_module): diff --git a/tests/test_fiscal_calendar.py b/tests/test_fiscal_calendar.py new file mode 100644 index 0000000..21c59ed --- /dev/null +++ b/tests/test_fiscal_calendar.py @@ -0,0 +1,85 @@ +"""F3 — Fiscal calendar dataset + reminder logic tests.""" +from datetime import date + +import pytest + +from modules.fiscal_calendar.calendar_data import ( + all_deadlines, + upcoming, + available_forms, +) + + +def test_dataset_has_quarterly_and_annual_forms(): + forms = available_forms() + for f in ('303', '130', '349', '390', '100'): + assert f in forms + + +def test_q4_window_rolls_into_next_january(): + # Modelo 303 Q4 2026 files 1-30 January 2027. + q4 = [d for d in all_deadlines() + if d['form'] == '303' and d['quarter'] == 4 and d['year'] == 2026] + assert len(q4) == 1 + assert q4[0]['window_start'] == date(2027, 1, 1) + assert q4[0]['window_end'] == date(2027, 1, 30) + + +def test_upcoming_marks_window_open(): + # 10 April 2026 -> Q1 filing window (1-20 Apr) is OPEN. + items = upcoming(30, today=date(2026, 4, 10), selected_forms={'303'}) + q1 = [d for d in items if d['quarter'] == 1 and d['year'] == 2026] + assert q1 and q1[0]['state'] == 'open' + assert q1[0]['days_to_end'] == 10 + + +def test_upcoming_filters_by_selected_forms(): + items = upcoming(400, today=date(2026, 1, 1), selected_forms={'130'}) + assert items + assert all(d['form'] == '130' for d in items) + + +def test_upcoming_excludes_closed_windows(): + # After Q1 2026 window closed (25 Apr), Q1 must not appear. + items = upcoming(5, today=date(2026, 4, 25), selected_forms={'303'}) + assert not any(d['quarter'] == 1 and d['year'] == 2026 for d in items) + + +def test_upcoming_respects_lookahead_horizon(): + # 1 Jan 2026: Q1 window opens 1 Apr (~90 days). A 30-day horizon hides it. + near = upcoming(30, today=date(2026, 1, 1), selected_forms={'303'}) + assert not any(d['quarter'] == 1 and d['year'] == 2026 for d in near) + far = upcoming(120, today=date(2026, 1, 1), selected_forms={'303'}) + assert any(d['quarter'] == 1 and d['year'] == 2026 for d in far) + + +# --- reminder job (DB-backed) ---------------------------------------------- + +@pytest.fixture +def fiscal_module(loaded_modules): + import json + from app import db + fc = loaded_modules.modules['fiscal_calendar'] + # Pin selected forms so the test is deterministic. + row = fc.Config.query.filter_by(key='selected_forms').first() + if not row: + row = fc.Config(key='selected_forms') + db.session.add(row) + row.value = json.dumps(['303']) + # Reset any prior reminder state. + state = fc.Config.query.filter_by(key='reminder_state').first() + if state: + state.value = '{}' + db.session.commit() + return fc + + +def test_reminders_fire_once_and_are_idempotent(fiscal_module): + fc = fiscal_module + # 6 days before Q1 2026 303 window close (20 Apr) -> the T-7 threshold on 13 Apr. + day = date(2026, 4, 13) + fired = fc._run_reminders(today=day) + assert any(k.startswith('303:Q1 2026:7') for k in fired) + # Second run same day -> nothing new (idempotent, survives restart via config). + assert fc._run_reminders(today=day) == [] + diff --git a/tests/test_invoice_lifecycle.py b/tests/test_invoice_lifecycle.py new file mode 100644 index 0000000..00441b0 --- /dev/null +++ b/tests/test_invoice_lifecycle.py @@ -0,0 +1,190 @@ +"""F2 — Invoice lifecycle hardening tests: issue, sequence, snapshot, lock, +rectificative, annul, and issue-blocking hooks.""" +import json +from datetime import date +from types import SimpleNamespace + +import pytest + + +def _draft(db, Invoice, Customer, amount_eur=1000.0, tax_type='standard', + invoice_date=date(2026, 2, 1), number='DRAFT-1'): + customer = Customer(name='ACME', tax_type=tax_type) + db.session.add(customer) + db.session.flush() + inv = Invoice(invoice_number=number, client_name='ACME', amount_usd=0.0, + amount_eur=amount_eur, exchange_rate=1.0, invoice_date=invoice_date, + status='draft', currency='EUR', customer_id=customer.id) + db.session.add(inv) + db.session.flush() + return inv + + +def _auth(client): + with client.session_transaction() as sess: + sess['authenticated'] = True + + +# --- F2-D2 / F2-D4: issue assigns sequence + freezes snapshot --------------- + +def test_issue_assigns_series_sequence_and_snapshot(app): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + + assert inv.status == 'issued' + assert inv.series == '2026' + assert inv.sequence_number == 1 + assert inv.invoice_number == '2026/0001' + assert inv.issued_at is not None + # Snapshot frozen: standard customer -> 21% of 1000. + assert inv.snap_vat_rate == 21.0 + assert inv.snap_taxable_base == 1000.0 + assert inv.snap_vat_amount == 210.0 + snap = json.loads(inv.snap_customer) + assert snap['tax_type'] == 'standard' + + +def test_snapshot_survives_customer_change(app): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + # Change the customer AFTER issue — snapshot must not move. + inv.customer.tax_type = 'non_eu' + db.session.commit() + assert inv.snap_vat_rate == 21.0 + assert inv.snap_vat_amount == 210.0 + + +def test_sequence_increments_per_series(app): + from app import db, Invoice, Customer, _issue_invoice + + a = _draft(db, Invoice, Customer, number='DRAFT-A') + _issue_invoice(a, request_obj=None) + b = _draft(db, Invoice, Customer, number='DRAFT-B') + _issue_invoice(b, request_obj=None) + assert a.invoice_number == '2026/0001' + assert b.invoice_number == '2026/0002' + + +def test_non_standard_customer_has_zero_vat_snapshot(app): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer, tax_type='eu_b2b') + _issue_invoice(inv, request_obj=None) + assert inv.snap_vat_rate == 0.0 + assert inv.snap_vat_amount == 0.0 + + +# --- F2-D1: lock enforcement ----------------------------------------------- + +def test_locked_helper(): + from app import _invoice_locked + assert _invoice_locked(SimpleNamespace(status='issued')) is True + assert _invoice_locked(SimpleNamespace(status='paid')) is True + assert _invoice_locked(SimpleNamespace(status='draft')) is False + assert _invoice_locked(SimpleNamespace(status='pending')) is False + assert _invoice_locked(None) is False + + +def test_edit_route_blocks_issued_invoice(app, client): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + db.session.commit() + _auth(client) + resp = client.get(f'/edit/{inv.id}', follow_redirects=False) + assert resp.status_code == 302 + assert f'/view/{inv.id}' in resp.headers['Location'] + + +def test_delete_route_blocks_issued_invoice(app, client): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + db.session.commit() + _auth(client) + resp = client.post(f'/delete/{inv.id}', follow_redirects=False) + assert resp.status_code == 302 + # Still present. + assert db.session.get(Invoice, inv.id) is not None + + +def test_invoice_service_locks_issued(app): + from app import db, Invoice, Customer, _issue_invoice + from module_manager import InvoiceService + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + svc = InvoiceService(SimpleNamespace(db=db)) + assert svc.is_locked(inv) is True + + +# --- F2-D3: rectify + annul ------------------------------------------------- + +def test_rectify_creates_linked_draft_without_mutating_original(app, client): + from app import db, Invoice, InvoiceItem, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + db.session.add(InvoiceItem(invoice_id=inv.id, description='Work', quantity=1, + unit_price_usd=1000, subtotal_usd=1000)) + db.session.commit() + _issue_invoice(inv, request_obj=None) + db.session.commit() + original_number = inv.invoice_number + + _auth(client) + resp = client.post(f'/rectify/{inv.id}', data={'rectification_type': 'sustitucion'}, + follow_redirects=False) + assert resp.status_code == 302 + draft = Invoice.query.filter_by(rectifies_invoice_id=inv.id).first() + assert draft is not None + assert draft.status == 'draft' + assert draft.rectification_type == 'sustitucion' + assert len(draft.items) == 1 + # Original untouched. + assert inv.invoice_number == original_number + assert inv.status == 'issued' + + +def test_annul_marks_cancelled_and_retains(app, client): + from app import db, Invoice, Customer, _issue_invoice + + inv = _draft(db, Invoice, Customer) + _issue_invoice(inv, request_obj=None) + db.session.commit() + _auth(client) + resp = client.post(f'/annul/{inv.id}', follow_redirects=False) + assert resp.status_code == 302 + refreshed = db.session.get(Invoice, inv.id) + assert refreshed is not None + assert refreshed.status == 'cancelled' + + +# --- F2-D5: issue-blocking hooks ------------------------------------------- + +def test_issue_hook_veto_aborts(app, monkeypatch): + import sys + from app import db, Invoice, Customer, _issue_invoice + appmod = sys.modules['app'] + + class Vetoer: + def on_invoice_issued(self, invoice, request): + raise ValueError('compliance veto') + + class FakeManager: + def on_invoice_issued(self, invoice, request): + Vetoer().on_invoice_issued(invoice, request) + + monkeypatch.setattr(appmod, 'module_manager', FakeManager()) + inv = _draft(db, Invoice, Customer) + with pytest.raises(ValueError): + _issue_invoice(inv, request_obj=None) + db.session.rollback() + # Transition aborted — not left issued after rollback. + refreshed = db.session.get(Invoice, inv.id) + assert refreshed is None or refreshed.status != 'issued' diff --git a/tests/test_tax_es_forms.py b/tests/test_tax_es_forms.py new file mode 100644 index 0000000..9039876 --- /dev/null +++ b/tests/test_tax_es_forms.py @@ -0,0 +1,167 @@ +"""Unit tests for the tax_es_forms Modelo 303/130 aggregation engine. + +Pure-math tests: feed plain objects, assert box numbers. No DB, no Flask. +""" +from datetime import date +from types import SimpleNamespace + +from modules.tax_es_forms.calculator import ( + compute_modelo_303, + compute_modelo_130, + quarter_of, + quarter_bounds, +) + + +def _inv(amount_eur, tax_type='standard', status='pending', d=date(2026, 2, 1), + snap_vat_rate=None, snap_vat_amount=None, snap_taxable_base=None): + customer = SimpleNamespace(tax_type=tax_type) if tax_type else None + return SimpleNamespace(amount_eur=amount_eur, customer=customer, + status=status, invoice_date=d, + snap_vat_rate=snap_vat_rate, snap_vat_amount=snap_vat_amount, + snap_taxable_base=snap_taxable_base) + + +def _exp(amount, currency='EUR', d=date(2026, 2, 1), **extra): + return SimpleNamespace(amount=amount, currency=currency, expense_date=d, **extra) + + +# --- quarter helpers -------------------------------------------------------- + +def test_quarter_of_boundaries(): + assert quarter_of(date(2026, 1, 1)) == 1 + assert quarter_of(date(2026, 3, 31)) == 1 + assert quarter_of(date(2026, 4, 1)) == 2 + assert quarter_of(date(2026, 12, 31)) == 4 + + +def test_quarter_bounds_q1_and_q4(): + assert quarter_bounds(2026, 1) == (date(2026, 1, 1), date(2026, 3, 31)) + assert quarter_bounds(2026, 4) == (date(2026, 10, 1), date(2026, 12, 31)) + + +# --- Modelo 303 ------------------------------------------------------------- + +def test_303_output_vat_only_on_standard_customers(): + invoices = [ + _inv(1000.0, tax_type='standard'), + _inv(500.0, tax_type='eu_b2b'), # reverse charge -> no output VAT + _inv(300.0, tax_type='non_eu'), # export -> no output VAT + ] + result = compute_modelo_303(invoices, [], vat_rate=0.21) + assert result['boxes']['01'] == 1000.0 # base only from standard + assert result['boxes']['03'] == 210.0 # 21% of 1000 + assert result['boxes']['27'] == 210.0 + # No deductible expenses -> result equals output VAT. + assert result['boxes']['71'] == 210.0 + + +def test_303_excludes_cancelled_invoices(): + invoices = [ + _inv(1000.0, tax_type='standard', status='pending'), + _inv(1000.0, tax_type='standard', status='cancelled'), + ] + result = compute_modelo_303(invoices, [], vat_rate=0.21) + assert result['boxes']['01'] == 1000.0 + + +def test_303_counts_expenses_missing_vat_data(): + expenses = [_exp(121.0), _exp(50.0)] # no vat_amount attr -> missing + result = compute_modelo_303([], expenses, vat_rate=0.21) + assert result['boxes']['29'] == 0.0 + assert result['meta']['missing_expense_vat_count'] == 2 + + +def test_303_deducts_expense_vat_when_present(): + # F4-shaped expense: carries explicit VAT fields. + expenses = [ + _exp(121.0, net_amount=100.0, vat_amount=21.0, deductible=True, + deductible_pct=100.0), + _exp(110.0, net_amount=100.0, vat_amount=10.0, deductible=True, + deductible_pct=50.0), # 50% deductible (e.g. vehicle) + ] + result = compute_modelo_303([], expenses, vat_rate=0.21) + assert result['boxes']['28'] == 150.0 # 100 + 100*0.5 + assert result['boxes']['29'] == 26.0 # 21 + 10*0.5 + assert result['boxes']['45'] == 26.0 + assert result['meta']['missing_expense_vat_count'] == 0 + + +def test_303_uses_fiscal_snapshot_when_present(): + # Issued invoice: snapshot wins over the live customer tax_type. + invoices = [_inv(1000.0, tax_type='non_eu', status='issued', + snap_vat_rate=21.0, snap_taxable_base=1000.0, snap_vat_amount=210.0)] + result = compute_modelo_303(invoices, [], vat_rate=0.10) # live rate ignored + assert result['boxes']['01'] == 1000.0 + assert result['boxes']['03'] == 210.0 # from snapshot, not 0.10*1000 + + +def test_303_snapshot_zero_rate_excludes_from_base(): + # Issued eu_b2b invoice: snapshot rate 0 -> not in the general base. + invoices = [_inv(500.0, tax_type='eu_b2b', status='issued', + snap_vat_rate=0.0, snap_taxable_base=500.0, snap_vat_amount=0.0)] + result = compute_modelo_303(invoices, [], vat_rate=0.21) + assert result['boxes']['01'] == 0.0 + assert result['boxes']['03'] == 0.0 + + +def test_130_prefers_net_amount_over_gross(): + invoices = [_inv(1000.0, tax_type='standard')] + # F4 expense: gross 121, net 100 -> IRPF gastos should use net (100). + expenses = [_exp(121.0, net_amount=100.0, vat_amount=21.0)] + result = compute_modelo_130(invoices, expenses, irpf_rate=0.20) + assert result['boxes']['02'] == 100.0 + + +def test_303_result_is_devengado_minus_deducible(): + invoices = [_inv(1000.0, tax_type='standard')] + expenses = [_exp(121.0, net_amount=100.0, vat_amount=21.0)] + result = compute_modelo_303(invoices, expenses, vat_rate=0.21) + assert result['boxes']['27'] == 210.0 + assert result['boxes']['45'] == 21.0 + assert result['boxes']['71'] == 189.0 # 210 - 21 + + +# --- Modelo 130 ------------------------------------------------------------- + +def test_130_basic_quarter_no_prior_payments(): + invoices = [_inv(3000.0, tax_type='standard')] + expenses = [_exp(1000.0)] + result = compute_modelo_130(invoices, expenses, irpf_rate=0.20) + assert result['boxes']['01'] == 3000.0 + assert result['boxes']['02'] == 1000.0 + assert result['boxes']['03'] == 2000.0 + assert result['boxes']['04'] == 400.0 # 20% of 2000 + assert result['boxes']['05'] == 0.0 + assert result['boxes']['07'] == 400.0 + + +def test_130_subtracts_prior_quarter_payments(): + # YTD to Q2: 6000 income, 2000 expenses -> rendimiento 4000, pago 800. + # Prior (Q1): 3000 income, 1000 expenses -> prior pago 400. + invoices = [_inv(6000.0, tax_type='standard')] + expenses = [_exp(2000.0)] + result = compute_modelo_130(invoices, expenses, irpf_rate=0.20, + prior_income=3000.0, prior_expenses=1000.0) + assert result['boxes']['04'] == 800.0 + assert result['boxes']['05'] == 400.0 + assert result['boxes']['07'] == 400.0 # 800 - 400 + + +def test_130_never_negative(): + invoices = [_inv(500.0, tax_type='standard')] + expenses = [_exp(2000.0)] # loss-making quarter + result = compute_modelo_130(invoices, expenses, irpf_rate=0.20) + assert result['boxes']['03'] == -1500.0 # rendimiento can be negative + assert result['boxes']['04'] == 0.0 # but pago floors at 0 + assert result['boxes']['07'] == 0.0 + + +def test_130_converts_expense_currency(): + invoices = [_inv(1000.0, tax_type='standard')] + # 100 USD expense, convert fn doubles it to EUR. + expenses = [_exp(100.0, currency='USD')] + convert = lambda amount, currency, when: amount * 2 if currency == 'USD' else amount + result = compute_modelo_130(invoices, expenses, irpf_rate=0.20, + convert_expense=convert) + assert result['boxes']['02'] == 200.0