60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import subprocess
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / 'scripts' / 'ding_daily_sync.py'
|
|
SPEC = importlib.util.spec_from_file_location('ding_daily_sync_retry_test', SCRIPT)
|
|
SYNC = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(SYNC)
|
|
|
|
|
|
class Result:
|
|
def __init__(self, returncode, stdout=b''):
|
|
self.returncode = returncode
|
|
self.stdout = stdout
|
|
|
|
|
|
class DingCommandRetryTest(unittest.TestCase):
|
|
def test_dws_command_retries_then_succeeds(self):
|
|
results = iter([Result(1), Result(1), Result(0, b'published')])
|
|
sleeps = []
|
|
with mock.patch.object(SYNC.subprocess, 'run', side_effect=lambda *args, **kwargs: next(results)), \
|
|
mock.patch.object(SYNC.time, 'sleep', side_effect=sleeps.append):
|
|
self.assertEqual(SYNC.dws_command([SYNC.DWS, 'drive', 'list']), b'published')
|
|
self.assertEqual(sleeps, [2, 5])
|
|
|
|
def test_dws_command_stops_after_bounded_attempts(self):
|
|
calls = []
|
|
|
|
def run(*args, **kwargs):
|
|
calls.append(args)
|
|
return Result(1)
|
|
|
|
with mock.patch.object(SYNC.subprocess, 'run', side_effect=run), \
|
|
mock.patch.object(SYNC.time, 'sleep'):
|
|
with self.assertRaisesRegex(RuntimeError, r'failed \(exit 1\) after 4 attempts'):
|
|
SYNC.dws_command([SYNC.DWS, 'drive', 'download'])
|
|
self.assertEqual(len(calls), 4)
|
|
|
|
def test_dws_command_retries_timeout_without_leaking_output(self):
|
|
calls = 0
|
|
|
|
def run(*args, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
raise subprocess.TimeoutExpired(args[0], kwargs['timeout'])
|
|
return Result(0, b'ok')
|
|
|
|
with mock.patch.object(SYNC.subprocess, 'run', side_effect=run), \
|
|
mock.patch.object(SYNC.time, 'sleep'):
|
|
self.assertEqual(SYNC.dws_command([SYNC.DWS, 'drive', 'list']), b'ok')
|
|
self.assertEqual(calls, 2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|