4871dd2d9e
- Introduced tests for mail templates including ranking, rendering, and attachment handling. - Added tests for time billing selection and approval processes, ensuring proper validation and state management. - Implemented tests for case types in JavaScript, focusing on loading and editing functionality. - Created tests for contact mail AI integration, ensuring proper parsing and error handling. - Developed tests for dark theme CSS contrast and loading order. - Added tests for invoice fee service, covering various scenarios including exemptions and invalid inputs. - Implemented vendor resource tests, ensuring proper file handling and query validation.
229 lines
11 KiB
Python
229 lines
11 KiB
Python
from decimal import Decimal
|
|
from unittest.mock import Mock
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from app.services import invoice_fee_service as fees
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
state = {'setting': {'value':'9'}, 'customer': {'id':3,'name':'Testkunde','invoice_fee_exempt':False,'invoice_fee_amount':None,'supplier_service_enrolled':False},
|
|
'product': {'id':9,'name':'Faktureringsgebyr','sales_price':49,'sales_currency':'DKK','status':'active','sku_internal':'GEBYR'}}
|
|
def one(sql,params):
|
|
if 'FROM settings' in sql:return state['setting']
|
|
if 'FROM customers' in sql:return state['customer']
|
|
if 'FROM products' in sql:return state['product']
|
|
raise AssertionError(sql)
|
|
monkeypatch.setattr(fees,'execute_query_single',one)
|
|
return state
|
|
|
|
def lines():return [{'line_key':'work','source_type':'sale','quantity':2,'unit_price':100}]
|
|
|
|
def test_standard_price_and_identity_snapshot(db):
|
|
original=lines();result=fees.with_invoice_fee(original,3);fee=result[-1]
|
|
assert len(original)==1 and len(result)==2
|
|
assert fee['product_id']==9 and fee['quantity']==1 and fee['unit_price']==49
|
|
assert fee['source_type']=='invoice_fee' and fee['fee_snapshot']['amount']=='49.00'
|
|
assert sum(row['quantity']*row['unit_price'] for row in result)==249
|
|
|
|
@pytest.mark.parametrize('amount',[12.5,99])
|
|
def test_customer_override(db,amount):
|
|
db['customer']['invoice_fee_amount']=amount
|
|
assert fees.with_invoice_fee(lines(),3)[-1]['unit_price']==amount
|
|
|
|
@pytest.mark.parametrize('field,value',[('invoice_fee_exempt',True),('supplier_service_enrolled',True),('invoice_fee_amount',0)])
|
|
def test_exemptions(db,field,value):
|
|
db['customer'][field]=value
|
|
assert fees.with_invoice_fee(lines(),3)==lines()
|
|
|
|
def test_no_setting_preserves_legacy_workflows(db):
|
|
db['setting']=None
|
|
assert fees.with_invoice_fee(lines(),3)==lines()
|
|
assert fees.with_invoice_fee(lines(),3,legacy=True)[-1]['unit_price']==49
|
|
|
|
def test_invalid_selected_product_is_not_silently_skipped(db):
|
|
db['product']=None
|
|
with pytest.raises(HTTPException) as error:fees.with_invoice_fee(lines(),3)
|
|
assert error.value.status_code==409
|
|
|
|
@pytest.mark.parametrize('amount',[-1,float('nan'),float('inf')])
|
|
def test_invalid_amount_rejected(db,amount):
|
|
db['customer']['invoice_fee_amount']=amount
|
|
with pytest.raises(HTTPException):fees.with_invoice_fee(lines(),3)
|
|
|
|
def test_native_subscription_shape(db):
|
|
result=fees.with_invoice_fee([{'quantity':1,'unitNetPrice':100,'product':{'productNumber':'SUB'}}],3,native=True)
|
|
fee=result[-1]
|
|
assert fee['unitNetPrice']==49 and fee['product']['productNumber']=='GEBYR'
|
|
assert fee['product_id']==9
|
|
|
|
def test_credit_and_empty_orders_are_not_charged(db):
|
|
assert fees.with_invoice_fee([],3)==[]
|
|
credit=[{'quantity':1,'unitNetPrice':-100}]
|
|
assert fees.with_invoice_fee(credit,3,native=True)==credit
|
|
|
|
def test_incoming_fee_is_replaced_with_one_server_calculated_fee(db):
|
|
incoming=lines()+[{'source_type':'invoice_fee','quantity':1,'unit_price':1}]
|
|
result=fees.with_invoice_fee(incoming,3)
|
|
assert len(result)==2 and result[-1]['unit_price']==49
|
|
|
|
def test_snapshot_does_not_change_when_customer_price_changes(db):
|
|
saved=fees.with_invoice_fee(lines(),3)
|
|
db['customer']['invoice_fee_amount']=20
|
|
assert saved[-1]['unit_price']==49
|
|
second=fees.with_invoice_fee(lines(),3)
|
|
assert second[-1]['unit_price']==20
|
|
assert saved[-1]['line_key']!=second[-1]['line_key']
|
|
|
|
def test_consolidation_preserves_each_orders_unique_fee(db):
|
|
first=fees.with_invoice_fee(lines(),3);second=fees.with_invoice_fee(lines(),3)
|
|
merged=first+second
|
|
assert len([row for row in merged if row['source_type']=='invoice_fee'])==2
|
|
|
|
def test_currency_mismatch_fails_instead_of_relabelling(db):
|
|
with pytest.raises(HTTPException) as error:fees.with_invoice_fee([{'quantity':1,'unit_price':100,'currency':'EUR'}],3)
|
|
assert error.value.status_code==422
|
|
|
|
def test_inactive_product(db):
|
|
db['product']['status']='inactive'
|
|
with pytest.raises(HTTPException):fees.with_invoice_fee(lines(),3)
|
|
|
|
def test_customer_must_exist(db):
|
|
db['customer']=None
|
|
with pytest.raises(HTTPException):fees.with_invoice_fee(lines(),3)
|
|
|
|
def test_cursor_is_reused_without_separate_database_connection(db,monkeypatch):
|
|
cursor=Mock()
|
|
cursor.fetchone.side_effect=[db['setting'],db['customer'],db['product']]
|
|
monkeypatch.setattr(fees,'execute_query_single',Mock(side_effect=AssertionError('outside transaction')))
|
|
assert fees.with_invoice_fee(lines(),3,cursor=cursor)[-1]['unit_price']==49
|
|
|
|
@pytest.mark.parametrize('value',['bad','0','-1'])
|
|
def test_invalid_product_setting(db,value):
|
|
db['setting']['value']=value
|
|
with pytest.raises(HTTPException):fees.with_invoice_fee(lines(),3)
|
|
|
|
def test_free_or_deselected_order_does_not_get_a_fee(db):
|
|
free=[{'quantity':1,'unit_price':100,'discount_percentage':100}]
|
|
assert fees.with_invoice_fee(free,3)==free
|
|
deselected=[{'quantity':1,'unit_price':100,'selected':False}]
|
|
assert fees.with_invoice_fee(deselected,3)==deselected
|
|
|
|
def test_manual_draft_persists_fee_and_update_keeps_original_amount(db,monkeypatch):
|
|
import asyncio,json
|
|
from unittest.mock import AsyncMock
|
|
from app.modules.orders.backend import router as orders
|
|
from app.core import database
|
|
captured=[]
|
|
monkeypatch.setattr(orders,'_check_draft_prices',AsyncMock())
|
|
monkeypatch.setattr(orders,'_get_user_id_from_request',lambda _:None)
|
|
monkeypatch.setattr(orders,'_snapshot_lines',lambda rows:rows)
|
|
def save(sql,params):
|
|
captured.append((sql,params))
|
|
return [{'id':8,'lines_json':json.loads(params[2])}]
|
|
monkeypatch.setattr(database,'execute_query',save)
|
|
request=orders.OrdreDraftUpsertRequest(title='Testordre',customer_id=3,lines=lines())
|
|
created=asyncio.run(orders.create_ordre_draft(request,None))
|
|
assert created['lines_json'][-1]['unit_price']==49
|
|
db['customer']['invoice_fee_amount']=100
|
|
monkeypatch.setattr(database,'execute_query_single',lambda *args:{'customer_id':3,'lines_json':created['lines_json'],'sync_status':'pending'})
|
|
updated=asyncio.run(orders.update_ordre_draft(8,orders.OrdreDraftUpsertRequest(title='Testordre',customer_id=3,lines=created['lines_json']),None))
|
|
assert updated['lines_json'][-1]['unit_price']==49
|
|
assert len(updated['lines_json'])==2
|
|
|
|
|
|
def test_customer_api_accepts_null_override_and_rejects_negative():
|
|
from pydantic import ValidationError
|
|
from app.customers.backend.router import CustomerUpdate
|
|
assert CustomerUpdate(invoice_fee_amount=None).model_dump(exclude_unset=True)=={'invoice_fee_amount':None}
|
|
assert CustomerUpdate(invoice_fee_exempt=True).invoice_fee_exempt
|
|
with pytest.raises(ValidationError):CustomerUpdate(invoice_fee_amount=-1)
|
|
|
|
|
|
def test_fee_is_excluded_from_explicit_catalog_repricing(monkeypatch):
|
|
from app.products.backend import economic_pricing
|
|
monkeypatch.setattr(economic_pricing,'query',lambda *_args,**_kwargs:[])
|
|
assert economic_pricing.proposals({'id':8,'customer_id':3,'lines_json':[{'source_type':'invoice_fee','product_id':9,'unit_price':25}]})['changes']==[]
|
|
|
|
def test_plain_subscription_cursor_rows_are_converted_to_named_fields(db,monkeypatch):
|
|
class TupleCursor:
|
|
description=None
|
|
def execute(self,sql,params):
|
|
if 'FROM settings' in sql:row=db['setting']
|
|
elif 'FROM customers' in sql:row=db['customer']
|
|
else:row=db['product']
|
|
self.description=[(key,) for key in row]
|
|
self.row=tuple(row.values())
|
|
def fetchone(self):return self.row
|
|
monkeypatch.setattr(fees,'execute_query_single',Mock(side_effect=AssertionError('outside transaction')))
|
|
assert fees.with_invoice_fee(lines(),3,cursor=TupleCursor(),native=True)[-1]['unitNetPrice']==49
|
|
|
|
def _sub(identifier=1):
|
|
from datetime import date
|
|
return {'id':identifier,'customer_id':3,'customer_name':'Testkunde','billing_interval':'monthly',
|
|
'period_start':date(2026,10,1),'next_invoice_date':date(2026,10,1),
|
|
'line_items':[{'product_id':20,'description':'Abonnement','quantity':1,'unit_price':100,'line_total':100}]}
|
|
|
|
def _cursor_fee_data(db,monkeypatch):
|
|
def one(sql,params,cursor=None):
|
|
if 'FROM settings' in sql:return db['setting']
|
|
if 'FROM customers' in sql:return db['customer']
|
|
if 'FROM products' in sql:return db['product']
|
|
raise AssertionError(sql)
|
|
monkeypatch.setattr(fees,'_one',one)
|
|
|
|
|
|
def test_subscription_job_adds_one_fee_per_group_and_advances_after_creation(db,monkeypatch):
|
|
import asyncio,json
|
|
from app.jobs import process_subscriptions as job
|
|
_cursor_fee_data(db,monkeypatch)
|
|
cursor=Mock();cursor.fetchone.side_effect=[(1,),(2,),(100,)]
|
|
conn=Mock();conn.cursor.return_value=cursor
|
|
monkeypatch.setattr(job,'get_db_connection',lambda:conn)
|
|
assert asyncio.run(job._process_subscription_group([_sub(1),_sub(2)]))==2
|
|
insertion=next(call for call in cursor.execute.call_args_list if 'INSERT INTO ordre_drafts' in call.args[0])
|
|
saved=json.loads(insertion.args[1][2])
|
|
assert len(saved)==3 and sum(row.get('source_type')=='invoice_fee' for row in saved)==1
|
|
assert saved[-1]['product_id']==9
|
|
conn.commit.assert_called_once()
|
|
|
|
|
|
def test_subscription_fee_failure_rolls_back_and_does_not_advance_period(db,monkeypatch):
|
|
import asyncio
|
|
from app.jobs import process_subscriptions as job
|
|
db['product']=None;_cursor_fee_data(db,monkeypatch)
|
|
cursor=Mock();cursor.fetchone.return_value=(1,)
|
|
conn=Mock();conn.cursor.return_value=cursor
|
|
monkeypatch.setattr(job,'get_db_connection',lambda:conn)
|
|
with pytest.raises(HTTPException):asyncio.run(job._process_subscription_group([_sub()]))
|
|
conn.rollback.assert_called_once();conn.commit.assert_not_called()
|
|
assert not any('UPDATE sag_subscriptions' in call.args[0] for call in cursor.execute.call_args_list)
|
|
|
|
|
|
def test_customer_scoped_rental_billing_uses_one_fee(db,monkeypatch):
|
|
import json
|
|
from datetime import date
|
|
from app.modules.rentals.backend import router as rentals
|
|
_cursor_fee_data(db,monkeypatch)
|
|
cursor=Mock();cursor.__enter__=Mock(return_value=cursor);cursor.__exit__=Mock(return_value=False)
|
|
cursor.fetchall.return_value=[_sub()];cursor.fetchone.return_value=None
|
|
conn=Mock();conn.cursor.return_value=cursor
|
|
monkeypatch.setattr(rentals,'get_db_connection',lambda:conn)
|
|
monkeypatch.setattr(rentals,'release_db_connection',lambda _:None)
|
|
result=rentals._generate_due_invoices_for_customer(date(2026,10,4),3)
|
|
assert result['created_drafts']==1
|
|
insertion=next(call for call in cursor.execute.call_args_list if 'INSERT INTO ordre_drafts' in call.args[0])
|
|
saved=json.loads(insertion.args[1][2]);assert saved[-1]['source_type']=='invoice_fee'
|
|
assert saved[-1]['unit_price']==49
|
|
|
|
|
|
def test_legacy_time_order_fee_changes_net_and_vat_but_not_working_hours(db,monkeypatch):
|
|
from app.economy.backend import router as economy
|
|
monkeypatch.setattr(economy,'execute_query_single',lambda *args:{'id':4,'hub_customer_id':3,'name':'Test','hourly_rate':100})
|
|
writes=[]
|
|
monkeypatch.setattr(economy,'execute_insert',lambda sql,params:writes.append((sql,params)) or 100)
|
|
economy._create_order_from_selected(4,[{'id':1,'approved_hours':Decimal('2')}],None)
|
|
order=writes[0][1]
|
|
assert order[2]==Decimal('2') and order[4]==Decimal('249')
|
|
assert order[6]==Decimal('62.25') and order[7]==Decimal('311.25')
|
|
assert "'invoice_fee'" in writes[-1][0]
|