73 lines
3.1 KiB
Python
73 lines
3.1 KiB
Python
"""Financial safety regressions; DB integration tests use a dedicated database only."""
|
|
from decimal import Decimal
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from app.core import database
|
|
from app.workflows import settlement
|
|
|
|
|
|
class Cursor:
|
|
description = [('id',)]
|
|
rowcount = 1
|
|
def __init__(self, conn): self.conn=conn
|
|
def __enter__(self): return self
|
|
def __exit__(self,*args): pass
|
|
def execute(self,query,params): self.conn.queries.append(query)
|
|
def fetchone(self): return {'id':1}
|
|
def fetchall(self): return [{'id':1}]
|
|
|
|
|
|
class Connection:
|
|
def __init__(self): self.commits=0;self.rollbacks=0;self.queries=[]
|
|
def cursor(self,**kwargs): return Cursor(self)
|
|
def commit(self): self.commits+=1
|
|
def rollback(self): self.rollbacks+=1
|
|
|
|
|
|
def test_atomic_helpers_commit_once(monkeypatch):
|
|
conn=Connection();released=[]
|
|
monkeypatch.setattr(database,'get_db_connection',lambda:conn)
|
|
monkeypatch.setattr(database,'release_db_connection',released.append)
|
|
with database.atomic_transaction():
|
|
database.execute_query('UPDATE card SET used=1 RETURNING id')
|
|
database.execute_insert('INSERT INTO ledger DEFAULT VALUES RETURNING id')
|
|
database.execute_update('UPDATE time SET settled=true')
|
|
assert conn.commits==0 and not released
|
|
assert conn.commits==1 and conn.rollbacks==0 and released==[conn]
|
|
|
|
|
|
def test_atomic_helpers_rollback_all_on_failure(monkeypatch):
|
|
conn=Connection()
|
|
monkeypatch.setattr(database,'get_db_connection',lambda:conn)
|
|
monkeypatch.setattr(database,'release_db_connection',lambda c:None)
|
|
with pytest.raises(RuntimeError):
|
|
with database.atomic_transaction():
|
|
database.execute_update('UPDATE card SET used=1')
|
|
raise RuntimeError('failure before time is settled')
|
|
assert conn.commits==0 and conn.rollbacks==1
|
|
# Context is cleared after rollback, so subsequent legacy writes still commit.
|
|
database.execute_update('UPDATE unrelated SET flag=true')
|
|
assert conn.commits==1
|
|
|
|
|
|
def test_stale_or_review_decision_cannot_settle(monkeypatch):
|
|
monkeypatch.setattr(settlement,'execute_query_single',lambda *a:{'revision':2,'status':'decided','method':'invoice'})
|
|
with pytest.raises(HTTPException) as exc: settlement.current_decision(1,3)
|
|
assert exc.value.status_code==409
|
|
monkeypatch.setattr(settlement,'execute_query_single',lambda *a:{'revision':3,'status':'manual_review','method':None})
|
|
with pytest.raises(HTTPException): settlement.current_decision(1,3)
|
|
|
|
|
|
def test_coverage_exclusions_win():
|
|
resource={'workflow_coverage':{'tag_ids':[2],'excluded_tag_ids':[3]}}
|
|
with pytest.raises(HTTPException):
|
|
settlement._check_coverage(resource,{'details':{'input':{'tag_ids':[2,3]}}})
|
|
|
|
|
|
def test_disabled_feature_uses_no_workflow_customer(monkeypatch):
|
|
from app.workflows.integration import active_for_customer,settings
|
|
monkeypatch.setattr(settings,'WORKFLOWS_ENABLED',False)
|
|
monkeypatch.setattr(settings,'WORKFLOWS_MODE','active')
|
|
monkeypatch.setattr(settings,'WORKFLOWS_CUSTOMER_IDS',[24])
|
|
assert not active_for_customer(24)
|