import asyncio from contextlib import contextmanager from datetime import datetime from unittest.mock import Mock import pytest from fastapi import HTTPException from pydantic import ValidationError from app.economy.backend import router as queue @pytest.mark.parametrize('fields,stage,label', [ ({}, 'pending', 'Afventer godkendelse'), ({'status':'approved'}, 'approved','Godkendt'), ({'status':'approved','economy_order_draft_id':4,'order_sync_status':'exported'},'sent','Sendt til fakturering'), ({'economy_order_draft_id':4,'economic_invoice_number':'1001'},'invoiced','Faktureret'), ({'billing_method':'prepaid','status':'billed','economy_settled_at':'today'},'settled','Afregnet'), ({'status':'billed'},'settled','Afregnet'), ]) def test_lifecycle_requires_actual_invoice(fields,stage,label): result = queue._queue_lifecycle(dict(billing_method='invoice',**fields)) if 'billing_method' not in fields else queue._queue_lifecycle(fields) assert (result['lifecycle_status'],result['lifecycle_label']) == (stage,label) if stage in {'sent','invoiced','settled'}: assert not result['can_settle'] and not result['can_change_billing'] def setup_review(monkeypatch): stamp=datetime(2026,10,9,12,0) raw={'id':1,'status':'pending','updated_at':stamp,'billing_method':'internal','original_hours':1} monkeypatch.setattr(queue.settings,'WORKFLOWS_ENABLED',False) monkeypatch.setattr(queue,'_selected_time_entries',lambda ids:[dict(raw)]) monkeypatch.setattr(queue,'_build_settlement_preview',lambda payload:{'valid':True}) payload=queue.ReviewedTimeRequest(ids=[1],snapshots=[{'id':1,'updated_at':stamp.isoformat(),'entry_fingerprint':queue._queue_review_fingerprint(raw),'billing_method':'internal'}]) return raw,payload @pytest.mark.parametrize('change', [{'status':'rejected'},{'economy_order_draft_id':9},{'economy_settled_at':'done'},{'updated_at':datetime(2026,10,9,12,1)}]) def test_reviewed_selection_rejects_changed_or_processed(monkeypatch,change): raw,payload=setup_review(monkeypatch) raw.update(change) with pytest.raises(HTTPException) as error: queue._validate_reviewed_rows(payload,[raw]) assert error.value.status_code == 409 def test_snapshot_must_cover_every_id(monkeypatch): raw,payload=setup_review(monkeypatch) payload.ids=[1,2] with pytest.raises(HTTPException) as error: queue._validate_reviewed_rows(payload,[raw]) assert error.value.status_code == 422 def test_review_cannot_silently_accept_unsaved_override(): with pytest.raises(ValidationError): queue.ReviewedTimeRequest(ids=[1],snapshots=[{'id':1,'updated_at':'today','entry_fingerprint':'test','billing_method':'internal'}],billing_method='invoice') def test_pure_approval_no_financial_operation(monkeypatch): raw,payload=setup_review(monkeypatch) async def actor(request): return {'id':7,'username':'test','is_superadmin':True} monkeypatch.setattr(queue,'_queue_actor',actor) @contextmanager def transaction(): yield None import app.core.database as database monkeypatch.setattr(database,'atomic_transaction',transaction) monkeypatch.setattr(queue,'execute_query',lambda sql,params:[raw]) update=Mock();monkeypatch.setattr(queue,'execute_update',update) insert=Mock(side_effect=AssertionError('Approval must not create financial records')) monkeypatch.setattr(queue,'execute_insert',insert) result=asyncio.run(queue.approve_reviewed_time_queue(payload,None)) assert result['approved']==1 assert update.call_count==1 sql=update.call_args.args[0] assert "status='approved'" in sql and 'economy_settled_at' not in sql insert.assert_not_called() def test_approval_validation_failure_writes_nothing(monkeypatch): raw,payload=setup_review(monkeypatch);raw['status']='rejected' async def actor(request): return {'id':7,'username':'test','is_superadmin':True} monkeypatch.setattr(queue,'_queue_actor',actor) @contextmanager def transaction(): yield None import app.core.database as database monkeypatch.setattr(database,'atomic_transaction',transaction) monkeypatch.setattr(queue,'execute_query',lambda sql,params:[raw]) update=Mock();monkeypatch.setattr(queue,'execute_update',update) with pytest.raises(HTTPException):asyncio.run(queue.approve_reviewed_time_queue(payload,None)) update.assert_not_called() def test_same_revision_new_decision_is_rejected(monkeypatch): raw,payload=setup_review(monkeypatch) monkeypatch.setattr(queue.settings,'WORKFLOWS_ENABLED',True) monkeypatch.setattr(queue,'_attach_workflow_decisions',lambda rows: rows[0].update(workflow_active=True,workflow_stale=False,workflow_decision={'id':11,'entry_revision':0})) payload.snapshots[0].workflow_revision=0 payload.snapshots[0].workflow_decision_id=10 with pytest.raises(HTTPException) as error: queue._validate_reviewed_rows(payload,[raw]) assert error.value.status_code==409 def test_approval_permission_denied_before_transaction(monkeypatch): raw,payload=setup_review(monkeypatch) async def actor(request):return {'id':7,'username':'limited','is_superadmin':False} monkeypatch.setattr(queue,'_queue_actor',actor) from app.core.auth_dependencies import AuthService monkeypatch.setattr(AuthService,'user_has_permission',lambda *args:False) write=Mock();monkeypatch.setattr(queue,'execute_update',write) with pytest.raises(HTTPException) as error:asyncio.run(queue.approve_reviewed_time_queue(payload,None)) assert error.value.status_code==403 write.assert_not_called() def test_reviewed_send_rolls_back_all_effects_on_failure(monkeypatch): raw,payload=setup_review(monkeypatch) async def actor(request):return {'id':7,'username':'test','is_superadmin':True} monkeypatch.setattr(queue,'_queue_actor',actor) effects=[] @contextmanager def transaction(): try:yield None except BaseException: effects.clear() raise import app.core.database as database monkeypatch.setattr(database,'atomic_transaction',transaction) monkeypatch.setattr(queue,'execute_query',lambda sql,params:[raw]) async def settle(payload,request): effects.append('draft') raise HTTPException(409,'A later selected entry failed') monkeypatch.setattr(queue,'settle_time_queue',settle) with pytest.raises(HTTPException):asyncio.run(queue.send_reviewed_time_queue(payload,None)) assert effects==[] @pytest.mark.parametrize('change',[{'original_hours':2},{'description':'Changed'},{'billing_customer_id':9},{'worked_date':'2026-10-10'}]) def test_fingerprint_binds_financial_input_even_without_timestamp_change(monkeypatch,change): raw,payload=setup_review(monkeypatch) raw.update(change) with pytest.raises(HTTPException) as error: queue._validate_reviewed_rows(payload,[raw]) assert error.value.status_code==409 def test_fingerprint_decimal_representation_is_stable(): from decimal import Decimal assert queue._queue_review_fingerprint({'original_hours':Decimal('1.000')}) == queue._queue_review_fingerprint({'original_hours':1.0})