railiance-apps/tests/test_vergabe_pilot_chart.py
tegwick 959eba637f Prepare durable isolated Vergabe pilot deployment and recovery
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
2026-09-11 16:35:14 +02:00

110 lines
4.7 KiB
Python

"""Render-level checks for data durability and separation; no cluster mutations.
Run with: python3 -m unittest discover -s tests -p 'test_vergabe_pilot_chart.py'
Requires Helm and PyYAML.
"""
import subprocess
import unittest
from pathlib import Path
import yaml
CHART = Path(__file__).resolve().parents[1] / 'charts/vergabe-teilnahme'
DIGEST = 'sha256:' + 'a' * 64
def render(values):
return subprocess.run(
['helm', 'template', 'pilot', str(CHART), '-n', 'pilot-test', '-f', '-'],
input=yaml.safe_dump(values), text=True, capture_output=True, check=False,
)
def pilot():
return {
'pilot': {'enabled': True},
'image': {'digest': DIGEST},
'persistence': {'media': {'enabled': True}, 'appState': {'enabled': True}},
}
class PilotChartTests(unittest.TestCase):
def objects(self, values):
result = render(values)
self.assertEqual(result.returncode, 0, result.stderr)
return list(yaml.safe_load_all(result.stdout))
def test_existing_tag_profile_retains_original_rollout(self):
objects = self.objects({'image': {'tag': 'legacy-test'}})
deployment = next(o for o in objects if o['kind'] == 'Deployment')
self.assertEqual(deployment['spec']['strategy']['type'], 'RollingUpdate')
self.assertFalse(any(o['kind'] == 'PersistentVolumeClaim' for o in objects))
def test_pilot_pins_image_and_separates_durable_state(self):
objects = self.objects(pilot())
deployment = next(o for o in objects if o['kind'] == 'Deployment')
claims = {o['metadata']['name']: o for o in objects
if o['kind'] == 'PersistentVolumeClaim'}
self.assertEqual(len(claims), 2)
self.assertEqual(deployment['spec']['strategy'], {'type': 'Recreate'})
self.assertEqual(deployment['spec']['replicas'], 1)
pod = deployment['spec']['template']['spec']
app = pod['containers'][0]
self.assertTrue(app['image'].endswith('@' + DIGEST))
self.assertEqual({v['name']: v['mountPath'] for v in app['volumeMounts']},
{'media': '/app/media', 'app-state': '/app/.issue-facade'})
for volume in pod['volumes']:
claim = claims[volume['persistentVolumeClaim']['claimName']]
self.assertEqual(claim['metadata']['annotations']['helm.sh/resource-policy'], 'keep')
self.assertEqual(pod['securityContext']['fsGroup'], 999)
def test_pilot_refuses_each_ephemeral_store(self):
for store in ('media', 'appState'):
with self.subTest(store=store):
values = pilot()
values['persistence'][store]['enabled'] = False
result = render(values)
self.assertNotEqual(result.returncode, 0)
self.assertIn('persistent media and appState', result.stderr)
def test_pilot_refuses_mutable_image_and_bad_digest(self):
for image in ({'tag': 'latest', 'digest': ''}, {'digest': 'sha256:invalid'}):
with self.subTest(image=image):
values = pilot()
values['image'] = image
self.assertNotEqual(render(values).returncode, 0)
def test_pilot_refuses_multiple_replicas(self):
values = pilot()
values['replicaCount'] = 2
result = render(values)
self.assertNotEqual(result.returncode, 0)
self.assertIn('exactly one', result.stderr)
def test_restored_claims_are_reused_without_recreation(self):
values = pilot()
for store in ('media', 'appState'):
values['persistence'][store]['existingClaim'] = 'restored-' + store.lower()
objects = self.objects(values)
self.assertFalse(any(o['kind'] == 'PersistentVolumeClaim' for o in objects))
deployment = next(o for o in objects if o['kind'] == 'Deployment')
volumes = deployment['spec']['template']['spec']['volumes']
self.assertEqual({v['persistentVolumeClaim']['claimName'] for v in volumes},
{'restored-media', 'restored-appstate'})
def test_pilot_refuses_operational_state_in_download_volume(self):
for media, state in (('shared-data', 'shared-data'),
('vergabe-teilnahme-app-state', ''),
('', 'vergabe-teilnahme-media')):
with self.subTest(media=media, state=state):
values = pilot()
values['persistence']['media']['existingClaim'] = media
values['persistence']['appState']['existingClaim'] = state
result = render(values)
self.assertNotEqual(result.returncode, 0)
self.assertIn('distinct claims', result.stderr)
if __name__ == '__main__':
unittest.main()