65 lines
2.9 KiB
Python
65 lines
2.9 KiB
Python
"""Transaction tests: a failed health check must never leave the candidate active."""
|
|
import logging
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from jr_einvoice.maintenance import Maintenance, locked, point
|
|
|
|
|
|
class MaintenanceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.maintenance = Maintenance(self.root, logging.getLogger('maintenance-test'))
|
|
self.state = self.maintenance.state
|
|
self.state.mkdir()
|
|
self.old = self.state / 'old'
|
|
self.new = self.state / 'new'
|
|
self.old.mkdir()
|
|
self.new.mkdir()
|
|
point(self.state / 'current', self.old)
|
|
|
|
def test_success_and_manual_rollback(self):
|
|
with patch.object(self.maintenance, 'health'):
|
|
self.maintenance.activate(self.new)
|
|
self.assertEqual((self.state / 'current').resolve(), self.new)
|
|
self.maintenance.rollback()
|
|
self.assertEqual((self.state / 'current').resolve(), self.old)
|
|
self.assertEqual((self.state / 'previous').resolve(), self.new)
|
|
|
|
def test_post_activation_failure_and_interrupt_restore_old(self):
|
|
for failure in (RuntimeError('bad invoice'), KeyboardInterrupt()):
|
|
with self.subTest(failure=type(failure).__name__):
|
|
with patch.object(self.maintenance, 'health', side_effect=failure):
|
|
with self.assertRaises(type(failure)):
|
|
self.maintenance.activate(self.new)
|
|
self.assertEqual((self.state / 'current').resolve(), self.old)
|
|
|
|
def test_failed_rollback_precheck_keeps_active(self):
|
|
point(self.state / 'previous', self.new)
|
|
with patch.object(self.maintenance, 'health', side_effect=RuntimeError('bad')):
|
|
with self.assertRaises(RuntimeError):
|
|
self.maintenance.rollback()
|
|
self.assertEqual((self.state / 'current').resolve(), self.old)
|
|
|
|
def test_parallel_update_is_rejected(self):
|
|
with locked(self.state):
|
|
with self.assertRaisesRegex(RuntimeError, 'anderer'):
|
|
with locked(self.state):
|
|
self.fail('lock not enforced')
|
|
|
|
def test_download_failure_never_switches(self):
|
|
with patch.object(self.maintenance, 'run'), patch('jr_einvoice.maintenance.install_mustang', side_effect=RuntimeError('checksum')):
|
|
with self.assertRaises(RuntimeError):
|
|
self.maintenance.update(True, '2.26.0')
|
|
self.assertEqual((self.state / 'current').resolve(), self.old)
|
|
|
|
def test_invalid_version_is_rejected_before_commands(self):
|
|
with patch.object(self.maintenance, 'run') as run:
|
|
with self.assertRaises(ValueError):
|
|
self.maintenance.update(True, '../bad')
|
|
run.assert_not_called()
|