69 lines
3.1 KiB
Python
69 lines
3.1 KiB
Python
import copy
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
path = Path(__file__).with_name('activate_forge_template.py')
|
|
spec = importlib.util.spec_from_file_location('template', path)
|
|
t = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(t)
|
|
|
|
|
|
def fixture():
|
|
def container(name):
|
|
return {'name': name, 'image': 'unchanged:fixture', 'resources': {'requests': {'cpu': '100m'}},
|
|
'volumeMounts': [{'name': 'data', 'mountPath': '/data'}],
|
|
'env': [{'name': 'EXISTING_REFERENCE', 'valueFrom': {'secretKeyRef': {'name': 'existing', 'key': 'reference-only'}}}]}
|
|
return {'metadata': {'uid': 'fixed', 'resourceVersion': '17'}, 'spec': {'template': {'metadata': {'labels': {'preserved': 'yes'}},
|
|
'spec': {'volumes': [{'name': 'data', 'persistentVolumeClaim': {'claimName': 'existing'}}],
|
|
'initContainers': [container('init-directories'), container('init-app-ini'), container('configure-gitea')],
|
|
'containers': [container('gitea')]}}}}
|
|
|
|
|
|
class Tests(unittest.TestCase):
|
|
def test_rollback_preserves_existing_configuration_and_explicitly_restores_default(self):
|
|
before = fixture()
|
|
active = t.apply_locally(before, t.patch_for(before))
|
|
rolled = t.apply_locally(active, t.patch_for(active, rollback=True))
|
|
expected = copy.deepcopy(before)
|
|
expected['spec']['template']['spec']['initContainers'][1]['env'].append({'name': t.SETTING, 'value': t.DEFAULT})
|
|
self.assertEqual(rolled, expected)
|
|
self.assertEqual(before, fixture())
|
|
|
|
def test_stale_resource_version_cannot_apply(self):
|
|
before = fixture()
|
|
patch = t.patch_for(before)
|
|
before['metadata']['resourceVersion'] = '18'
|
|
with self.assertRaisesRegex(ValueError, 'precondition'):
|
|
t.apply_locally(before, patch)
|
|
|
|
def test_existing_setting_not_overwritten(self):
|
|
d = fixture()
|
|
d['spec']['template']['spec']['initContainers'][1]['env'].append({'name': t.SETTING, 'value': '/unknown'})
|
|
with self.assertRaisesRegex(ValueError, 'setting-preexists'):
|
|
t.patch_for(d)
|
|
|
|
def test_conflicting_mount_not_overwritten(self):
|
|
d = fixture()
|
|
d['spec']['template']['spec']['containers'][0]['volumeMounts'].append({'name': 'other', 'mountPath': t.MOUNT['mountPath']})
|
|
with self.assertRaisesRegex(ValueError, 'mount-preexists'):
|
|
t.patch_for(d)
|
|
|
|
def test_rollback_preserves_concurrently_modified_mount(self):
|
|
d = fixture()
|
|
d = t.apply_locally(d, t.patch_for(d))
|
|
d['spec']['template']['spec']['containers'][0]['volumeMounts'][-1]['subPath'] = 'unknown'
|
|
original = copy.deepcopy(d)
|
|
with self.assertRaisesRegex(ValueError, 'rollback-mount-changed'):
|
|
t.patch_for(d, rollback=True)
|
|
self.assertEqual(d, original)
|
|
|
|
def test_missing_configuration_container_refused(self):
|
|
d = fixture()
|
|
d['spec']['template']['spec']['initContainers'].pop(1)
|
|
with self.assertRaisesRegex(ValueError, 'configuration-container'):
|
|
t.patch_for(d)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|