mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-09-16 03:36:44 +00:00
272 lines
16 KiB
Python
272 lines
16 KiB
Python
"""PR integration regressions. No network, guests, sidecars or notifications."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from test_update_method_choice import apps, routes
|
|
|
|
|
|
ITEM = {'api_host': 'ghcr.io', 'repository': 'example/app',
|
|
'reference': 'ghcr.io/example/app:latest', 'used_by': ['example'], 'standalone_containers': ['example'],
|
|
'remote_digest': 'sha256:' + 'a' * 64,
|
|
'installed_version': '1.0.0', 'update_available': True,
|
|
'available_version_source': 'remote_metadata_pending',
|
|
'installed_version_source': 'image_label:org.opencontainers.image.version',
|
|
'platform': {'os': 'linux', 'architecture': 'amd64', 'variant': ''}}
|
|
LABELS = {'org.opencontainers.image.version': '2.0.0'}
|
|
APP = {'id': 'a1', 'name': 'Example', 'installed_via': 'docker_exec',
|
|
'container_name': 'example', 'binary_path': '/app',
|
|
'binary_args': ['--version'], 'installed_regex': r'(\d+\.\d+\.\d+)',
|
|
'update_via': 'docker'}
|
|
|
|
|
|
class RegistryRegressions(unittest.TestCase):
|
|
def setUp(self):
|
|
apps._docker_remote_config_cache.clear()
|
|
|
|
def test_container_reference_does_not_follow_unused_aliases(self):
|
|
self.assertEqual(apps._docker_reference_identity('redis'), apps._docker_reference_identity('docker.io/library/redis:latest'))
|
|
self.assertNotEqual(apps._docker_reference_identity('redis:7-alpine'), apps._docker_reference_identity('redis:7.2.4-alpine'))
|
|
self.assertIsNone(apps._docker_reference_identity('redis@sha256:' + 'a' * 64))
|
|
self.assertIsNone(apps._docker_reference_identity('sha256:' + 'a' * 64))
|
|
|
|
def test_inventory_tracks_running_image_and_ignores_unused_alias_tag(self):
|
|
repo = 'ghcr.io/demo/app'
|
|
old_id, new_id = 'sha256:' + '1'*64, 'sha256:' + '2'*64
|
|
old_digest, new_digest = 'sha256:' + 'a'*64, 'sha256:' + 'b'*64
|
|
container = {'Name': '/example', 'Image': old_id, 'Config': {'Image': repo + ':latest', 'Labels': {}}}
|
|
image = lambda image_id, digest, version: {'Id': image_id, 'RepoDigests': [repo+'@'+digest],
|
|
'Config': {'Labels': {'org.opencontainers.image.version': version}}, 'Os': 'linux', 'Architecture': 'amd64'}
|
|
for pulled in (False, True):
|
|
def execute(vmid, argv, **kwargs):
|
|
if argv[:2] == ['docker', 'version']:
|
|
return 0, '26.1.5', ''
|
|
if argv[:2] == ['docker', 'ps']:
|
|
return 0, f'example\t{repo}:latest\tUp 1 minute', ''
|
|
if argv[:2] == ['docker', 'inspect']:
|
|
return 0, json.dumps(container), ''
|
|
if argv[:3] == ['docker', 'image', 'ls']:
|
|
return 0, '\n'.join([
|
|
f'{repo}\tlatest\t{new_digest if pulled else old_digest}\t{new_id if pulled else old_id}',
|
|
f'{repo}\t1.0.0\t{old_digest}\t{old_id}',
|
|
]), ''
|
|
if argv[:3] == ['docker', 'image', 'inspect']:
|
|
return 0, '\n'.join(json.dumps(item) for item in [image(old_id, old_digest, '1.0.0'), image(new_id, new_digest, '2.0.0')]), ''
|
|
raise AssertionError(argv)
|
|
with patch.object(apps, '_pct_exec', side_effect=execute), \
|
|
patch.object(apps, '_docker_service_catalog_meta', return_value={}), \
|
|
patch.object(apps, '_fetch_registry_manifest_digest', return_value=(new_digest, None)):
|
|
inventory = apps._docker_inventory_from_ct(101)
|
|
self.assertEqual(len(inventory['images']), 1, 'unused alias is not a workload')
|
|
actual = inventory['images'][0]
|
|
self.assertEqual(actual['reference'], repo+':latest')
|
|
self.assertEqual(actual['installed_version'], '1.0.0', 'a pull alone does not update a container')
|
|
self.assertEqual(actual['image_id'], old_id)
|
|
self.assertEqual(actual['local_digest'], old_digest)
|
|
self.assertTrue(actual['update_available'])
|
|
|
|
def test_transient_error_does_not_poison_retry(self):
|
|
with patch.object(apps, '_remote_image_config_labels', side_effect=[(None, 'timeout'), (LABELS, None)]) as read:
|
|
self.assertEqual(apps._docker_available_version_from_registry(ITEM)[1], 'remote_fetch_error')
|
|
self.assertEqual(apps._docker_available_version_from_registry(ITEM)[0], '2.0.0')
|
|
self.assertEqual(apps._docker_available_version_from_registry(ITEM)[0], '2.0.0')
|
|
self.assertEqual(read.call_count, 2)
|
|
|
|
def test_cache_key_includes_os_and_variant(self):
|
|
with patch.object(apps, '_remote_image_config_labels', side_effect=[({'Version': str(i)}, None) for i in range(3)]) as read:
|
|
for i, platform in enumerate([
|
|
{'os': 'linux', 'architecture': 'arm', 'variant': 'v7'},
|
|
{'os': 'linux', 'architecture': 'arm', 'variant': 'v6'},
|
|
{'os': 'other', 'architecture': 'arm', 'variant': 'v6'},
|
|
]):
|
|
self.assertEqual(apps._fetch_remote_image_config_labels(ITEM, ITEM['remote_digest'], platform)[0], {'Version': str(i)})
|
|
self.assertEqual(read.call_count, 3)
|
|
|
|
def test_concurrent_identical_images_only_fetch_once(self):
|
|
import concurrent.futures
|
|
def read(*_):
|
|
time.sleep(.02)
|
|
return LABELS, None
|
|
with patch.object(apps, '_remote_image_config_labels', side_effect=read) as fetch:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(apps._docker_available_version_from_registry, [ITEM] * 4))
|
|
self.assertEqual(fetch.call_count, 1)
|
|
self.assertTrue(all(result[0] == '2.0.0' for result in results))
|
|
|
|
def test_valid_digest_with_non_object_document_is_rejected(self):
|
|
for payload in ([], None, 'text', 42):
|
|
body = json.dumps(payload).encode()
|
|
digest = 'sha256:' + hashlib.sha256(body).hexdigest()
|
|
with patch.object(apps, '_registry_request', return_value=({}, body, None, None)):
|
|
document, _, error = apps._registry_get_document('https://example.test', digest, {}, None)
|
|
self.assertIsNone(document)
|
|
self.assertIn('not an object', error)
|
|
|
|
def test_malformed_nested_configs_do_not_escape(self):
|
|
for payload in ({'config': []}, {'config': {'Labels': ['bad']}}, {'config': {'Labels': {'version': {}}}}):
|
|
body = json.dumps(payload).encode()
|
|
digest = 'sha256:' + hashlib.sha256(body).hexdigest()
|
|
manifest = {'config': {'mediaType': 'application/vnd.oci.image.config.v1+json', 'digest': digest}}
|
|
with patch.object(apps, '_registry_get_document', return_value=(manifest, None, None)), \
|
|
patch.object(apps, '_registry_request', return_value=({}, body, None, None)):
|
|
labels, error = apps._remote_image_config_labels(ITEM, ITEM['remote_digest'], ITEM['platform'])
|
|
self.assertIsNone(labels)
|
|
self.assertTrue(error)
|
|
|
|
def test_redirect_authentication_failure_is_not_success(self):
|
|
with patch.object(apps, '_registry_open', side_effect=[
|
|
({}, None, 307, 'https://cdn.example.test/blob', None),
|
|
({}, None, 401, None, None),
|
|
]) as request:
|
|
self.assertEqual(apps._registry_request('https://registry.example.test', {}, token='secret')[3], 'registry HTTP 401')
|
|
self.assertNotIn('Authorization', request.call_args.args[1])
|
|
|
|
def test_no_network_on_cached_inventory_reads(self):
|
|
with patch.object(apps, '_docker_inventory_cache', {'101': {'available': True, 'checked_at_unix': time.time(), 'images': [ITEM]}}), \
|
|
patch.object(apps, '_remote_image_config_labels', side_effect=AssertionError('network')), \
|
|
patch.object(apps, '_docker_inventory_from_ct', side_effect=AssertionError('scan')):
|
|
self.assertEqual(len(apps.get_docker_inventory(101)['images']), 1)
|
|
self.assertEqual(len(apps.get_cached_docker_inventories()['101']['images']), 1)
|
|
|
|
def test_optional_metadata_does_not_block_or_overwrite_a_new_boot(self):
|
|
started, release = threading.Event(), threading.Event()
|
|
futures = []
|
|
import concurrent.futures
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
|
def submit(*args):
|
|
future = pool.submit(*args)
|
|
futures.append(future)
|
|
return future
|
|
def resolve(_):
|
|
started.set()
|
|
self.assertTrue(release.wait(2))
|
|
return '2.0.0', 'remote_image_label:version'
|
|
snapshot = {'available': True, 'images': [dict(ITEM)]}
|
|
new_boot = {'available': False, 'refreshing': True, 'images': []}
|
|
with patch.object(apps, '_docker_inventory_cache', {'101': snapshot}), \
|
|
patch.object(apps, '_docker_metadata_pool', types.SimpleNamespace(submit=submit)), \
|
|
patch.object(apps, '_docker_available_version_from_registry', side_effect=resolve):
|
|
try:
|
|
apps._queue_docker_metadata('101', snapshot)
|
|
self.assertTrue(started.wait(1))
|
|
self.assertIs(apps._docker_inventory_cache['101'], snapshot)
|
|
apps._docker_inventory_cache['101'] = new_boot
|
|
finally:
|
|
release.set()
|
|
for future in futures:
|
|
future.result(2)
|
|
self.assertIs(apps._docker_inventory_cache['101'], new_boot)
|
|
self.assertNotIn('available_version', snapshot['images'][0])
|
|
|
|
def test_failed_image_does_not_prevent_other_image_metadata(self):
|
|
import concurrent.futures
|
|
futures = []
|
|
snapshot = {'available': True, 'checked_at': 'original-scan', 'images': [dict(ITEM), dict(ITEM)]}
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
|
def submit(*args):
|
|
future = pool.submit(*args)
|
|
futures.append(future)
|
|
return future
|
|
with patch.object(apps, '_docker_inventory_cache', {'101': snapshot}), \
|
|
patch.object(apps, '_docker_metadata_pool', types.SimpleNamespace(submit=submit)), \
|
|
patch.object(apps, '_docker_available_version_from_registry', side_effect=[ValueError('bad image'), ('2.0.0', 'remote_image_label:version')]):
|
|
apps._queue_docker_metadata('101', snapshot)
|
|
for future in futures:
|
|
future.result(2)
|
|
self.assertEqual(snapshot['images'][0]['available_version_source'], 'remote_fetch_error')
|
|
self.assertEqual(snapshot['images'][1]['available_version'], '2.0.0')
|
|
self.assertTrue(snapshot['images'][0]['update_available'])
|
|
self.assertEqual(snapshot['checked_at'], 'original-scan')
|
|
|
|
|
|
class DelegationRegressions(unittest.TestCase):
|
|
def test_delegated_label_reads_actual_image_not_preserved_container_label(self):
|
|
image_id = 'sha256:' + 'a' * 64
|
|
config = {**APP, 'installed_via': 'docker_label', 'label': 'org.opencontainers.image.version'}
|
|
with patch.object(apps, '_pct_exec', side_effect=[(0, image_id, ''), (0, '2.0.0', '')]) as execute:
|
|
self.assertEqual(apps.detect_installed_version(101, config), ('2.0.0', None))
|
|
self.assertEqual(execute.call_args.args[1][:3], ['docker', 'image', 'inspect'])
|
|
self.assertEqual(execute.call_args.args[1][-1], image_id)
|
|
with patch.object(apps, '_pct_exec', return_value=(0, '1.0.0', '')) as execute:
|
|
self.assertEqual(apps.detect_installed_version(101, {**config, 'update_via': ''}), ('1.0.0', None))
|
|
self.assertEqual(execute.call_count, 1, 'existing non-delegated detector semantics remain unchanged')
|
|
|
|
def inventory(self):
|
|
return {'available': True, 'images': [dict(ITEM), {'reference': 'other:latest', 'used_by': ['other']}],
|
|
'update_units': [
|
|
{'id': 'docker-unit:aaaaaaaaaaaaaaaaaaaa', 'kind': 'standalone', 'references': [ITEM['reference']],
|
|
'standalone_containers': ['example']},
|
|
{'id': 'docker-unit:bbbbbbbbbbbbbbbbbbbb', 'kind': 'standalone', 'references': ['other:latest'],
|
|
'standalone_containers': ['other']}]}
|
|
|
|
def test_missing_inventory_clears_old_decoration(self):
|
|
records = [{**APP, 'docker_available_version': '2.0.0', 'docker_update_available': True}]
|
|
apps.annotate_delegated_apps(records, None)
|
|
self.assertIsNone(records[0]['docker_available_version'])
|
|
self.assertIsNone(records[0]['docker_update_available'])
|
|
self.assertEqual(records[0]['docker_binding_error'], 'inventory_unavailable')
|
|
|
|
def test_delegated_and_custom_or_helper_cannot_both_be_selected(self):
|
|
for extra in ({'update_command': '/opt/update.sh'}, {'update_method': 'helper', 'helper_slug': 'vaultwarden'}):
|
|
self.assertFalse(apps.validate_config({**APP, **extra})[0])
|
|
self.assertTrue(apps.validate_config({**APP, 'update_via': '', 'update_command': '/opt/update.sh'})[0])
|
|
|
|
def test_only_followed_workloads_are_offered(self):
|
|
inventory = self.inventory()
|
|
scoped = apps.docker_inventory_for_apps([APP], inventory)
|
|
self.assertEqual([image['reference'] for image in scoped['images']], [ITEM['reference']])
|
|
self.assertEqual([unit['id'] for unit in scoped['update_units']], ['docker-unit:aaaaaaaaaaaaaaaaaaaa'])
|
|
self.assertEqual(len(inventory['images']), 2)
|
|
|
|
def test_notifications_without_engine_registration_honor_opt_out_and_dedup(self):
|
|
manager = Mock()
|
|
inventory = self.inventory()
|
|
records = [dict(APP), {**APP, 'id': 'a2'}]
|
|
with patch.object(apps, 'get_cached_docker_inventories', return_value={'101': inventory}), \
|
|
patch.object(apps, '_read_sidecar', return_value={'apps': records}), \
|
|
patch.dict(sys.modules, {'notification_manager': types.SimpleNamespace(notification_manager=manager)}):
|
|
self.assertEqual(apps.emit_all_pending_docker_stacks(), 1)
|
|
first = manager.emit_event.call_args.kwargs
|
|
self.assertEqual(first['data']['count'], 1)
|
|
inventory['images'][0]['available_version'] = '2.0.0'
|
|
self.assertEqual(apps.emit_all_pending_docker_stacks(), 1)
|
|
self.assertEqual(first['entity_id'], manager.emit_event.call_args.kwargs['entity_id'])
|
|
for record in records:
|
|
record['notifications_enabled'] = False
|
|
self.assertEqual(apps.emit_all_pending_docker_stacks(), 0)
|
|
|
|
def test_bulk_allows_followed_image_but_not_engine_or_other_image(self):
|
|
api = routes()
|
|
with patch.object(apps, 'load_sidecar', return_value={'apps': [APP]}), \
|
|
patch.object(apps, '_read_sidecar', return_value={'apps': [APP]}), \
|
|
patch.object(apps, 'get_docker_inventory', return_value=self.inventory()):
|
|
plan = api['_resolve_bulk_update_plan'](101, ['os', 'docker-unit:aaaaaaaaaaaaaaaaaaaa'])
|
|
self.assertTrue(plan['ok'], plan)
|
|
self.assertFalse(plan['unavailable'], plan)
|
|
self.assertIn('example', plan['docker_standalone_targets'])
|
|
plan = api['_resolve_bulk_update_plan'](101, ['os', 'docker-unit:bbbbbbbbbbbbbbbbbbbb', 'docker-engine'])
|
|
self.assertEqual(len(plan['unavailable']), 2, plan)
|
|
|
|
def test_schedule_runs_only_selected_followed_container(self):
|
|
api = routes()
|
|
with patch.object(apps, '_read_sidecar', return_value={'apps': [APP]}), \
|
|
patch.object(apps, 'load_sidecar', return_value={'apps': [APP]}), \
|
|
patch.object(apps, 'get_docker_inventory', return_value=self.inventory()):
|
|
result = api['_run_scheduled_update'](101, {'targets': ['docker-container:example', 'docker-container:other', 'docker-engine']})
|
|
self.assertEqual(result['status'], 'partial', result)
|
|
self.assertEqual(result['executed_targets'], ['docker-container:example'])
|
|
env = api['subprocess'].run.call_args.kwargs['env']
|
|
self.assertEqual(env['DOCKER_STANDALONE_TARGETS'], 'example')
|
|
self.assertEqual(env['UPDATE_DOCKER_ENGINE'], '0')
|
|
self.assertEqual(env['RUN_HELPER'], '0')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|