Deploy achievement-safe Factorio 2.0.77 with strict placement, monitoring and limited CD

tests/test_placement.py

@@ -0,0 +1,46 @@
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import time
+import unittest
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'scripts/placement.sh'
+
+class PlacementTests(unittest.TestCase):
+    def run_case(self, status='True', age=600, missing=False, unavailable=False, cordoned=False, fallback=False, error=False):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            now = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(time.time()-age))
+            node = {'metadata': {'labels': {'factorio.brunner.ninja/unavailable': str(unavailable).lower()}}, 'spec': {'unschedulable': cordoned}, 'status': {'conditions':[{'type':'Ready','status':status,'lastTransitionTime':now}]}}
+            nodes = ['odroid-h3-'+str(i).zfill(2) for i in range(6)] if fallback else ['arschrock']
+            deployment={'metadata':{'resourceVersion':'123'},'spec':{'template':{'spec':{'affinity':{'nodeAffinity':{'requiredDuringSchedulingIgnoredDuringExecution':{'nodeSelectorTerms':[{'matchExpressions':[{'values':nodes}]}]}}}}}}}
+            (root/'node').write_text('' if missing else json.dumps(node))
+            (root/'deployment').write_text(json.dumps(deployment))
+            fake=root/'kubectl'
+            fake.write_text('#!/bin/sh\ncase "$*" in\n *"get node"*) '+('exit 1' if error else 'cat "$FIXTURES/node"')+' ;;\n *"get deployment"*) cat "$FIXTURES/deployment" ;;\n *"patch deployment"*) printf "%s\\n" "$*" > "$FIXTURES/patch" ;;\n *) exit 99 ;;\nesac\n')
+            fake.chmod(0o755)
+            result=subprocess.run(['sh',str(SCRIPT)],env={**os.environ,'PATH':tmp+':'+os.environ['PATH'],'FIXTURES':tmp},capture_output=True,text=True)
+            return result.returncode,(root/'patch').read_text() if (root/'patch').exists() else ''
+
+    def test_healthy_primary_stays(self):
+        self.assertEqual(self.run_case(),(0,''))
+    def test_notready_fails_over_after_grace(self):
+        code,patch=self.run_case(status='False'); self.assertEqual(code,0); self.assertIn('odroid-h3-05',patch)
+    def test_unreachable_fails_over(self):
+        self.assertIn('odroid-h3-00',self.run_case(status='Unknown')[1])
+    def test_missing_primary_fails_over(self):
+        self.assertIn('odroid-h3-00',self.run_case(missing=True)[1])
+    def test_transient_failure_does_not_move_world(self):
+        self.assertEqual(self.run_case(status='False',age=30),(0,''))
+    def test_recovery_waits_for_stability(self):
+        self.assertEqual(self.run_case(age=30,fallback=True),(0,''))
+    def test_stable_recovery_returns_to_primary(self):
+        self.assertIn('arschrock',self.run_case(fallback=True)[1])
+    def test_maintenance_moves_immediately(self):
+        self.assertIn('odroid-h3-00',self.run_case(unavailable=True,age=5)[1])
+    def test_cordon_moves_immediately(self):
+        self.assertIn('odroid-h3-00',self.run_case(cordoned=True)[1])
+    def test_api_error_is_not_a_missing_node(self):
+        code,patch=self.run_case(error=True); self.assertNotEqual(code,0); self.assertEqual(patch,'')