40 lines
1.9 KiB
Python
40 lines
1.9 KiB
Python
import subprocess
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from cleanup import container_absent
|
|
|
|
|
|
class PositiveCleanupEvidence(unittest.TestCase):
|
|
def query(self, absent, available):
|
|
def run(args, **options):
|
|
if not available: return subprocess.CompletedProcess(args,1,'','daemon unavailable')
|
|
if 'inspect' in args: return subprocess.CompletedProcess(args,1 if absent else 0,'','no such object' if absent else '')
|
|
return subprocess.CompletedProcess(args,0,'' if absent else '0123456789abcdef\n','')
|
|
return run
|
|
|
|
def test_successful_daemon_query_proves_absence(self):
|
|
with patch('cleanup.subprocess.run',side_effect=self.query(True,True)):
|
|
self.assertTrue(container_absent('jt-sec02-owned'))
|
|
|
|
def test_present_container_is_not_removed(self):
|
|
with patch('cleanup.subprocess.run',side_effect=self.query(False,True)):
|
|
self.assertFalse(container_absent('jt-sec02-owned'))
|
|
|
|
def test_daemon_failure_cannot_certify_cleanup(self):
|
|
with patch('cleanup.subprocess.run',side_effect=self.query(True,False)):
|
|
with self.assertRaises(RuntimeError): container_absent('jt-sec02-owned')
|
|
|
|
def test_timeout_is_unknown_and_calls_are_bounded(self):
|
|
with patch('cleanup.subprocess.run',side_effect=subprocess.TimeoutExpired('docker',5)) as query:
|
|
with self.assertRaises(RuntimeError): container_absent('jt-sec02-owned')
|
|
self.assertEqual(query.call_args.kwargs['timeout'],5)
|
|
|
|
def test_exact_owned_name_filter_is_used(self):
|
|
with patch('cleanup.subprocess.run',side_effect=self.query(True,True)) as query:
|
|
self.assertTrue(container_absent('jt-sec02-owned'))
|
|
self.assertIn('name=^/jt-sec02-owned$',query.call_args.args[0])
|
|
self.assertNotIn('inspect',query.call_args.args[0])
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|