f05ef49c69
Admit vergabe_demo_company as the third declared consumer, track the bounded provisioner and scoped controls, and refresh admission occupancy to 3/3. Overflow remains apps-pg-2 before any fourth. Assistant: grok Assistant-Session: 01a0a23b-3bf0-7341-b4e5-9dc05f72573a
87 lines
4.4 KiB
Python
87 lines
4.4 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('provision', Path(__file__).resolve().parents[1] / 'tools/provision-vergabe-demo.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class DemoProvisioningTests(unittest.TestCase):
|
|
def cluster(self):
|
|
return {'metadata': {'uid': 'synthetic', 'resourceVersion': '7'},
|
|
'spec': {'managed': {'roles': [{'name': 'vergabe', 'connectionLimit': 20},
|
|
{'name': 'coulomb_social', 'connectionLimit': 20}]}}}
|
|
|
|
def test_patch_only_appends_reviewed_third_role_with_revision_tests(self):
|
|
role, _ = m.declarations()
|
|
before = self.cluster()
|
|
result = m.role_patch(before, role)
|
|
self.assertEqual([x['op'] for x in result], ['test', 'test', 'test', 'add'])
|
|
self.assertEqual(result[-1], {'op': 'add', 'path': '/spec/managed/roles/-', 'value': role})
|
|
self.assertEqual(len(before['spec']['managed']['roles']), 2)
|
|
|
|
def test_refuses_fourth_consumer_duplicates_and_relaxed_limits(self):
|
|
role, _ = m.declarations()
|
|
for change in ['fourth', 'duplicate', 'limit']:
|
|
before = self.cluster()
|
|
roles = before['spec']['managed']['roles']
|
|
if change == 'fourth':
|
|
roles.append({'name': 'another', 'connectionLimit': 20})
|
|
elif change == 'duplicate':
|
|
roles.append(roles[0])
|
|
else:
|
|
roles[0]['connectionLimit'] = -1
|
|
with self.subTest(change=change), self.assertRaises(m.Refused):
|
|
m.role_patch(before, role)
|
|
|
|
def test_exact_owned_role_is_noop_but_conflict_refused(self):
|
|
role, _ = m.declarations()
|
|
before = self.cluster()
|
|
before['spec']['managed']['roles'].append(dict(role))
|
|
self.assertIsNone(m.role_patch(before, role))
|
|
before['spec']['managed']['roles'][-1]['superuser'] = True
|
|
with self.assertRaises(m.Refused):
|
|
m.role_patch(before, role)
|
|
|
|
def test_existing_secret_requires_owner_and_exact_keys(self):
|
|
obj = {'metadata': {'labels': dict(m.OWNER)}, 'data': {'username': 'x', 'password': 'x'}}
|
|
m.valid_owned_secret(obj, ['username', 'password'])
|
|
for changed in [dict(obj, data=dict(obj['data'], extra='x')),
|
|
dict(obj, metadata={'labels': {}}),
|
|
dict(obj, metadata={'labels': m.OWNER, 'ownerReferences': [{'uid': 'other'}]})]:
|
|
with self.assertRaises(m.Refused):
|
|
m.valid_owned_secret(changed, ['username', 'password'])
|
|
|
|
def test_sensitive_child_error_is_not_forwarded(self):
|
|
result = SimpleNamespace(returncode=1, stdout='synthetic-secret', stderr='synthetic-secret')
|
|
with patch.object(m.subprocess, 'run', return_value=result), self.assertRaises(m.Refused) as caught:
|
|
m.kube(['create', '-f', '-'], {'stringData': {'password': 'synthetic-secret'}})
|
|
self.assertEqual(str(caught.exception), 'kubernetes_operation_failed')
|
|
|
|
def test_credentials_travel_on_stdin_not_arguments(self):
|
|
result = SimpleNamespace(returncode=0, stdout='{}', stderr='')
|
|
with patch.object(m.subprocess, 'run', return_value=result) as run:
|
|
m.kube(['create', '-f', '-'], {'stringData': {'password': 'synthetic-secret'}})
|
|
args, kwargs = run.call_args
|
|
self.assertNotIn('synthetic-secret', str(args))
|
|
self.assertIn('synthetic-secret', kwargs['input'])
|
|
self.assertTrue(kwargs['capture_output'])
|
|
|
|
def test_source_declarations_match_tracked_cluster_and_database_manifests(self):
|
|
import yaml
|
|
role, database = m.declarations()
|
|
cluster = yaml.safe_load((m.ROOT / 'helm/apps-pg-cluster.yaml').read_text())
|
|
tracked = [item for item in cluster['spec']['managed']['roles'] if item['name'] == m.NAME]
|
|
self.assertEqual(tracked, [role])
|
|
documents = list(yaml.safe_load_all((m.ROOT / 'helm/apps-pg-databases.yaml').read_text()))
|
|
named = next(item for item in documents if item['metadata']['name'] == 'vergabe-demo-company-db')
|
|
self.assertEqual(named['spec']['name'], database['spec']['name'])
|
|
self.assertEqual(named['spec']['owner'], database['spec']['owner'])
|
|
self.assertEqual(named['spec']['cluster'], database['spec']['cluster'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|