mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
MacRimi
89fdfd0952
- Notification settings show the last 20 notifications with the result of each send, the error when one failed and every daily digest run, also on days with nothing to send; the history is read without the flag the API never returned, and the page does not offer to erase it, as Audit & Report reads it. - The Ctrl menu of the three web terminals sends Ctrl+D, which ends a text pasted in the terminal, also from a phone or a tablet. - The repository check lists every Proxmox VE or Enterprise Ceph repository whose suite does not match the installed version, with its channel and file, and says where to disable it; repository addresses are never shown. - Adding an NVIDIA GPU to a VM writes its VFIO udev rule again when it is missing or was edited, and the pre-check counts a missing rule as a pending host change. - New texts in English and Spanish, modelled in the translation steady-state test, and the notifications docs page describes the history section. - Corrected the automatic translations of the new LXC update and power control texts, and the start, shutdown, reboot and stop buttons of VMs and LXCs in German, French, Portuguese and Swedish. - Tests for the history panel, the terminal Ctrl+D, the repository message and the VFIO rule.
770 lines
43 KiB
Python
770 lines
43 KiB
Python
"""Fixture-only policy checks: never call host pvesh or edit host APT sources."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
POLICY = ROOT / 'scripts/global/repository_policy.py'
|
|
|
|
|
|
def repo(components, *, uri='https://mirror.example/debian/pve', suite='trixie', enabled=True):
|
|
return {'Types': ['deb'], 'URIs': [uri], 'Suites': [suite],
|
|
'Components': components.split(), 'Enabled': enabled}
|
|
|
|
|
|
def fixture(*entries):
|
|
files = {}
|
|
for path, entries_for_file in entries:
|
|
files[path] = {'path': path, 'repositories': entries_for_file}
|
|
return {'files': list(files.values()), 'errors': [], 'digest': 'fixture-digest',
|
|
'standard-repos': [{'handle': 'no-subscription', 'name': 'PVE No-Subscription'}]}
|
|
|
|
|
|
class RepositoryPolicyTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
spec = importlib.util.spec_from_file_location('repository_policy', POLICY)
|
|
cls.module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(cls.module)
|
|
|
|
def setUp(self):
|
|
self.calls = []
|
|
self.mutate = True
|
|
self.after_write = lambda: None
|
|
self.writes = 0
|
|
self.subscription = 'status: notfound\nmessage: There is no subscription key\n'
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/custom.sources', [repo('pve-enterprise')]),
|
|
('/etc/apt/sources.list.d/debian.sources', [repo('main', uri='https://deb.example/debian')]),
|
|
)
|
|
|
|
def run_policy(self, apply=False, suite='trixie'):
|
|
def run(args):
|
|
self.calls.append(args)
|
|
if args[:2] == ['pvesubscription', 'get']:
|
|
if isinstance(self.subscription, Exception):
|
|
raise ValueError('service unavailable') from self.subscription
|
|
return self.subscription
|
|
if args[:2] == ['pvesh', 'get']:
|
|
return json.dumps(self.repositories)
|
|
if args[1] in ('create', 'set'):
|
|
self.assertEqual(args[args.index('--digest') + 1], self.repositories['digest'])
|
|
self.writes += 1
|
|
if self.mutate and args[:2] == ['pvesh', 'create']:
|
|
path = args[args.index('--path') + 1]
|
|
index = int(args[args.index('--index') + 1])
|
|
next(f for f in self.repositories['files'] if f['path'] == path)['repositories'][index]['Enabled'] = False
|
|
if self.mutate and args[:2] == ['pvesh', 'set']:
|
|
self.repositories['files'].append({'path': '/etc/apt/sources.list.d/proxmox.sources',
|
|
'repositories': [repo('pve-no-subscription')]})
|
|
self.repositories['digest'] = f'write-{self.writes}'
|
|
self.after_write()
|
|
return ''
|
|
with patch.object(self.module, 'run', side_effect=run):
|
|
return self.module.evaluate(suite, apply=apply)
|
|
|
|
def test_authentic_pve92_public_capture_preserved_without_subscription_or_writes(self):
|
|
capture = (ROOT / 'tests/fixtures/pr407-macrimi-pve9.2-no-subscription.json').read_bytes()
|
|
for apply in (False, True):
|
|
with self.subTest(apply=apply):
|
|
self.setUp()
|
|
self.repositories = json.loads(capture)
|
|
self.subscription = RuntimeError('must not query subscription')
|
|
before = json.dumps(self.repositories)
|
|
self.assertEqual(self.run_policy(apply=apply), 'preserve')
|
|
self.assertEqual(json.dumps(self.repositories), before)
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual([c[:2] for c in self.calls], [['pvesh', 'get']])
|
|
|
|
def simulated_enterprise_capture(self):
|
|
# GENERATED scenario, NOT a maintainer before capture: remove public PVE,
|
|
# enable Enterprise PVE/Ceph, and omit their serialized Enabled option.
|
|
data = json.loads((ROOT / 'tests/fixtures/pr407-macrimi-pve9.2-no-subscription.json').read_bytes())
|
|
data['files'] = [file for file in data['files']
|
|
if file['path'] != '/etc/apt/sources.list.d/proxmox.sources']
|
|
for file in data['files']:
|
|
for row in file['repositories']:
|
|
if row['Components'] in (['pve-enterprise'], ['enterprise']):
|
|
row['Enabled'] = 1
|
|
row['Options'] = [option for option in row['Options']
|
|
if option['Key'] != 'Enabled']
|
|
data['digest'] = 'simulated-enterprise-before'
|
|
return data
|
|
|
|
def test_simulated_capture_disable_accepts_equivalent_enabled_option_serialization(self):
|
|
for explicit_before in (False, True):
|
|
with self.subTest(explicit_before=explicit_before):
|
|
self.setUp()
|
|
self.repositories = self.simulated_enterprise_capture()
|
|
if explicit_before:
|
|
for file in self.repositories['files'][:2]:
|
|
file['repositories'][0]['Options'].append({'Key': 'Enabled', 'Values': ['true']})
|
|
original_options = [file['repositories'][0]['Options'][0].copy()
|
|
for file in self.repositories['files'][:2]]
|
|
|
|
def serialize():
|
|
for file in self.repositories['files']:
|
|
file['digest'] = [self.writes] * 32
|
|
for row in file['repositories']:
|
|
row['Enabled'] = int(row['Enabled'])
|
|
if row['Components'] in (['pve-enterprise'], ['enterprise']):
|
|
row['Options'] = [option for option in row['Options']
|
|
if option['Key'] != 'Enabled']
|
|
row['Options'].append({'Key': 'Enabled', 'Values': [
|
|
'true' if row['Enabled'] else 'false']})
|
|
|
|
self.after_write = serialize
|
|
self.assertEqual(self.run_policy(), 'offer')
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
self.assertEqual(self.writes, 3)
|
|
for file, option in zip(self.repositories['files'][:2], original_options):
|
|
row = file['repositories'][0]
|
|
self.assertEqual(row['Enabled'], 0)
|
|
self.assertEqual(row['Options'], [option, {'Key': 'Enabled', 'Values': ['false']}])
|
|
self.assertEqual([c[c.index('--path') + 1] for c in self.calls if c[1] == 'create'],
|
|
['/etc/apt/sources.list.d/pve-enterprise.sources',
|
|
'/etc/apt/sources.list.d/ceph.sources'])
|
|
self.assertEqual([c[:2] for c in self.calls if c[1] in ('create', 'set')],
|
|
[['pvesh', 'create'], ['pvesh', 'create'], ['pvesh', 'set']])
|
|
|
|
def test_malformed_or_conflicting_enabled_options_stop_before_any_write(self):
|
|
invalid_options = (
|
|
None, {}, 'Enabled: false', [None], [{'Key': 'Enabled'}],
|
|
[{'Key': 'Enabled', 'Values': 'false'}],
|
|
[{'Key': 'Enabled', 'Values': [False]}],
|
|
[{'Key': 'Enabled', 'Values': []}],
|
|
[{'Key': 'Enabled', 'Values': ['false', 'false']}],
|
|
[{'Key': 'Enabled', 'Values': ['unknown']}],
|
|
[{'Key': 'Enabled', 'Values': ['true']}], # conflicts with Enabled=0
|
|
[{'Key': 'Enabled', 'Values': ['false'], 'extra': 'unexpected'}],
|
|
[{'Key': 'Enabled', 'Values': ['false']}] * 2,
|
|
[{'Values': ['false']}], [{'Key': 1, 'Values': ['false']}],
|
|
[{'Key': 'Signed-By', 'Values': [42]}],
|
|
)
|
|
for options in invalid_options:
|
|
with self.subTest(options=options):
|
|
self.setUp()
|
|
self.repositories = json.loads((ROOT / 'tests/fixtures/pr407-macrimi-pve9.2-no-subscription.json').read_bytes())
|
|
self.repositories['files'][0]['repositories'][0]['Options'] = options
|
|
with self.assertRaisesRegex(ValueError, 'inventory'):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual([c[:2] for c in self.calls], [['pvesh', 'get']])
|
|
|
|
def test_enabled_normalizes_only_booleans_and_integer_zero_one(self):
|
|
for enabled in (True, False, 0, 1):
|
|
with self.subTest(enabled=enabled):
|
|
self.setUp()
|
|
self.repositories['files'][0]['repositories'][0]['Enabled'] = enabled
|
|
with patch.object(self.module, 'run', return_value=json.dumps(self.repositories)):
|
|
data, entries = self.module.repository_state('trixie')
|
|
self.assertIs(entries[0][2]['Enabled'], bool(enabled))
|
|
self.assertIs(data['files'][0]['repositories'][0]['Enabled'], bool(enabled))
|
|
for enabled in (None, 0.0, 1.0, 0.5, -1, 2, '0', '1', 'false', [], {}):
|
|
with self.subTest(invalid=enabled):
|
|
self.setUp()
|
|
self.repositories['files'][0]['repositories'][0]['Enabled'] = enabled
|
|
with self.assertRaisesRegex(ValueError, 'inventory'):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual([c[:2] for c in self.calls], [['pvesh', 'get']])
|
|
|
|
def test_redundant_enabled_option_keeps_authoritative_state_and_other_options(self):
|
|
for enabled, values in ((True, ('true', 'yes', '1', 'TRUE')),
|
|
(False, ('false', 'no', '0', 'FALSE'))):
|
|
for key in ('Enabled', 'enabled'):
|
|
for value in values:
|
|
with self.subTest(enabled=enabled, key=key, value=value):
|
|
self.setUp()
|
|
row = self.repositories['files'][0]['repositories'][0]
|
|
row['Enabled'] = enabled
|
|
signed_by = {'Key': 'Signed-By', 'Values': ['/operator/keyring.gpg']}
|
|
row['Options'] = [signed_by, {'Key': key, 'Values': [value]}]
|
|
with patch.object(self.module, 'run', return_value=json.dumps(self.repositories)):
|
|
data, entries = self.module.repository_state('trixie')
|
|
self.assertIs(entries[0][2]['Enabled'], enabled)
|
|
self.assertEqual(data['files'][0]['repositories'][0]['Options'], [signed_by])
|
|
self.assertEqual(len(row['Options']), 2) # input untouched
|
|
|
|
def test_simulated_capture_unrelated_state_edits_abort_before_next_mutation(self):
|
|
for field in ('Enabled', 'Signed-By', 'option-added', 'URIs', 'Suites', 'Components'):
|
|
with self.subTest(field=field):
|
|
self.setUp()
|
|
self.repositories = self.simulated_enterprise_capture()
|
|
target = self.repositories['files'][0]['repositories'][0]
|
|
unrelated = self.repositories['files'][2]['repositories'][0]
|
|
|
|
def external_edit():
|
|
target['Options'].append({'Key': 'Enabled', 'Values': ['false']})
|
|
if field == 'Enabled':
|
|
unrelated['Enabled'] = 0
|
|
unrelated['Options'].append({'Key': 'Enabled', 'Values': ['false']})
|
|
elif field == 'Signed-By':
|
|
unrelated['Options'][0]['Values'] = ['/external/keyring.gpg']
|
|
elif field == 'option-added':
|
|
unrelated['Options'].append({'Key': 'Trusted', 'Values': ['yes']})
|
|
else:
|
|
unrelated[field].append('external-edit')
|
|
self.repositories['digest'] = 'external-edit'
|
|
|
|
self.after_write = external_edit
|
|
with self.assertRaisesRegex(ValueError, 'partial.*unknown'):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, 1)
|
|
self.assertTrue(self.repositories['files'][1]['repositories'][0]['Enabled'])
|
|
self.assertFalse(any(c[1] == 'set' for c in self.calls))
|
|
|
|
def test_simulated_capture_conflicting_readback_stops_after_first_write(self):
|
|
for mode in ('conflicting-option', 'malformed-option', 'reenabled-target'):
|
|
with self.subTest(mode=mode):
|
|
self.setUp()
|
|
self.repositories = self.simulated_enterprise_capture()
|
|
target = self.repositories['files'][0]['repositories'][0]
|
|
|
|
def external_edit():
|
|
value = 'unknown' if mode == 'malformed-option' else 'true'
|
|
if mode == 'reenabled-target':
|
|
target['Enabled'] = 1
|
|
target['Options'].append({'Key': 'Enabled', 'Values': [value]})
|
|
|
|
self.after_write = external_edit
|
|
with self.assertRaisesRegex(ValueError, 'partial.*unknown') as error:
|
|
self.run_policy(apply=True)
|
|
self.assertNotIn('no repository changed', str(error.exception))
|
|
self.assertEqual(self.writes, 1)
|
|
self.assertTrue(self.repositories['files'][1]['repositories'][0]['Enabled'])
|
|
self.assertFalse(any(c[1] == 'set' for c in self.calls))
|
|
|
|
def test_synthetic_pve8_list_inventory_supports_boolean_and_integer_enabled(self):
|
|
for integer in (False, True):
|
|
with self.subTest(integer=integer):
|
|
self.setUp()
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/enterprise.list', [repo('pve-enterprise', suite='bookworm')]),
|
|
('/etc/apt/sources.list', [repo('main', uri='https://deb.example/debian', suite='bookworm')]),
|
|
)
|
|
for file in self.repositories['files']:
|
|
file['file-type'] = 'list'
|
|
for row in file['repositories']:
|
|
row['FileType'] = 'list'
|
|
row['Enabled'] = 1 if integer else True
|
|
def serialize():
|
|
for file in self.repositories['files']:
|
|
for row in file['repositories']:
|
|
if integer:
|
|
row['Enabled'] = int(row['Enabled'])
|
|
# Synthetic standard handle adds the requested suite.
|
|
if row['Components'] == ['pve-no-subscription']:
|
|
row['Suites'] = ['bookworm']
|
|
self.after_write = serialize
|
|
self.assertEqual(self.run_policy(suite='bookworm'), 'offer')
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual(self.run_policy(apply=True, suite='bookworm'), 'changed')
|
|
self.assertEqual(self.writes, 2)
|
|
|
|
def test_concurrent_inventory_edit_aborts_before_second_mutation(self):
|
|
for edit in ('replace', 'reorder', 'unrelated'):
|
|
with self.subTest(edit=edit):
|
|
self.setUp()
|
|
rows = self.repositories['files'][0]['repositories']
|
|
rows.extend([repo('enterprise', uri='https://example/ceph-squid'),
|
|
repo('custom')])
|
|
|
|
def external_edit():
|
|
if self.writes != 1:
|
|
return
|
|
if edit == 'replace':
|
|
rows[1] = repo('custom-replacement')
|
|
elif edit == 'reorder':
|
|
rows[1], rows[2] = rows[2], rows[1]
|
|
else:
|
|
self.repositories['files'][1]['repositories'][0]['Options'] = [
|
|
{'Key': 'Signed-By', 'Values': ['/external/keyring.gpg']}]
|
|
self.repositories['digest'] = 'external-edit'
|
|
|
|
self.after_write = external_edit
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, 1)
|
|
self.assertTrue(rows[1]['Enabled'])
|
|
self.assertTrue(rows[2]['Enabled'])
|
|
|
|
def test_failed_post_write_readback_reports_partial_or_unknown_state(self):
|
|
for after in (1, 2):
|
|
with self.subTest(after=after):
|
|
self.setUp()
|
|
|
|
def break_readback():
|
|
if self.writes == after:
|
|
self.repositories['errors'] = [{'error': 'external parse error'}]
|
|
|
|
self.after_write = break_readback
|
|
with self.assertRaises(ValueError) as error:
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, after)
|
|
self.assertNotIn('no repository changed', str(error.exception))
|
|
self.assertRegex(str(error.exception), 'partial|unknown')
|
|
|
|
def test_inventory_comparison_accepts_api_serialization_and_new_digests(self):
|
|
self.repositories['files'][0]['repositories'].append(
|
|
repo('enterprise', uri='https://example/ceph-squid'))
|
|
self.repositories['files'].append({'path': '/etc/apt/sources.list.d/flat.list',
|
|
'repositories': [repo('', uri='https://example/flat', suite='./')]})
|
|
|
|
def serialize():
|
|
for file in self.repositories['files']:
|
|
file['digest'] = f'file-digest-{self.writes}'
|
|
for row in file['repositories']:
|
|
if row.get('Components') == []:
|
|
del row['Components']
|
|
self.repositories['files'].reverse()
|
|
|
|
self.after_write = serialize
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
self.assertEqual(self.writes, 3)
|
|
|
|
def test_fresh_iso_requires_consent_before_any_write(self):
|
|
self.assertEqual(self.run_policy(), 'offer')
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_expired_subscription_offers_switch_then_applies_with_consent(self):
|
|
self.subscription = 'key: FIXTURE-SECRET\nstatus: expired\nserverid: FIXTURE-SECRET\n'
|
|
before = json.dumps(self.repositories)
|
|
self.assertEqual(self.run_policy(), 'offer')
|
|
self.assertEqual(json.dumps(self.repositories), before)
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual([c[:2] for c in self.calls],
|
|
[['pvesh', 'get'], ['pvesubscription', 'get']])
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
self.assertEqual(self.writes, 2)
|
|
|
|
def test_consented_switch_disables_enterprise_then_adds_public_pve(self):
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
writes = [c for c in self.calls if c[1] in ('create', 'set')]
|
|
self.assertEqual([c[1] for c in writes], ['create', 'set'])
|
|
self.assertEqual(writes[0][writes[0].index('--path') + 1], '/etc/apt/sources.list.d/custom.sources')
|
|
self.assertEqual(writes[0][writes[0].index('--enabled') + 1], '0')
|
|
self.assertEqual(writes[1][writes[1].index('--handle') + 1], 'no-subscription')
|
|
|
|
def test_api_noop_after_write_does_not_report_success(self):
|
|
self.mutate = False
|
|
with self.assertRaisesRegex(ValueError, 'verify'):
|
|
self.run_policy(apply=True)
|
|
|
|
def test_flat_repository_without_components_is_preserved(self):
|
|
self.subscription = 'status: active\n'
|
|
flat = repo('', uri='https://vendor.example/flat', suite='./')
|
|
del flat['Components']
|
|
self.repositories['files'].append({'path': '/etc/apt/sources.list.d/vendor.list',
|
|
'repositories': [flat]})
|
|
before = json.dumps(self.repositories)
|
|
self.assertEqual(self.run_policy(apply=True), 'preserve')
|
|
self.assertEqual(json.dumps(self.repositories), before)
|
|
self.assertEqual(self.writes, 0)
|
|
|
|
def test_malformed_components_fail_closed(self):
|
|
self.subscription = 'status: active\n'
|
|
for components in (None, 'main', {}, [42]):
|
|
with self.subTest(components=components):
|
|
self.repositories['files'][1]['repositories'][0]['Components'] = components
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual(self.writes, 0)
|
|
|
|
def test_active_subscription_preserves_enterprise(self):
|
|
self.subscription = 'key: pve1x-SECRET\nstatus: active\nserverid: SECRET\n'
|
|
self.assertEqual(self.run_policy(apply=True), 'preserve')
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_active_subscription_without_pve_source_stops_without_add(self):
|
|
self.subscription = 'status: active\n'
|
|
self.repositories['files'][0]['repositories'][0]['Enabled'] = False
|
|
with self.assertRaisesRegex(ValueError, 'active subscription'):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_existing_public_or_test_preserved_without_subscription_lookup(self):
|
|
# Synthetic inventory only: this does not model real pvesh serialization.
|
|
for channel in ('pve-no-subscription', 'pve-test', 'pvetest'):
|
|
for subscription in ('active', 'notfound', 'expired', 'new', 'suspended',
|
|
'invalid', 'unknown', RuntimeError('lookup failed')):
|
|
for apply in (False, True):
|
|
with self.subTest(channel=channel, subscription=subscription, apply=apply):
|
|
self.setUp()
|
|
self.subscription = (subscription if isinstance(subscription, Exception)
|
|
else f'status: {subscription}\n')
|
|
self.repositories['files'][0]['repositories'].append(repo(channel))
|
|
before = json.dumps(self.repositories)
|
|
self.assertEqual(self.run_policy(apply=apply), 'preserve')
|
|
self.assertEqual(json.dumps(self.repositories), before)
|
|
self.assertEqual(self.writes, 0)
|
|
self.assertEqual([c[:2] for c in self.calls], [['pvesh', 'get']])
|
|
|
|
def test_invalid_inventory_stops_before_subscription_lookup(self):
|
|
self.repositories['errors'] = [{'error': 'fixture parse failure'}]
|
|
with self.assertRaisesRegex(ValueError, 'inventory'):
|
|
self.run_policy(apply=True)
|
|
self.assertEqual([c[:2] for c in self.calls], [['pvesh', 'get']])
|
|
self.assertEqual(self.writes, 0)
|
|
|
|
def test_legacy_pvetest_list_on_pve8_is_preserved(self):
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/operator.list', [repo('pvetest', suite='bookworm')]),
|
|
('/etc/apt/sources.list', [repo('main', uri='https://deb.example/debian', suite='bookworm')]),
|
|
)
|
|
self.assertEqual(self.run_policy(suite='bookworm'), 'preserve')
|
|
|
|
def test_disabled_enterprise_stanza_is_not_disabled_again(self):
|
|
self.repositories['files'][0]['repositories'][0]['Enabled'] = False
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
self.assertFalse(any(c[1] == 'create' for c in self.calls))
|
|
|
|
def test_unrecognized_and_failed_subscription_cannot_mutate(self):
|
|
for response in ('status: unknown\n', 'status: invalid\n', 'status: new\n',
|
|
'status: suspended\n', 'status: active\nstatus: notfound\n',
|
|
'key: secret\n', RuntimeError('service unavailable')):
|
|
with self.subTest(response=response):
|
|
self.subscription = response
|
|
self.calls = []
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_multiple_stanzas_disables_only_pve_enterprise_and_ceph_enterprise(self):
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/anything.sources', [
|
|
repo('main', uri='https://debian.example/debian'),
|
|
repo('pve-enterprise'),
|
|
repo('enterprise', uri='https://enterprise.proxmox.com/debian/ceph-squid'),
|
|
repo('pve-enterprise', enabled=False),
|
|
]),
|
|
)
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
writes = [c for c in self.calls if c[1] == 'create']
|
|
self.assertEqual([int(c[c.index('--index') + 1]) for c in writes], [1, 2])
|
|
|
|
def test_mixed_components_block_switch_without_partial_write(self):
|
|
self.repositories['files'][0]['repositories'] = [repo('pve-enterprise custom')]
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_wrong_suite_and_parse_errors_block_before_writes(self):
|
|
for mode in ('suite', 'error'):
|
|
with self.subTest(mode=mode):
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/custom.list', [repo('pve-enterprise', suite='bookworm')]),
|
|
('/etc/apt/sources.list.d/debian.sources', [repo('main', uri='https://deb.example/debian')]),
|
|
)
|
|
if mode == 'error':
|
|
self.repositories['errors'] = [{'path': 'custom.list', 'error': 'invalid'}]
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_wrong_suite_names_each_stale_source_and_where_to_fix_it(self):
|
|
self.repositories = fixture(
|
|
('/etc/apt/sources.list.d/pve.list', [repo('pve-no-subscription')]),
|
|
('/etc/apt/sources.list.d/pve-install-repo.list',
|
|
[repo('pve-no-subscription', uri='https://user:secret@mirror.example/debian/pve', suite='bookworm')]),
|
|
('/etc/apt/sources.list.d/debian.sources', [repo('main', uri='https://deb.example/debian')]),
|
|
)
|
|
with self.assertRaises(ValueError) as error:
|
|
self.run_policy(apply=True)
|
|
message = str(error.exception)
|
|
self.assertIn('/etc/apt/sources.list.d/pve-install-repo.list', message)
|
|
self.assertIn('pve-no-subscription', message)
|
|
self.assertIn('suite bookworm', message)
|
|
self.assertIn('runs Proxmox VE on trixie', message)
|
|
self.assertIn('Node > Updates > Repositories', message)
|
|
self.assertNotIn('/etc/apt/sources.list.d/pve.list', message)
|
|
self.assertNotIn('secret', message)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_ceph_enterprise_mirror_is_disabled_without_choosing_ceph_channel(self):
|
|
self.repositories['files'].append({'path': '/etc/apt/sources.list.d/storage.list',
|
|
'repositories': [repo('enterprise', uri='https://mirror.example/ceph-squid')]})
|
|
self.assertEqual(self.run_policy(apply=True), 'changed')
|
|
writes = [c for c in self.calls if c[1] == 'create']
|
|
self.assertEqual(len(writes), 2)
|
|
self.assertFalse(any(c[1] == 'set' and 'ceph' in ' '.join(c) for c in self.calls))
|
|
|
|
def test_wrong_suite_ceph_stops_before_switch(self):
|
|
self.repositories['files'].append({'path': '/etc/apt/sources.list.d/ceph.sources',
|
|
'repositories': [repo('enterprise', uri='https://enterprise.proxmox.com/debian/ceph-reef', suite='bookworm')]})
|
|
with self.assertRaisesRegex(ValueError, 'suite'):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
def test_missing_debian_base_blocks_switch(self):
|
|
self.repositories['files'].pop()
|
|
with self.assertRaises(ValueError):
|
|
self.run_policy(apply=True)
|
|
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
|
|
|
|
|
|
class CallerPropagationTest(unittest.TestCase):
|
|
def test_safe_update_refusal_blocks_apt_update(self):
|
|
source = (ROOT / 'scripts/global/update-pve-safe.sh').read_text()
|
|
body = source.split('update_pve_safe() {', 1)[1].split(
|
|
'\nif [[ "${BASH_SOURCE[0]}"', 1)[0]
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
body = body.replace('/var/log/', f'{d}/').replace(
|
|
'/tmp/proxmenux_screen_capture_', f'{d}/proxmenux_screen_capture_')
|
|
script = '''
|
|
pveversion() { printf 'pve-manager/9.0.0\\n'; }
|
|
translate() { printf '%s' "$1"; }
|
|
msg_info2() { :; }
|
|
msg_error() { :; }
|
|
df() { printf 'Filesystem 1K-blocks Used Available Use%% Mounted on\\n';
|
|
printf 'fixture 9999999 1 9999999 1%% /\\n'; }
|
|
download_common_functions() { :; }
|
|
source_install_functions() { :; }
|
|
ensure_repositories() { return 1; }
|
|
apt-get() { touch "$TEST_ROOT/apt_called"; }
|
|
update_pve_safe() {
|
|
''' + body + '\nupdate_pve_safe\n'
|
|
result = subprocess.run(['bash', '-c', script], env=dict(os.environ, TEST_ROOT=d),
|
|
capture_output=True, text=True, stdin=subprocess.DEVNULL,
|
|
timeout=15)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((Path(d) / 'apt_called').exists())
|
|
|
|
def test_nvidia_refusal_preserves_working_driver_before_uninstall(self):
|
|
source = (ROOT / 'scripts/gpu_tpu/nvidia_installer.sh').read_text()
|
|
main = 'main() {' + source.split('main() {', 1)[1].split(
|
|
'\n# ==========================================================\n# Non-interactive', 1)[0]
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
script = '''
|
|
LOG_FILE="$TEST_ROOT/log"; screen_capture="$TEST_ROOT/screen"
|
|
NVIDIA_GPU_PRESENT=true; CURRENT_DRIVER_INSTALLED=true
|
|
CURRENT_DRIVER_VERSION=580.1; DRIVER_VERSION=580.2; ACTION=install
|
|
for fn in detect_nvidia_gpus detect_driver_status check_gpu_not_in_vm_passthrough \\
|
|
check_stale_vfio_config_for_nvidia show_action_menu_if_installed \\
|
|
show_install_overview get_system_info show_version_menu hybrid_yesno \\
|
|
show_proxmenux_logo msg_title msg_info2 sleep; do
|
|
eval "$fn() { :; }"
|
|
done
|
|
translate() { printf '%s' "$1"; }
|
|
ensure_repos_and_headers() { return 1; }
|
|
download_nvidia_installer() { touch "$TEST_ROOT/downloaded"; return 1; }
|
|
complete_nvidia_uninstall() { touch "$TEST_ROOT/uninstalled"; }
|
|
''' + main + '\nmain\n'
|
|
result = subprocess.run(['bash', '-c', script],
|
|
env=dict(os.environ, TEST_ROOT=d),
|
|
capture_output=True, text=True, timeout=15)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((Path(d) / 'downloaded').exists())
|
|
self.assertFalse((Path(d) / 'uninstalled').exists())
|
|
|
|
def test_nvidia_repository_failure_blocks_header_install(self):
|
|
source = (ROOT / 'scripts/gpu_tpu/nvidia_installer.sh').read_text()
|
|
fn = source.split('ensure_repos_and_headers() {', 1)[1].split(
|
|
'\n_nouveau_legacy_file_is_proxmenux_shape()', 1)[0]
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
script = '''
|
|
LOG_FILE="$TEST_ROOT/log"; screen_capture="$TEST_ROOT/screen"
|
|
translate() { printf %s "$1"; }
|
|
msg_error() { :; }
|
|
ensure_repositories() { return 1; }
|
|
apt-get() { touch "$TEST_ROOT/apt_called"; }
|
|
ensure_repos_and_headers() {
|
|
''' + fn + '\nensure_repos_and_headers\n'
|
|
result = subprocess.run(['bash', '-c', script], env=dict(os.environ, TEST_ROOT=d),
|
|
capture_output=True, text=True, timeout=15)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((Path(d) / 'apt_called').exists())
|
|
|
|
|
|
class ShellFlowTest(unittest.TestCase):
|
|
def test_expired_and_notfound_consent_use_real_policy_with_synthetic_api(self):
|
|
helper = ROOT / 'scripts/global/repository-functions.sh'
|
|
# The runner replaces every API command; bool Enabled is synthetic and
|
|
# deliberately does not claim to reproduce the maintainer's JSON.
|
|
runner_source = f'''import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
spec = importlib.util.spec_from_file_location('policy', {str(POLICY)!r})
|
|
policy = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(policy)
|
|
root = Path(os.environ['TEST_ROOT'])
|
|
state_path = root / 'inventory.json'
|
|
state = json.loads(state_path.read_text())
|
|
def run(args):
|
|
verb = args[:2]
|
|
with (root / 'calls').open('a') as log:
|
|
log.write(' '.join(verb) + '\\n')
|
|
if verb == ['pvesubscription', 'get']:
|
|
return 'key: FIXTURE-SECRET\\nstatus: ' + os.environ['STATUS'] + '\\n'
|
|
if verb == ['pvesh', 'get']:
|
|
return json.dumps(state)
|
|
if verb == ['pvesh', 'create']:
|
|
path = args[args.index('--path') + 1]
|
|
index = int(args[args.index('--index') + 1])
|
|
next(f for f in state['files'] if f['path'] == path)['repositories'][index]['Enabled'] = False
|
|
elif verb == ['pvesh', 'set']:
|
|
row = dict(state['files'][0]['repositories'][0])
|
|
row['Enabled'] = True
|
|
row['Components'] = ['pve-no-subscription']
|
|
state['files'].append({{'path': '/etc/apt/sources.list.d/public.sources', 'repositories': [row]}})
|
|
else:
|
|
raise AssertionError('Unexpected fixture API command')
|
|
state['digest'] += '-write'
|
|
state_path.write_text(json.dumps(state))
|
|
return ''
|
|
policy.run = run
|
|
raise SystemExit(policy.main())
|
|
'''
|
|
for status, consent, refresh_rc in (('expired', False, 0), ('expired', True, 0),
|
|
('expired', True, 100), ('notfound', False, 0),
|
|
('notfound', True, 0)):
|
|
with self.subTest(status=status, consent=consent, refresh_rc=refresh_rc):
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
root = Path(d)
|
|
runner = root / 'fixture_policy.py'
|
|
runner.write_text(runner_source)
|
|
state_path = root / 'inventory.json'
|
|
state_path.write_text(json.dumps(fixture(
|
|
('/etc/apt/sources.list.d/operator.sources', [repo('pve-enterprise')]),
|
|
('/etc/apt/sources.list.d/base.sources', [repo('main')]))))
|
|
before = state_path.read_bytes()
|
|
script = f'''source "{helper}"
|
|
pveversion() {{ echo pve-manager/9.0; }}
|
|
translate() {{ printf %s "$1"; }}
|
|
msg_error() {{ printf '%s\\n' "$1" >&2; }}
|
|
is_web_mode() {{ return 0; }}
|
|
hybrid_yesno() {{ printf 'prompt\\n' >> "$TEST_ROOT/calls";
|
|
printf '%s\\n' "$2" >&2; {'return 0' if consent else 'return 1'}; }}
|
|
repository_policy() {{ python3 "{runner}" "$@"; }}
|
|
apt-get() {{ printf 'apt %s\\n' "$*" >> "$TEST_ROOT/calls"; return {refresh_rc}; }}
|
|
if ensure_repositories; then printf 'consumer\\n' >> "$TEST_ROOT/calls"; exit 0;
|
|
else exit $?; fi
|
|
'''
|
|
result = subprocess.run(['bash', '-c', script], input='', text=True,
|
|
capture_output=True, timeout=15,
|
|
env=dict(os.environ, TEST_ROOT=d, STATUS=status))
|
|
calls = (root / 'calls').read_text().splitlines()
|
|
self.assertEqual(calls[:3], ['pvesh get', 'pvesubscription get', 'prompt'])
|
|
self.assertIn('no active subscription', result.stderr)
|
|
self.assertNotIn('FIXTURE-SECRET', result.stdout + result.stderr)
|
|
if not consent:
|
|
self.assertEqual(calls, ['pvesh get', 'pvesubscription get', 'prompt'])
|
|
self.assertEqual(state_path.read_bytes(), before)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
else:
|
|
self.assertEqual(calls[3:10], ['pvesh get', 'pvesubscription get',
|
|
'pvesh create', 'pvesh get', 'pvesh set', 'pvesh get', 'apt update'])
|
|
state = json.loads(state_path.read_text())
|
|
self.assertFalse(state['files'][0]['repositories'][0]['Enabled'])
|
|
self.assertEqual(state['files'][-1]['repositories'][0]['Components'],
|
|
['pve-no-subscription'])
|
|
self.assertEqual(result.returncode, refresh_rc)
|
|
if refresh_rc:
|
|
self.assertNotIn('consumer', calls)
|
|
self.assertIn('sources changed, but APT', result.stderr)
|
|
else:
|
|
self.assertEqual(calls[-1], 'consumer')
|
|
|
|
def test_refresh_only_after_changed_apply_and_failure_stops_consumer(self):
|
|
helper = ROOT / 'scripts/global/repository-functions.sh'
|
|
for plan, apply, refresh_rc in (('offer', 'changed', 0), ('offer', 'changed', 100),
|
|
('offer', 'preserve', 0), ('preserve', 'changed', 0),
|
|
('offer', 'unexpected', 0), ('offer', 'failed', 0)):
|
|
for pipefail in (False, True):
|
|
with self.subTest(plan=plan, apply=apply, refresh_rc=refresh_rc, pipefail=pipefail):
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
log = Path(d) / 'calls'
|
|
script = f'''source "{helper}"
|
|
set -e; set {'-o' if pipefail else '+o'} pipefail
|
|
pveversion() {{ echo pve-manager/9.0; }}
|
|
translate() {{ printf %s "$1"; }}
|
|
msg_error() {{ printf '%s\\n' "$1" >&2; }}
|
|
is_web_mode() {{ return 0; }}
|
|
hybrid_yesno() {{ printf 'prompt\\n' >> "{log}"; }}
|
|
repository_policy() {{
|
|
printf '%s\\n' "$1" >> "{log}"
|
|
if [[ "$1" == plan ]]; then echo {plan};
|
|
elif [[ {apply} == failed ]]; then return 1;
|
|
else echo {apply}; fi
|
|
}}
|
|
apt-get() {{ printf 'apt %s\\n' "$*" >> "{log}"; return {refresh_rc}; }}
|
|
# Exercise the conditional context used by most callers:
|
|
# inherited errexit cannot be relied on inside the helper.
|
|
if ensure_repositories; then
|
|
printf 'consumer\\n' >> "{log}"; exit 0
|
|
else exit $?; fi
|
|
'''
|
|
result = subprocess.run(['bash', '-c', script], input='', text=True,
|
|
capture_output=True, timeout=15)
|
|
calls = log.read_text().splitlines()
|
|
expected = ['plan']
|
|
if plan == 'offer':
|
|
expected += ['prompt', 'apply']
|
|
if apply == 'changed':
|
|
expected += ['apt update']
|
|
success = (plan == 'preserve' or apply in ('changed', 'preserve')) and refresh_rc == 0
|
|
if success:
|
|
expected += ['consumer']
|
|
self.assertEqual(calls, expected)
|
|
self.assertEqual(result.returncode == 0, success, result.stderr)
|
|
if refresh_rc:
|
|
self.assertRegex(result.stderr, 'sources changed|repositories changed')
|
|
self.assertIn('APT', result.stderr)
|
|
self.assertNotIn('no APT source changed', result.stderr)
|
|
self.assertNotIn('success', result.stdout.lower())
|
|
|
|
def test_web_consent_applies_only_after_prompt(self):
|
|
helper = ROOT / 'scripts/global/repository-functions.sh'
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
log = Path(d) / 'calls'
|
|
script = f'''source "{helper}"
|
|
pveversion() {{ echo pve-manager/9.0; }}
|
|
translate() {{ printf %s "$1"; }}
|
|
msg_error() {{ :; }}
|
|
is_web_mode() {{ return 0; }}
|
|
hybrid_yesno() {{ printf 'prompt\\n' >> "{log}"; [[ "$2" == *'This host has no active subscription, switch to the no-subscription repository?'* ]]; }}
|
|
repository_policy() {{ printf '%s\\n' "$1" >> "{log}"; [[ "$1" == plan ]] && echo offer || echo changed; }}
|
|
apt-get() {{ printf 'apt %s\\n' "$*" >> "{log}"; }}
|
|
ensure_repositories
|
|
'''
|
|
result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(log.read_text().splitlines(), ['plan', 'prompt', 'apply', 'apt update'])
|
|
|
|
def test_refusal_and_noninteractive_never_apply(self):
|
|
helper = ROOT / 'scripts/global/repository-functions.sh'
|
|
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
|
|
mock = Path(d) / 'policy.py'
|
|
mock.write_text('import sys\nfrom pathlib import Path\n'
|
|
'Path(sys.argv[1]).open("a").write(sys.argv[2]+"\\n")\n'
|
|
'print("offer" if sys.argv[2] == "plan" else "changed")\n')
|
|
for web, confirm in ((False, True), (True, False)):
|
|
with self.subTest(web=web, confirm=confirm):
|
|
log = Path(d) / 'calls'
|
|
log.unlink(missing_ok=True)
|
|
script = f'''source "{helper}"
|
|
pveversion() {{ echo pve-manager/9.0; }}
|
|
translate() {{ printf %s "$1"; }}
|
|
msg_error() {{ :; }}
|
|
msg_info2() {{ :; }}
|
|
is_web_mode() {{ {'return 0' if web else 'return 1'}; }}
|
|
hybrid_yesno() {{ {'return 0' if confirm else 'return 1'}; }}
|
|
repository_policy() {{ python3 "{mock}" "{log}" "$1"; }}
|
|
apt-get() {{ printf 'apt %s\\n' "$*" >> "{log}"; }}
|
|
ensure_repositories
|
|
'''
|
|
result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(log.read_text(), 'plan\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|