81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise cleanup ownership checks without a Docker daemon or build."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPT = Path(__file__).with_name('cleanup-integration.sh')
|
|
FAKE_DOCKER = '''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
args = sys.argv[1:]
|
|
with open(os.environ['DOCKER_CALLS'], 'a') as output:
|
|
output.write(json.dumps(args) + '\\n')
|
|
if os.environ.get('EMPTY_INVENTORY'):
|
|
sys.exit(0)
|
|
if args[0] == 'ps':
|
|
if os.environ.get('FAIL_LIST'):
|
|
sys.exit(1)
|
|
print('test-container')
|
|
elif args[:2] == ['volume', 'ls']:
|
|
print('test-volume')
|
|
elif args[:2] == ['network', 'ls']:
|
|
print('test-network')
|
|
elif 'inspect' in args:
|
|
print(os.environ.get('RESOURCE_OWNER', 'prompt-ci-605'))
|
|
'''
|
|
|
|
|
|
class CleanupTests(unittest.TestCase):
|
|
def run_cleanup(self, run_id='605', **extra):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
docker = root / 'docker'
|
|
docker.write_text(FAKE_DOCKER)
|
|
docker.chmod(0o700)
|
|
calls = root / 'calls'
|
|
env = {**os.environ, 'PATH': directory + os.pathsep + os.environ['PATH'],
|
|
'GITEA_RUN_ID': run_id, 'DOCKER_CALLS': str(calls), **extra}
|
|
result = subprocess.run(['bash', str(SCRIPT)], env=env, capture_output=True)
|
|
return result.returncode, [json.loads(line) for line in calls.read_text().splitlines()] if calls.exists() else []
|
|
|
|
def test_removes_only_run_owned_resources_in_dependency_order(self):
|
|
code, calls = self.run_cleanup()
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual([args for args in calls if 'rm' in args], [
|
|
['rm', '-f', 'test-container'], ['volume', 'rm', 'test-volume'],
|
|
['network', 'rm', 'test-network']])
|
|
for args in calls:
|
|
if '--filter' in args:
|
|
self.assertEqual(args[-1], 'label=com.docker.compose.project=prompt-ci-605')
|
|
self.assertEqual(sum('inspect' in args for args in calls), 3)
|
|
|
|
def test_already_clean_stack_is_a_successful_noop(self):
|
|
code, calls = self.run_cleanup(EMPTY_INVENTORY='1')
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(len(calls), 3)
|
|
self.assertFalse(any('rm' in args for args in calls))
|
|
|
|
def test_invalid_or_production_project_never_contacts_docker(self):
|
|
for value in ['', 'prompt', '0', '-1', '605*', '605/../prompt']:
|
|
with self.subTest(value=value):
|
|
code, calls = self.run_cleanup(value)
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_foreign_label_prevents_removal(self):
|
|
code, calls = self.run_cleanup(RESOURCE_OWNER='prompt')
|
|
self.assertNotEqual(code, 0)
|
|
self.assertFalse(any('rm' in args for args in calls))
|
|
|
|
def test_inventory_failure_does_not_continue_with_deletion(self):
|
|
code, calls = self.run_cleanup(FAIL_LIST='1')
|
|
self.assertNotEqual(code, 0)
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|