63 lines
2.9 KiB
Python
63 lines
2.9 KiB
Python
"""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()
|