sashatrask
197 lines
9.7 KiB
Python
197 lines
9.7 KiB
Python
"""Pure Compose regressions: python -I -S -B docker/test_verify.py -v."""
|
|
|
|
from copy import deepcopy
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
'truf_verify', Path(__file__).with_name('verify.py'))
|
|
verify = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(verify)
|
|
|
|
|
|
class ValidateComposeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
# Only project/images are needed; skip all host setup and never run main.
|
|
self.verifier = object.__new__(verify.Verifier)
|
|
self.verifier.project = 'truf-worker-test-' + 'a' * 32
|
|
self.verifier.images = {'runtime': 'sha256:' + '1' * 64,
|
|
'test': 'sha256:' + '2' * 64}
|
|
common = {
|
|
'pull_policy': 'never', 'read_only': True, 'network_mode': 'none',
|
|
'environment': dict(verify.PROXY_ENV), 'init': False,
|
|
'user': '10001:10001', 'cap_drop': ['ALL'],
|
|
'security_opt': ['no-new-privileges:true'], 'restart': 'no',
|
|
'cpus': 2, 'mem_limit': 6442450944, 'pids_limit': 512,
|
|
'shm_size': 268435456, 'stop_signal': 'SIGTERM',
|
|
'stop_grace_period': '10m0s',
|
|
'logging': {'driver': 'json-file',
|
|
'options': {'max-size': '16m', 'max-file': '4'}},
|
|
'tmpfs': [target + ':' + options for target, options in verify.TMPFS.items()],
|
|
}
|
|
data = {'type': 'volume', 'source': 'data', 'target': '/data', 'volume': {}}
|
|
tools = {'type': 'volume', 'source': 'tools', 'target': '/opt/truf/tests',
|
|
'read_only': True, 'volume': {'nocopy': True}}
|
|
services = {name: deepcopy(common)
|
|
for name in ('tools', 'provision', 'prepare', 'runtime', 'stopped')}
|
|
for name, service in services.items():
|
|
service['image'] = self.verifier.images['test' if name == 'tools' else 'runtime']
|
|
service['volumes'] = [deepcopy(data)]
|
|
# Compose 5 omits false nocopy/read_only; byte sizes are normalized integers.
|
|
services['tools'].update(
|
|
volumes=[{'type': 'volume', 'source': 'tools',
|
|
'target': '/opt/truf/tests', 'volume': {}}],
|
|
entrypoint=[*verify.PYTHON, '-c'], command=['pass'])
|
|
services['provision'].update(
|
|
user='0:0', cap_add=['CHOWN', 'DAC_OVERRIDE', 'FOWNER'],
|
|
entrypoint=None, command=['provision'])
|
|
services['prepare'].update(
|
|
volumes=[deepcopy(data), deepcopy(tools)],
|
|
entrypoint=[*verify.PYTHON, verify.DRIVER],
|
|
command=['prepare', '--config', verify.CONFIG])
|
|
services['runtime'].update(
|
|
volumes=[deepcopy(data), deepcopy(tools)], entrypoint=None,
|
|
command=['run', '--config', verify.CONFIG],
|
|
healthcheck={'test': list(verify.HEALTH), 'interval': '5s',
|
|
'timeout': '15s', 'start_period': '4m0s', 'retries': 3})
|
|
services['stopped'].update(
|
|
volumes=[dict(data, read_only=True, volume={'nocopy': True})],
|
|
entrypoint=[*verify.PYTHON, '-c'], command=['pass'])
|
|
self.value = {
|
|
'name': self.verifier.project, 'services': services,
|
|
'volumes': {name: {'name': self.verifier.project + '_' + name, 'driver': 'local'}
|
|
for name in ('data', 'tools')},
|
|
}
|
|
|
|
def test_normalized_fixture(self):
|
|
self.verifier.validate_compose(self.value)
|
|
|
|
def test_rejects_production_project_and_volume_names(self):
|
|
value = deepcopy(self.value)
|
|
value['name'] = 'truf-docker'
|
|
with self.assertRaisesRegex(verify.Failure, '^worker_test_project_guard$'):
|
|
self.verifier.validate_compose(value)
|
|
value = deepcopy(self.value)
|
|
value['volumes']['data']['name'] = 'truf-docker_data'
|
|
with self.assertRaisesRegex(verify.Failure, '^production_volume_forbidden$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_rejects_bind_mounts_and_published_ports(self):
|
|
value = deepcopy(self.value)
|
|
value['services']['runtime']['volumes'][0] = {
|
|
'type': 'bind', 'source': r'D:\truf-docker', 'target': '/data',
|
|
}
|
|
with self.assertRaisesRegex(verify.Failure, '^bind_mount_forbidden$'):
|
|
self.verifier.validate_compose(value)
|
|
value = deepcopy(self.value)
|
|
value['services']['runtime']['ports'] = [{'target': 5432, 'published': '5432'}]
|
|
with self.assertRaisesRegex(verify.Failure, '^published_port_contract$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_rejects_shared_images_and_nonisolated_networks(self):
|
|
value = deepcopy(self.value)
|
|
value['services']['runtime']['image'] = 'truf-local:runtime'
|
|
with self.assertRaisesRegex(verify.Failure, '^worker_test_image_reference_guard$'):
|
|
self.verifier.validate_compose(value)
|
|
value = deepcopy(self.value)
|
|
value['services']['runtime']['network_mode'] = 'bridge'
|
|
with self.assertRaisesRegex(verify.Failure, '^internal_network_contract$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_seed_accepts_false_or_omitted_nocopy(self):
|
|
for volume in (None, {}, {'nocopy': False}):
|
|
with self.subTest(volume=volume):
|
|
value = deepcopy(self.value)
|
|
mount = value['services']['tools']['volumes'][0]
|
|
if volume is None:
|
|
del mount['volume']
|
|
else:
|
|
mount['volume'] = volume
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_seed_rejects_nocopy_other_than_false(self):
|
|
for nocopy in (True, None, 0, 'false'):
|
|
with self.subTest(nocopy=nocopy):
|
|
value = deepcopy(self.value)
|
|
value['services']['tools']['volumes'][0]['volume']['nocopy'] = nocopy
|
|
with self.assertRaisesRegex(verify.Failure, '^tools_copy_up_contract$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_consumers_require_explicit_true_nocopy(self):
|
|
for name in ('prepare', 'runtime'):
|
|
for volume in (None, {}, {'nocopy': False}, {'nocopy': 1}, {'nocopy': 'true'}):
|
|
with self.subTest(service=name, volume=volume):
|
|
value = deepcopy(self.value)
|
|
mount = value['services'][name]['volumes'][1]
|
|
if volume is None:
|
|
del mount['volume']
|
|
else:
|
|
mount['volume'] = volume
|
|
with self.assertRaisesRegex(verify.Failure, '^tools_copy_up_contract$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_runtime_and_provision_inherit_entrypoint(self):
|
|
for name in ('runtime', 'provision'):
|
|
for omitted in (False, True):
|
|
with self.subTest(service=name, omitted=omitted):
|
|
value = deepcopy(self.value)
|
|
if omitted:
|
|
del value['services'][name]['entrypoint']
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_runtime_and_provision_reject_entrypoint_overrides(self):
|
|
for name, guard in (('runtime', 'production_entrypoint_contract'),
|
|
('provision', 'prepare_command_contract')):
|
|
for entrypoint in ([], ['/bin/sh', '-c'], ''):
|
|
with self.subTest(service=name, entrypoint=entrypoint):
|
|
value = deepcopy(self.value)
|
|
value['services'][name]['entrypoint'] = entrypoint
|
|
with self.assertRaisesRegex(verify.Failure, '^' + guard + '$'):
|
|
self.verifier.validate_compose(value)
|
|
|
|
def test_foreign_snapshot_excludes_only_revalidated_owned_identities(self):
|
|
owned_container = 'a' * 64
|
|
foreign_container = 'b' * 64
|
|
owned_volume = {'name': 'owned', 'identity': 'captured'}
|
|
foreign_volume = {'name': 'foreign', 'identity': 'stable'}
|
|
current = {'owned': owned_volume, 'foreign': foreign_volume}
|
|
verifier = object.__new__(verify.Verifier)
|
|
verifier.owned_containers = {owned_container: {'id': owned_container}}
|
|
verifier.owned_volumes = {'owned': owned_volume}
|
|
verifier.metadata_names = lambda kind: (
|
|
{owned_container, foreign_container} if kind == 'container'
|
|
else {'owned', 'foreign'}
|
|
)
|
|
verifier.container_metadata = lambda identifier: {'id': identifier}
|
|
verifier.volume_metadata = lambda name: current[name]
|
|
inspected = []
|
|
verifier.inspect = lambda identifier, label: inspected.append((identifier, label))
|
|
|
|
snapshot = verifier.metadata_snapshot(exclude_owned=True)
|
|
|
|
self.assertEqual(snapshot, {
|
|
'containers': {foreign_container: {'id': foreign_container}},
|
|
'volumes': {'foreign': foreign_volume},
|
|
})
|
|
self.assertEqual(inspected, [(owned_container, 'foreign_owned_exclusion_guard')])
|
|
|
|
current['owned'] = {'name': 'owned', 'identity': 'replaced'}
|
|
self.assertIn('owned', verifier.metadata_snapshot(exclude_owned=True)['volumes'])
|
|
|
|
def test_cleanup_uses_exact_resource_operations_and_foreign_guard(self):
|
|
source = Path(verify.__file__).read_text(encoding='ascii')
|
|
self.assertNotIn("compose_command('down'", source)
|
|
self.assertNotIn("'--force-recreate'", source)
|
|
self.assertIn("'cleanup_container_remove_guard'", source)
|
|
self.assertIn("'cleanup_volume_identity_changed'", source)
|
|
self.assertIn('self.assert_foreign_unchanged()', source)
|
|
self.assertIn("'failure_stop_ownership_guard'", source)
|
|
self.assertIn("'stage': label, 'class': failure_class", source)
|
|
self.assertIn("'exit_code': exit_code", source)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|