Build local product stack with Vault-managed secrets (#42)

dev/test_local.py

@@ -0,0 +1,62 @@
+"""Focused regression tests for release selection and bootstrap failure paths."""
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import bootstrap
+import images
+from vault import Vault
+
+
+class Releases(unittest.TestCase):
+    def test_selects_exact_stable_digest_not_prerelease(self):
+        digest = 'sha256:' + 'a' * 64
+        versions = [{'name': name, 'digest': digest} for name in ['latest', '2', '2.1', '2.1.0', '2.2.0-rc1']]
+        with patch.object(images, 'get', side_effect=[{'digest': digest}, {'results': versions}]):
+            self.assertEqual(images.image('hashicorp/vault', 'latest'), 'hashicorp/vault:2.1.0@' + digest)
+
+    def test_missing_stable_mapping_fails(self):
+        digest = 'sha256:' + 'a' * 64
+        with patch.object(images, 'get', side_effect=[{'digest': digest}, {'results': [{'name': '3.0.0-rc1', 'digest': digest}]}]):
+            with self.assertRaises(ValueError):
+                images.image('hashicorp/vault', 'latest')
+
+    def test_upstream_failure_does_not_replace_existing_pins(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            state = Path(temporary)
+            target = state / 'images.env'
+            target.write_text('previous build')
+            with patch.object(images, 'STATE', state), patch.object(images, 'get', side_effect=OSError('offline')):
+                with self.assertRaises(OSError):
+                    images.main()
+            self.assertEqual(target.read_text(), 'previous build')
+
+
+class Bootstrap(unittest.TestCase):
+    def test_missing_resend_credentials_stops_before_product_start(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            state = Path(temporary)
+            (state / 'images.env').touch()
+            from unittest.mock import Mock
+            vault = Mock()
+            vault.kv.return_value = {'driver': 'resend', 'api_key': '', 'sender_address': 'notifications@localhost.test'}
+            with patch.object(bootstrap, 'STATE', state), patch.object(bootstrap, 'compose') as compose, \
+                 patch.object(bootstrap, 'provision_mongo'), patch.object(bootstrap, 'provision_agents'):
+                with self.assertRaisesRegex(RuntimeError, 'Resend api_key'):
+                    bootstrap.up(vault)
+            self.assertEqual(compose.call_args_list[-1].args[-1], 'nats')
+            email_seed = next(c for c in vault.ensure.call_args_list if c.args[0] == 'email').args[1]()
+            self.assertEqual(email_seed['driver'], 'resend')
+            self.assertEqual(email_seed['api_key'], '')
+
+    def test_initialized_vault_without_local_keys_fails(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            with patch('vault.STATE', Path(temporary)), patch.object(Vault, 'call', return_value={'initialized': True}):
+                with self.assertRaisesRegex(RuntimeError, 'Missing .local'):
+                    Vault().initialize()
+
+
+
+if __name__ == '__main__':
+    unittest.main()