36 lines
2.1 KiB
Python
36 lines
2.1 KiB
Python
|
import asyncio
|
||
|
|
from datetime import date, timedelta
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
import pytest
|
||
|
|
from fastapi import HTTPException
|
||
|
|
from app.subscriptions.backend import router
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('status,days',[('draft',0),('pending',0),('draft',7)])
|
||
|
|
def test_save_without_approval_applies_or_schedules(status,days):
|
||
|
|
cursor=MagicMock(); cursor.__enter__.return_value=cursor
|
||
|
|
cursor.fetchone.return_value={'id':1,'main_sag_id':79,'status':status,'effective_date':date.today()+timedelta(days=days),'created_by_user_id':4}
|
||
|
|
conn=MagicMock();conn.cursor.return_value=cursor
|
||
|
|
with patch.object(router,'get_db_connection',return_value=conn),patch.object(router,'release_db_connection'),patch.object(router,'apply_subscription_change_request',return_value={'id':1,'status':'applied'}) as apply:
|
||
|
|
result=asyncio.run(router.save_subscription_change_directly(1,{'id':4}))
|
||
|
|
conn.commit.assert_called_once()
|
||
|
|
assert 'approved_by_user_id=NULL' in cursor.execute.call_args_list[1].args[0]
|
||
|
|
assert 'change_saved_directly' in cursor.execute.call_args_list[2].args[0]
|
||
|
|
if days:
|
||
|
|
apply.assert_not_called();assert result['status']=='approved_scheduled'
|
||
|
|
else:
|
||
|
|
apply.assert_called_once_with(1,4);assert result['status']=='applied'
|
||
|
|
|
||
|
|
|
||
|
|
def test_repeat_save_does_not_apply_twice_and_rejected_is_blocked():
|
||
|
|
for status in ['applied','rejected']:
|
||
|
|
cursor=MagicMock();cursor.__enter__.return_value=cursor
|
||
|
|
cursor.fetchone.return_value={'status':status}
|
||
|
|
conn=MagicMock();conn.cursor.return_value=cursor
|
||
|
|
with patch.object(router,'get_db_connection',return_value=conn),patch.object(router,'release_db_connection'),patch.object(router,'apply_subscription_change_request') as apply:
|
||
|
|
if status=='applied': assert asyncio.run(router.save_subscription_change_directly(1,{'id':4}))['status']=='applied'
|
||
|
|
else:
|
||
|
|
with pytest.raises(HTTPException) as error: asyncio.run(router.save_subscription_change_directly(1,{'id':4}))
|
||
|
|
assert error.value.status_code==409
|
||
|
|
apply.assert_not_called();conn.rollback.assert_called_once()
|