Prepare durable isolated Vergabe pilot deployment and recovery

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_vergabe_pilot_chart.py

@@ -0,0 +1,110 @@
+"""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()