179 lines
8.1 KiB
Python
179 lines
8.1 KiB
Python
"""Regression coverage for Marketplace period amounts; no external writes."""
|
||
from copy import deepcopy
|
||
from decimal import Decimal
|
||
import json
|
||
|
||
import pytest
|
||
from fastapi import HTTPException
|
||
|
||
from app.modules.also.backend.billing import period_amount_line, check_period_amounts, _period_text
|
||
from app.modules.orders.backend.router import OrdreLineInput
|
||
from app.products.backend.economic_documents import line_snapshot
|
||
|
||
|
||
CASES = [
|
||
(116, '67.08', '7028.25', '01.08.2026 - 29.08.2026', '28 af 31'),
|
||
(116, '67.08', '753.03', '29.08.2026 - 01.09.2026', '3 af 31'),
|
||
(108, '67.08', '6543.55', '01.08.2026 - 29.08.2026', '28 af 31'),
|
||
(108, '67.08', '701.09', '29.08.2026 - 01.09.2026', '3 af 31'),
|
||
(1, '27.90', '17.10', '13.08.2026 - 01.09.2026', '19 af 31'),
|
||
(1, '50.28', '19.46', '20.08.2026 - 01.09.2026', '12 af 31'),
|
||
]
|
||
|
||
|
||
def source(index=0):
|
||
quantity, price, amount, interval, _ = CASES[index]
|
||
return dict(line_key=f'also:{index+1}', source_type='also_cloud', source_id=index+1,
|
||
product_id=12, description='Microsoft abonnement\nPeriode: ' + interval,
|
||
quantity=quantity, unit_price=price, amount=amount,
|
||
discount_percentage=0, currency='DKK', meta={'billing_period': interval})
|
||
|
||
|
||
def test_norva_period_amounts_survive_request_model_and_catalog_snapshot():
|
||
total = Decimal(0)
|
||
product = dict(id=12, name='Microsoft abonnement', economic_product_number='MS-365',
|
||
economic_connection_id=1, economic_unit_number=1, status='active')
|
||
for index, (_, _, amount, _, days) in enumerate(CASES):
|
||
old = source(index)
|
||
before = deepcopy(old)
|
||
line = period_amount_line(old)
|
||
assert old == before
|
||
assert period_amount_line(line) == line
|
||
assert days + ' dage.' in line['description']
|
||
assert 'Reference:' not in line['description']
|
||
request_line = OrdreLineInput(**line).model_dump()
|
||
check_period_amounts([request_line], [old])
|
||
snapshot = line_snapshot(request_line, product, 1, 'DKK')
|
||
assert snapshot['quantity'] == '1.00'
|
||
assert snapshot['unit_price'] == amount
|
||
assert snapshot['net_total'] == amount
|
||
total += Decimal(snapshot['net_total'])
|
||
assert total == Decimal('15062.48')
|
||
|
||
|
||
def test_exact_customer_text_and_no_reference():
|
||
old = source()
|
||
old['description'] += '\nReference: sag-xx11.'
|
||
line = period_amount_line(old)
|
||
assert line['description'] == ('Microsoft abonnement\n116 licenser á 67,08 kr./måned. '
|
||
'Periode: 01.08.–28.08.2026, 28 af 31 dage.')
|
||
|
||
|
||
@pytest.mark.parametrize('interval,expected', [
|
||
('01.02.2024 - 01.03.2024', '29 af 29 dage.'),
|
||
('01.02.2026 - 01.03.2026', '28 af 28 dage.'),
|
||
('31.12.2026 - 02.01.2027', '31.12.2026–01.01.2027, 2 dage.'),
|
||
('31.08.2026 - 02.09.2026', '31.08.–01.09.2026, 2 dage.'),
|
||
])
|
||
def test_exclusive_period_end(interval, expected):
|
||
assert expected in _period_text(interval)
|
||
|
||
|
||
@pytest.mark.parametrize('changes', [dict(amount=None), dict(amount='NaN'), dict(amount='Infinity'),
|
||
dict(amount='-1'), dict(quantity=0), dict(quantity=None),
|
||
dict(unit_price=None), dict(discount_percentage=10)])
|
||
def test_invalid_source_blocks_instead_of_silent_fallback(changes):
|
||
with pytest.raises(HTTPException):
|
||
period_amount_line(dict(source(), **changes))
|
||
|
||
|
||
def test_zero_is_preserved():
|
||
line = period_amount_line(dict(source(), amount=0, unit_price=0))
|
||
assert line['unit_price'] == line['amount'] == 0
|
||
|
||
|
||
@pytest.mark.parametrize('changes', [dict(quantity=116), dict(unit_price=67.08),
|
||
dict(product_id=99), dict(currency='EUR'),
|
||
dict(discount_percentage=1), dict(source_type='manual'),
|
||
dict(description='changed')])
|
||
def test_stale_or_modified_export_is_blocked(changes):
|
||
line = dict(period_amount_line(source()), **changes)
|
||
with pytest.raises(HTTPException):
|
||
check_period_amounts([line], [source()])
|
||
|
||
|
||
def test_unsaved_also_line_cannot_bypass_approved_amount():
|
||
with pytest.raises(HTTPException):
|
||
check_period_amounts([period_amount_line(source())], [])
|
||
|
||
|
||
def test_other_line_types_are_unchanged():
|
||
line = dict(source(), source_type='manual')
|
||
assert period_amount_line(line) == line
|
||
check_period_amounts([line], [])
|
||
|
||
|
||
def test_repricing_skips_approved_also_period_amounts(monkeypatch):
|
||
from app.products.backend import economic_pricing
|
||
monkeypatch.setattr(economic_pricing, 'query', lambda *args, **kwargs: [])
|
||
assert economic_pricing.proposals({'id': 1, 'customer_id': 1, 'lines_json': [period_amount_line(source())]})['changes'] == []
|
||
|
||
|
||
@pytest.mark.parametrize('raw_line', [{}, {'Charge': '7028.25', 'Unit Price': '67.08'}])
|
||
def test_approval_stores_period_price_with_source_identity(monkeypatch, raw_line):
|
||
from app.modules.also.backend import service
|
||
raw = dict(id=1, import_job_id=5, matched_customer_id=8, matched_product_id=12,
|
||
matched_customer_name='Testkunde', product_name='Microsoft abonnement',
|
||
currency='DKK', billing_start='2026-08-01', billable_parameters=116,
|
||
unit_price=Decimal('67.08'), total_price=Decimal('7028.25'),
|
||
charge_interval='01.08.2026 - 29.08.2026', raw_line_json=raw_line)
|
||
writes = []
|
||
def query(sql, params=()):
|
||
if 'SELECT' in sql and 'FROM also_import_lines l' in sql:
|
||
return [raw]
|
||
writes.append((sql, params))
|
||
return []
|
||
def single(sql, params=()):
|
||
if 'INSERT INTO ordre_drafts' in sql:
|
||
writes.append((sql, params))
|
||
return {'id': 44}
|
||
return None
|
||
monkeypatch.setattr(service.AlsoService, '_assert_enabled', lambda self: None)
|
||
monkeypatch.setattr(service, 'execute_query', query)
|
||
monkeypatch.setattr(service, 'execute_query_single', single)
|
||
monkeypatch.setattr(service, 'with_invoice_fee', lambda lines, customer: lines)
|
||
result = service.AlsoService().approve_lines_to_drafts(5, [1], 3)
|
||
insert = next(params for sql, params in writes if 'INSERT INTO ordre_drafts' in sql)
|
||
line = json.loads(insert[2])[0]
|
||
assert line['quantity'] == 1
|
||
assert line['unit_price'] == line['amount'] == 7028.25
|
||
assert line['source_id'] == 1
|
||
assert line['meta']['also_license_quantity'] == '116'
|
||
assert line['meta']['also_monthly_unit_price'] == '67.08'
|
||
assert result['approved_lines'] == 1
|
||
|
||
|
||
def test_duplicate_source_line_cannot_double_the_charge():
|
||
line = period_amount_line(source())
|
||
with pytest.raises(HTTPException):
|
||
check_period_amounts([line, line], [source()])
|
||
|
||
|
||
def test_existing_pending_draft_is_presented_correctly_without_database_write(monkeypatch):
|
||
import asyncio
|
||
from app.core import database
|
||
from app.modules.orders.backend.router import get_ordre_draft
|
||
old = source()
|
||
row = dict(id=44, sync_status='pending', lines_json=[old], export_status_json={})
|
||
monkeypatch.setattr(database, 'execute_query_single', lambda *args: deepcopy(row))
|
||
def no_write(*args, **kwargs):
|
||
raise AssertionError('Reading a pending draft must not write to the database')
|
||
monkeypatch.setattr(database, 'execute_query', no_write)
|
||
result = asyncio.run(get_ordre_draft(44, None))
|
||
assert result['lines_json'][0]['quantity'] == 1
|
||
assert result['lines_json'][0]['unit_price'] == 7028.25
|
||
assert row['lines_json'][0] == old
|
||
|
||
|
||
@pytest.mark.parametrize('state', [dict(sync_status='exported', economic_order_number='123'),
|
||
dict(sync_status='uncertain'),
|
||
dict(sync_status='pending', export_idempotency_key='attempt-123')])
|
||
def test_exported_or_uncertain_history_is_not_rewritten(monkeypatch, state):
|
||
import asyncio
|
||
from app.core import database
|
||
from app.modules.orders.backend.router import get_ordre_draft
|
||
row = dict(id=44, lines_json=[source()], export_status_json={}, **state)
|
||
monkeypatch.setattr(database, 'execute_query_single', lambda *args: deepcopy(row))
|
||
result = asyncio.run(get_ordre_draft(44, None))
|
||
assert result['lines_json'] == row['lines_json']
|