"""Ownership-only admission never reads allocation prerequisites or unrelated network details.""" import copy import json import unittest from network_contract import OWNER, SCHEMA, existing, plans, verify_created from network_owned_inventory import LISTING, ownership_inventory from network_pools import InventoryFailure from network_test_fixture import config, network class ScopedHost: def __init__(self): self.rows = [network('first', '172.18.0.0/24'), network('unrelated', '172.19.0.0/24')] self.calls, self.listings = [], 0 self.listing_override, self.inspection_override, self.after = None, None, None def __call__(self, argv, seconds=10): self.calls.append(list(argv)) if argv == LISTING: self.listings += 1 if self.listing_override is not None: if isinstance(self.listing_override, Exception): raise self.listing_override return self.listing_override rows = self.after if self.listings > 1 and self.after is not None else self.rows return b'\n'.join(json.dumps({key: row[key] for key in ('Id', 'Name')}).encode() for row in rows) if argv[:3] == ['docker', 'network', 'inspect']: rows = [row for row in self.rows if row['Id'] in argv[3:]] if self.inspection_override is not None: rows = self.inspection_override return json.dumps(rows).encode() raise AssertionError('Ownership read attempted an allocation-only command') class OwnershipInventoryTests(unittest.TestCase): def test_only_exact_requested_network_is_inspected_and_private_fields_are_discarded(self): host = ScopedHost() host.rows[1]['IPAM'] = {'Driver': 'unknown-custom-plugin', 'Config': 'unknown-private-schema'} host.rows[0]['Labels']['example.test'] = 'authored-value' host.rows[0]['Options']['example.option'] = 'retained' host.rows[0]['Containers'] = {'PRIVATE_SENTINEL': {'Name': 'PRIVATE_SENTINEL'}} host.rows[0]['ExtraPrivateData'] = 'PRIVATE_SENTINEL' host.rows[0]['IPAM']['Config'][0]['Gateway'] = '172.18.0.1' selected = plans(config('first', {'labels': {'example.test': 'authored-value'}}), 'first') host.rows[0]['Labels'].update({OWNER: selected[0]['contract'], SCHEMA: '1'}) result = ownership_inventory(host, selected) self.assertEqual(existing(selected[0], result['networks'])['Id'], host.rows[0]['Id']) verify_created(selected[0], result['networks'][0], '172.18.0.0/24') for key in ('Labels', 'Options', 'IPAM'): self.assertEqual(result['networks'][0][key], host.rows[0][key]) self.assertNotIn('PRIVATE_SENTINEL', json.dumps(result)) self.assertEqual(host.calls, [LISTING, ['docker', 'network', 'inspect', host.rows[0]['Id']], LISTING]) def test_authored_and_external_custom_ipam_driver_does_not_require_allocation_support(self): for settings in ({'external': True}, {'ipam': {'driver': 'private-plugin'}}): host = ScopedHost() host.rows[0]['IPAM']['Driver'] = 'private-plugin' selected = plans(config('first', settings), 'first') result = ownership_inventory(host, selected) self.assertEqual(existing(selected[0], result['networks'])['Id'], host.rows[0]['Id']) def test_missing_and_empty_requested_sets_recheck_without_inspecting_unrelated_networks(self): for selected in ([], plans(config('absent'), 'absent')): host = ScopedHost() self.assertEqual(ownership_inventory(host, selected), {'networks': []}) self.assertEqual(host.calls, [LISTING, LISTING]) def test_malformed_and_duplicate_listing_records_refuse(self): row = {'Id': 'a' * 64, 'Name': 'first_default'} cases = [b'not-json', b'{}', json.dumps({**row, 'Id': 'short'}).encode(), json.dumps({**row, 'Name': 'bad\nname'}).encode(), (json.dumps(row) + '\n' + json.dumps(row)).encode(), (json.dumps(row) + '\n' + json.dumps({**row, 'Id': 'b' * 64})).encode(), b'{"Id":"a","Id":"b","Name":"first_default"}'] for raw in cases: host = ScopedHost() host.listing_override = raw with self.subTest(raw=raw), self.assertRaises(InventoryFailure): ownership_inventory(host, plans(config('first'), 'first')) self.assertEqual(len(host.calls), 1) def test_incomplete_mismatching_and_malformed_selected_inspection_refuse(self): original = network('first', '172.18.0.0/24') cases = [[], [original, original], [{**original, 'Name': 'another_default'}], [{**original, 'Id': 'b' * 64}], [{**original, 'Labels': {'bad': 1}}], [{**original, 'IPAM': {'Config': [{'Subnet': 'invalid'}]}}]] for rows in cases: host = ScopedHost() host.inspection_override = rows with self.subTest(rows=rows), self.assertRaises(InventoryFailure): ownership_inventory(host, plans(config('first'), 'first')) def test_selected_name_removal_replacement_and_addition_refuse(self): for change in ('removed', 'renamed', 'replaced', 'added'): host = ScopedHost() host.after = copy.deepcopy(host.rows) selected = plans(config('first'), 'first') if change == 'removed': host.after.pop(0) elif change == 'renamed': host.after[0]['Name'] = 'renamed_default' elif change == 'replaced': host.after[0]['Id'] = 'b' * 64 else: selected = plans(config('absent'), 'absent') host.after.append(network('absent', '172.20.0.0/24')) with self.subTest(change=change), self.assertRaisesRegex( InventoryFailure, 'network_inventory_changed'): ownership_inventory(host, selected) def test_unrelated_name_churn_does_not_block_unchanged_requested_ownership(self): host = ScopedHost() host.after = [host.rows[0], network('new-unrelated', '172.21.0.0/24')] result = ownership_inventory(host, plans(config('first'), 'first')) self.assertEqual(result['networks'][0]['Id'], host.rows[0]['Id']) def test_failed_and_excessive_command_output_produce_only_fixed_refusals(self): for result in (RuntimeError('PRIVATE_SENTINEL'), b'x' * (1024 * 1024 + 1)): host = ScopedHost() host.listing_override = result with self.subTest(kind=type(result)), self.assertRaises(InventoryFailure) as caught: ownership_inventory(host, plans(config('first'), 'first')) self.assertNotIn('PRIVATE_SENTINEL', str(caught.exception)) if __name__ == '__main__': unittest.main()