64 lines
2.7 KiB
Python
64 lines
2.7 KiB
Python
"""Deletion must authorize and remove application/confirmation atomically."""
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
from app import app
|
|
from routes import spot
|
|
|
|
|
|
class ConfirmationDeleteTest(unittest.TestCase):
|
|
def delete(self, allowed=True, row=True, failure=False):
|
|
conn = MagicMock()
|
|
cur = conn.cursor.return_value.__enter__.return_value
|
|
cur.fetchone.return_value = {'id': 7, 'product_name': '测试'} if row else None
|
|
if failure:
|
|
def execute(sql, params):
|
|
if sql.startswith('DELETE FROM spot_applications'):
|
|
raise RuntimeError('simulated database failure')
|
|
cur.execute.side_effect = execute
|
|
with app.test_request_context('/api/spot/confirmation/7/delete', method='POST'), \
|
|
patch.object(spot, 'role_required', return_value=True), \
|
|
patch.object(spot, 'operation_required', return_value=allowed) as perm, \
|
|
patch.object(spot, 'get_db', return_value=conn) as db, \
|
|
patch.object(spot, 'bump_data_version') as bump, \
|
|
patch.object(spot, 'add_operation_log'):
|
|
response = spot.confirmation_delete.__wrapped__(7)
|
|
return response, conn, cur, db, bump, perm
|
|
|
|
def test_denied_before_db_access(self):
|
|
response, _, _, db, _, perm = self.delete(allowed=False)
|
|
self.assertEqual(response[1], 403)
|
|
db.assert_not_called()
|
|
perm.assert_called_once_with('spot_confirmation', 'delete')
|
|
|
|
def test_missing_record_never_deletes(self):
|
|
response, conn, cur, _, bump, _ = self.delete(row=False)
|
|
self.assertEqual(response[1], 404)
|
|
self.assertEqual(cur.execute.call_count, 1)
|
|
conn.commit.assert_not_called()
|
|
bump.assert_not_called()
|
|
|
|
def test_deletes_linked_rows_in_one_transaction(self):
|
|
response, conn, cur, _, bump, _ = self.delete()
|
|
self.assertEqual(response.get_json()['code'], 0)
|
|
calls = cur.execute.call_args_list
|
|
self.assertIn('FOR UPDATE', calls[0].args[0])
|
|
self.assertIn("status='通过'", calls[0].args[0])
|
|
self.assertIn('DELETE FROM spot_confirmations', calls[1].args[0])
|
|
self.assertIn('DELETE FROM spot_applications', calls[2].args[0])
|
|
for call in calls:
|
|
self.assertEqual(call.args[1], (7,))
|
|
conn.commit.assert_called_once()
|
|
conn.close.assert_called_once()
|
|
bump.assert_called_once()
|
|
|
|
def test_partial_failure_rolls_back_both_deletions(self):
|
|
response, conn, _, _, bump, _ = self.delete(failure=True)
|
|
self.assertEqual(response[1], 500)
|
|
conn.rollback.assert_called_once()
|
|
conn.commit.assert_not_called()
|
|
bump.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|