Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-10-08 14:36:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: normalize Proxmox repository state and refresh consented switches

Browse Source
This commit is contained in:
martino
2026-09-30 22:05:46 +02:00
parent e57a2f3aae
commit 028026a9a0
7 changed files with 1032 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
guides/repository-flow.md
-11
View File
@@ -1,11 +0,0 @@
# Repository choice in dependency and safe-update flows
On PVE 8/9, the shared helper checks `pvesubscription get` (only the `status` field, never logging the key) and reads parsed APT sources through Proxmox's `/nodes/localhost/apt/repositories` API. An **active** subscription preserves the configured Enterprise repository and all other sources; if no PVE channel is active, the flow stops for manual repair rather than adding one. An already active PVE no-subscription or test channel is likewise preserved. Repository files may use custom names, `.list` or deb822 `.sources`, and mirrors; presence of a filename alone does not establish an active channel.
A fresh ISO may have Enterprise enabled but **no subscription** (`status: notfound`). Before changing anything, interactive flows ask: **“This host has no subscription, switch to the no-subscription repository?”** Acceptance disables active Enterprise PVE entries with Proxmox's per-entry API and adds Proxmox's standard no-subscription PVE repository. If an Enterprise Ceph source is active, it is disabled too, but **no Ceph replacement is chosen**: the appropriate Ceph release/channel must be selected separately in **Node > Updates > Repositories** if Ceph is used. Unrelated stanzas and Debian sources remain untouched. A mixed-component Enterprise stanza cannot be disabled safely, so the flow stops for manual splitting instead. A missing Debian base source also stops for manual configuration, rather than guessing a mirror.
Declining or running without an interactive terminal/web dialog makes **no source changes**. Unrecognized, expired, invalid, suspended, malformed, or unavailable subscription status is **not** interpreted as no subscription; lookup and repository parser errors stop before changes. The apply step rechecks the subscription and repository inventory and uses the API digest on writes. A failure during a multi-entry API update can leave a partial switch; inspect the repository UI before retrying. General APT error-handling policy is unchanged in this focused patch; stricter index refresh and broader dependency validation remain separate work. The NVIDIA interactive driver installer checks repositories before downloading or removing the working driver, and the noninteractive reinstall propagates refusal.
This policy does **not** change the legacy automated post-install repository updater, which separately advertises and asks for free repositories. No fresh-ISO live integration has been run; fixture tests exercise the policy without touching the host's APT configuration.
Primary references: [Proxmox subscription CLI and status schema](https://github.com/proxmox/pve-manager/blob/master/PVE/CLI/pvesubscription.pm), [Proxmox APT repository API](https://github.com/proxmox/pve-manager/blob/master/PVE/API2/APT.pm), [Proxmox standard repository definitions](https://github.com/proxmox/proxmox-rs/blob/master/proxmox-apt/src/repositories/standard.rs), [Package Repositories](https://pve.proxmox.com/wiki/Package_Repositories).
scripts/global/repository-functions.sh
+14 -3
View File
@@ -7,7 +7,7 @@ repository_policy() {
} }
ensure_repositories() { ensure_repositories() {
local version suite decision local version suite decision refresh_status
version=$(pveversion 2>/dev/null | grep -oP 'pve-manager/\K[0-9]+' | head -1) version=$(pveversion 2>/dev/null | grep -oP 'pve-manager/\K[0-9]+' | head -1)
case "$version" in case "$version" in
8) suite=bookworm ;; 8) suite=bookworm ;;
@@ -22,15 +22,26 @@ ensure_repositories() {
*) msg_error "$(translate 'Repository policy returned an unexpected result.')"; return 1 ;; *) msg_error "$(translate 'Repository policy returned an unexpected result.')"; return 1 ;;
esac esac
if ! declare -F hybrid_yesno >/dev/null || { [[ ! -t 0 ]] && ! { declare -F is_web_mode >/dev/null && is_web_mode; }; }; then if ! declare -F hybrid_yesno >/dev/null || { [[ ! -t 0 ]] && ! { declare -F is_web_mode >/dev/null && is_web_mode; }; }; then
msg_error "$(translate 'No subscription and no usable PVE repository. Noninteractive mode cannot change APT sources; configure them in Node > Updates > Repositories.')" msg_error "$(translate 'No active subscription and no usable PVE repository. Noninteractive mode cannot change APT sources; configure them in Node > Updates > Repositories.')"
return 1 return 1
fi fi
if ! hybrid_yesno "$(translate 'Proxmox repository')" \ if ! hybrid_yesno "$(translate 'Proxmox repository')" \
"$(translate 'This host has no subscription, switch to the no-subscription repository? The inaccessible Enterprise PVE source will be disabled. Enterprise Ceph sources, if present, will be disabled without choosing a replacement Ceph channel; configure Ceph separately if needed.')" 16 90; then "$(translate 'This host has no active subscription, switch to the no-subscription repository? The inaccessible Enterprise PVE source will be disabled. Enterprise Ceph sources, if present, will be disabled without choosing a replacement Ceph channel; configure Ceph separately if needed.')" 16 90; then
msg_error "$(translate 'Repository switch declined; no APT source changed.')" msg_error "$(translate 'Repository switch declined; no APT source changed.')"
return 1 return 1
fi fi
decision=$(repository_policy apply "$suite") || return 1 decision=$(repository_policy apply "$suite") || return 1
[[ "$decision" == changed || "$decision" == preserve ]] || return 1 [[ "$decision" == changed || "$decision" == preserve ]] || return 1
if [[ "$decision" == changed ]]; then
# Direct callers may install immediately; refresh their package indexes
# before returning. Preserve paths must not trigger an extra refresh.
if apt-get update; then
return 0
else
refresh_status=$?
msg_error "$(translate 'Repository sources changed, but APT package-list refresh failed. Operation stopped; sources were not rolled back. Inspect Node > Updates > Repositories and retry apt-get update before continuing.')"
return "$refresh_status"
fi
fi
return 0 return 0
} }
scripts/global/repository_policy.py
+42 -7
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Proxmox repository policy via its own parsed APT repository API. """Proxmox repository policy via its own parsed APT repository API.
Only `notfound` permits an offered switch. Never print raw subscription or Only `notfound` or `expired` permits an offered switch. Never print raw subscription or
repository API responses: they can contain subscription keys or credentials. repository API responses: they can contain subscription keys or credentials.
""" """
import copy import copy
@@ -26,8 +26,8 @@ def subscription_status():
# including a secret key and server ID. Only parse the exact status line. # including a secret key and server ID. Only parse the exact status line.
output = run(['pvesubscription', 'get']) output = run(['pvesubscription', 'get'])
statuses = re.findall(r'^status: ([a-z]+)$', output, re.MULTILINE) statuses = re.findall(r'^status: ([a-z]+)$', output, re.MULTILINE)
if len(statuses) != 1 or statuses[0] not in ('active', 'notfound'): if len(statuses) != 1 or statuses[0] not in ('active', 'notfound', 'expired'):
raise ValueError('Subscription status is not unambiguously active or notfound. Check pvesubscription get locally; no repository changed.') raise ValueError('Subscription status is not unambiguously active, notfound or expired. Check pvesubscription get locally; no repository changed.')
return statuses[0] return statuses[0]
@@ -45,11 +45,44 @@ def repository_state(suite):
raise ValueError() raise ValueError()
# Flat repositories (e.g. suite './') legitimately omit this. # Flat repositories (e.g. suite './') legitimately omit this.
row.setdefault('Components', []) row.setdefault('Components', [])
if not isinstance(row['Enabled'], bool) or not all( # Captured PVE 9.2 JSON uses integer 0/1; older fixtures use bool.
# Reject float/string coercions and retain the authoritative state.
if type(row['Enabled']) not in (bool, int) or row['Enabled'] not in (0, 1):
raise ValueError()
row['Enabled'] = bool(row['Enabled'])
if not all(
isinstance(row[k], list) and all(isinstance(value, str) for value in row[k]) isinstance(row[k], list) and all(isinstance(value, str) for value in row[k])
for k in ('Types', 'URIs', 'Suites', 'Components') for k in ('Types', 'URIs', 'Suites', 'Components')
): ):
raise ValueError() raise ValueError()
# deb822 write/readback can insert Options/Enabled. It is
# redundant only when well-formed and consistent with Enabled;
# never discard the authoritative field or unrelated options.
options = row.get('Options', [])
if not isinstance(options, list):
raise ValueError()
remaining = []
seen_enabled = False
boolean_options = {'true': True, 'yes': True, '1': True,
'false': False, 'no': False, '0': False}
for option in options:
if not isinstance(option, dict) or not isinstance(option['Key'], str) \
or not isinstance(option['Values'], list) \
or not all(isinstance(value, str) for value in option['Values']):
raise ValueError()
if option['Key'].lower() != 'enabled':
remaining.append(option)
continue
if seen_enabled or set(option) != {'Key', 'Values'} or len(option['Values']) != 1:
raise ValueError()
value = boolean_options.get(option['Values'][0].lower())
if value is None or value != row['Enabled']:
raise ValueError()
seen_enabled = True
if remaining:
row['Options'] = remaining
else:
row.pop('Options', None)
entries.append((file['path'], index, row)) entries.append((file['path'], index, row))
return data, entries return data, entries
except (KeyError, TypeError, ValueError, IndexError) as exc: except (KeyError, TypeError, ValueError, IndexError) as exc:
@@ -59,7 +92,6 @@ def repository_state(suite):
def evaluate(suite, apply=False): def evaluate(suite, apply=False):
if suite not in ('bookworm', 'trixie'): if suite not in ('bookworm', 'trixie'):
raise ValueError('Unsupported Proxmox suite; no repository changed.') raise ValueError('Unsupported Proxmox suite; no repository changed.')
status = subscription_status() # fail closed before inventory or any write
data, entries = repository_state(suite) data, entries = repository_state(suite)
pve = [] pve = []
ceph = [] ceph = []
@@ -79,12 +111,15 @@ def evaluate(suite, apply=False):
raise ValueError('Enterprise Ceph repository suite does not match this PVE version; correct it in the Proxmox repository UI before switching.') raise ValueError('Enterprise Ceph repository suite does not match this PVE version; correct it in the Proxmox repository UI before switching.')
ceph.append((path, index, row)) ceph.append((path, index, row))
if any(set(row['Components']) & {'pve-no-subscription', 'pve-test', 'pvetest'} for _, _, row in pve):
return 'preserve'
# Entitlement lookup is only needed when considering a switch. Existing
# public/test channels are the operator's choice, even if lookup fails.
status = subscription_status() # fail closed before any write
if status == 'active': if status == 'active':
if not pve: if not pve:
raise ValueError('Host has an active subscription but no active PVE repository. Configure its Enterprise source in Node > Updates > Repositories; no repository changed.') raise ValueError('Host has an active subscription but no active PVE repository. Configure its Enterprise source in Node > Updates > Repositories; no repository changed.')
return 'preserve' return 'preserve'
if any(set(row['Components']) & {'pve-no-subscription', 'pve-test', 'pvetest'} for _, _, row in pve):
return 'preserve'
if not debian: if not debian:
raise ValueError('No active Debian base repository for this suite; configure it in the Proxmox repository UI before continuing.') raise ValueError('No active Debian base repository for this suite; configure it in the Proxmox repository UI before continuing.')
# No active PVE source or only inaccessible Enterprise. A mixed stanza # No active PVE source or only inaccessible Enterprise. A mixed stanza
tests/fixtures/README.md
Vendored
+57
View File
@@ -0,0 +1,57 @@
# PR407 repository API evidence
## Authentic capture (unchanged)
`pr407-macrimi-pve9.2-no-subscription.json` is the complete maintainer-provided
`pvesh get /nodes/localhost/apt/repositories --output-format json` attachment.
- Provider: MacRimi; reported host version: PVE 9.2.20.
- Context: <https://github.com/MacRimi/ProxMenux/pull/407#issuecomment-5918280377>
- Original attachment: <https://github.com/user-attachments/files/32873450/pvesh-apt-repositories-pve9.2-no-subscription.json>
- Size: 8403 bytes.
- SHA-256: `dfbc0560e280511ce5cef55b8f059095c57df74e062d149065f2363a80b3847f`.
This is **already a no-subscription configuration**, not a fresh-ISO Enterprise
before capture and not a captured before/after pair. It contains numeric
`Enabled: 0/1`, disabled Enterprise PVE/Ceph stanzas with
`Options: [{"Key": "Enabled", "Values": ["false"]}, ...]`, and an enabled public
PVE stanza whose Enabled option is `true`. Signed-By options remain in the file.
The authentic-capture test verifies both plan/apply preserve paths without any
subscription lookup or API write. The attachment is stored byte-for-byte; tests
parse separate in-memory copies, never rewrite this evidence.
## Generated scenarios (not captured host state)
`RepositoryPolicyTest.simulated_enterprise_capture()` in
`tests/test_repository_preservation.py` transforms an in-memory copy: removes the
public PVE file, enables Enterprise PVE/Ceph, removes their redundant Enabled
options, and labels the global digest `simulated-enterprise-before`. Remaining
capture metadata/file digests are only fixture input, not evidence of that
invented configuration. Tests optionally add a consistent Enabled=true option.
The fake API disables one targeted entry at a time, simulates numeric Enabled and
Enabled-option insertion/update on readback, changes fixture digests, and adds a
synthetic public PVE entry for the standard handle. This verifies policy behavior
and retention of unrelated Signed-By options; it is **not** an observed live API
transition or an authentic Enterprise-before capture. Additional edits simulate
concurrent URI/suite/component/option/enabled changes and malformed/conflicting
readback. They must stop before a second mutation and report partial/unknown
state after the attempted write.
The PVE 8 `.list` cases and other hand-built inventories are explicitly synthetic.
No real Proxmox API command, package operation, or host source-file change is
executed by these tests. Fresh-ISO validation is still required separately.
## Primary implementation references
- [Proxmox subscription CLI/status parsing](https://github.com/proxmox/pve-manager/blob/master/PVE/CLI/pvesubscription.pm)
- [Proxmox parsed APT repository API](https://github.com/proxmox/pve-manager/blob/master/PVE/API2/APT.pm)
- [Proxmox standard repository definitions](https://github.com/proxmox/proxmox-rs/blob/master/proxmox-apt/src/repositories/standard.rs)
- [Proxmox package repositories](https://pve.proxmox.com/wiki/Package_Repositories)
The policy normalizes bool and integer 0/1 only, retaining the normalized Enabled
field as authoritative. A single well-formed Enabled option is redundant only
when consistent with that field. Other options and all unrelated repository
fields remain part of semantic inventory comparison; per-file digests alone are
excluded. Missing/empty Options and omitted/empty Components are semantically
equivalent. Unknown or conflicting Enabled serialization fails closed.
tests/fixtures/pr407-macrimi-pve9.2-no-subscription.json
Vendored
+393
View File
@@ -0,0 +1,393 @@
{
"digest": "7fbffd0c075540790e7f2ec23c75aeba46de0678018ddaf8bbd651df83d877d5",
"errors": [],
"files": [
{
"digest": [
253,
100,
211,
177,
180,
35,
211,
239,
245,
90,
223,
51,
116,
46,
51,
147,
22,
10,
146,
148,
233,
90,
157,
51,
173,
107,
3,
6,
244,
161,
179,
36
],
"file-type": "sources",
"path": "/etc/apt/sources.list.d/pve-enterprise.sources",
"repositories": [
{
"Components": [
"pve-enterprise"
],
"Enabled": 0,
"FileType": "sources",
"Options": [
{
"Key": "Signed-By",
"Values": [
"/usr/share/keyrings/proxmox-archive-keyring.gpg"
]
},
{
"Key": "Enabled",
"Values": [
"false"
]
}
],
"Suites": [
"trixie"
],
"Types": [
"deb"
],
"URIs": [
"https://enterprise.proxmox.com/debian/pve"
]
}
]
},
{
"digest": [
123,
225,
114,
83,
116,
171,
113,
115,
199,
10,
41,
251,
134,
47,
44,
77,
199,
16,
248,
9,
228,
127,
22,
199,
36,
127,
226,
48,
146,
96,
111,
182
],
"file-type": "sources",
"path": "/etc/apt/sources.list.d/ceph.sources",
"repositories": [
{
"Components": [
"enterprise"
],
"Enabled": 0,
"FileType": "sources",
"Options": [
{
"Key": "Signed-By",
"Values": [
"/usr/share/keyrings/proxmox-archive-keyring.gpg"
]
},
{
"Key": "Enabled",
"Values": [
"false"
]
}
],
"Suites": [
"trixie"
],
"Types": [
"deb"
],
"URIs": [
"https://enterprise.proxmox.com/debian/ceph-squid"
]
}
]
},
{
"digest": [
6,
229,
80,
130,
43,
203,
73,
237,
251,
213,
123,
79,
85,
43,
70,
149,
17,
51,
41,
180,
231,
173,
109,
5,
44,
6,
197,
14,
77,
153,
233,
89
],
"file-type": "sources",
"path": "/etc/apt/sources.list.d/proxmox.sources",
"repositories": [
{
"Components": [
"pve-no-subscription"
],
"Enabled": 1,
"FileType": "sources",
"Options": [
{
"Key": "Enabled",
"Values": [
"true"
]
},
{
"Key": "Signed-By",
"Values": [
"/usr/share/keyrings/proxmox-archive-keyring.gpg"
]
}
],
"Suites": [
"trixie"
],
"Types": [
"deb"
],
"URIs": [
"http://download.proxmox.com/debian/pve"
]
}
]
},
{
"digest": [
183,
74,
82,
83,
226,
120,
146,
197,
41,
71,
199,
156,
112,
67,
208,
149,
172,
234,
178,
166,
137,
221,
198,
153,
50,
116,
80,
57,
229,
187,
82,
194
],
"file-type": "sources",
"path": "/etc/apt/sources.list.d/debian.sources",
"repositories": [
{
"Components": [
"main",
"contrib",
"non-free",
"non-free-firmware"
],
"Enabled": 1,
"FileType": "sources",
"Options": [
{
"Key": "Signed-By",
"Values": [
"/usr/share/keyrings/debian-archive-keyring.gpg"
]
}
],
"Suites": [
"trixie",
"trixie-updates"
],
"Types": [
"deb"
],
"URIs": [
"http://deb.debian.org/debian/"
]
},
{
"Components": [
"main",
"contrib",
"non-free",
"non-free-firmware"
],
"Enabled": 1,
"FileType": "sources",
"Options": [
{
"Key": "Signed-By",
"Values": [
"/usr/share/keyrings/debian-archive-keyring.gpg"
]
}
],
"Suites": [
"trixie-security"
],
"Types": [
"deb"
],
"URIs": [
"http://security.debian.org/debian-security/"
]
}
]
}
],
"infos": [
{
"index": 0,
"kind": "origin",
"message": "Proxmox",
"path": "/etc/apt/sources.list.d/pve-enterprise.sources"
},
{
"index": 0,
"kind": "origin",
"message": "Proxmox",
"path": "/etc/apt/sources.list.d/ceph.sources"
},
{
"index": 0,
"kind": "origin",
"message": "Proxmox",
"path": "/etc/apt/sources.list.d/proxmox.sources"
},
{
"index": 0,
"kind": "origin",
"message": "Debian",
"path": "/etc/apt/sources.list.d/debian.sources"
},
{
"index": 1,
"kind": "origin",
"message": "Debian",
"path": "/etc/apt/sources.list.d/debian.sources"
}
],
"standard-repos": [
{
"description": "This is the default, stable, and recommended repository, available for all Proxmox subscription users.",
"handle": "enterprise",
"name": "Enterprise",
"status": 0
},
{
"description": "This is the recommended repository for testing and non-production use. Its packages are not as heavily tested and validated as the production ready enterprise repository. You don't need a subscription key to access this repository.",
"handle": "no-subscription",
"name": "No-Subscription",
"status": 1
},
{
"description": "This repository contains the latest packages and is primarily used for test labs and by developers to test new features.",
"handle": "test",
"name": "Test"
},
{
"description": "This repository holds the production-ready Proxmox Ceph Squid packages.",
"handle": "ceph-squid-enterprise",
"name": "Ceph Squid Enterprise",
"status": 0
},
{
"description": "This repository holds the Proxmox Ceph Squid packages intended for non-production use.",
"handle": "ceph-squid-no-subscription",
"name": "Ceph Squid No-Subscription"
},
{
"description": "This repository contains the Ceph Squid packages before they are moved to the main repository.",
"handle": "ceph-squid-test",
"name": "Ceph Squid Test"
},
{
"description": "This repository holds the production-ready Proxmox Ceph Tentacle packages.",
"handle": "ceph-tentacle-enterprise",
"name": "Ceph Tentacle Enterprise"
},
{
"description": "This repository holds the Proxmox Ceph Tentacle packages intended for non-production use.",
"handle": "ceph-tentacle-no-subscription",
"name": "Ceph Tentacle No-Subscription"
},
{
"description": "This repository contains the Ceph Tentacle packages before they are moved to the main repository.",
"handle": "ceph-tentacle-test",
"name": "Ceph Tentacle Test"
}
]
}
tests/test_repository_callers.py
+149
View File
@@ -0,0 +1,149 @@
"""Isolated repository-check consumer gates; no full installer is sourced.
Synthetic policy decisions and mocked APT only. These are not live-host tests.
"""
import os
from pathlib import Path
import re
import shutil
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[1]
HELPER = ROOT / 'scripts/global/repository-functions.sh'
def function(path, name):
source = (ROOT / path).read_text()
match = re.search(rf'^( *){re.escape(name)}\(\) \{{', source, re.MULTILINE)
if not match:
raise AssertionError(f'Missing function {name} in {path}')
end = source.index('\n' + match[1] + '}', match.end()) + len(match[1]) + 2
return source[match.start():end]
class RepositoryCallerTest(unittest.TestCase):
def test_upgrade_repair_dispatch_reports_repository_failure_not_success(self):
for path in ('scripts/utilities/upgrade_pve8_to_pve9.sh',
'scripts/utilities/pve8to9_check.sh'):
source = (ROOT / path).read_text()
commands = re.findall(r'repair_commands\+=\("([^"\n]*ensure_repositories[^"\n]*)"\)', source)
gates = re.findall(r'if eval "\$\{repair_commands\[\$i\]\}"; then.*?\n\s*fi',
source, re.DOTALL)
self.assertTrue(commands, path)
self.assertEqual(len(commands), len(gates), path)
for command, gate in zip(commands, gates):
with self.subTest(path=path, command=command):
script = f'''
cleanup_duplicate_repos() {{ :; }}
ensure_repositories() {{ return 100; }}
translate() {{ printf %s "$1"; }}
msg_ok() {{ printf '%s\\n' "$1"; }}
msg_error() {{ printf '%s\\n' "$1"; }}
repair_commands=('{command}'); repair_descriptions=(fixture)
i=0; repair_success=0
{gate}
exit "$repair_success"
'''
result = subprocess.run(['/bin/bash', '-c', script], capture_output=True,
text=True, timeout=15)
self.assertNotEqual(result.returncode, 0)
self.assertIn('Failed', result.stdout)
self.assertNotIn('Success', result.stdout)
def test_changed_refresh_failure_stops_all_package_consumers(self):
consumers = [
('scripts/post_install/auto_post_install.sh', 'setup_proxmox_repositories', ''),
('scripts/post_install/customizable_post_install.sh', 'setup_proxmox_repositories', ''),
('scripts/utilities/system_utils.sh', 'install_utility_group', 'fixture fixturepkg'),
('scripts/utilities/system_utils.sh', 'install_selected_utilities', 'fixturepkg'),
('scripts/post_install/customizable_post_install.sh', 'install_system_utils', ''),
('scripts/menus/network_menu.sh', '_ensure_network_tool', 'fixturepkg'),
('scripts/utilities/import_vm_ova_ovf.sh', 'ensure_gawk', ''),
('scripts/gpu_tpu/nvidia_installer.sh', 'ensure_repos_and_headers', ''),
('scripts/global/update-pve-safe.sh', 'update_pve_safe', ''),
]
for path, name, args in consumers:
for pipefail in (False, True):
with self.subTest(path=path, function=name, pipefail=pipefail):
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as directory:
d = Path(directory)
# Only allow read-only pipeline/log helpers from PATH.
bindir = d / 'bin'
bindir.mkdir()
for cmd in ('grep', 'head', 'tee', 'date', 'awk', 'rm'):
target = shutil.which(cmd)
if target is None:
self.fail(f'Missing fixture command: {cmd}')
(bindir / cmd).symlink_to(target)
body = function(path, name)
if name == 'install_system_utils':
# Only the actual pre-install gate, before mktemp and
# package iteration, is in scope for this fixture.
self.assertIn(' new_packages_tmp="$(mktemp)"', body)
body = body.split(' new_packages_tmp="$(mktemp)"', 1)[0]
body += ' install_single_package fixturepkg\n}'
if name == 'update_pve_safe':
self.assertEqual(body.count('/var/log/'), 1)
self.assertEqual(body.count('/tmp/proxmenux_screen_capture_'), 1)
body = body.replace('/var/log/', f'{d}/').replace(
'/tmp/proxmenux_screen_capture_', f'{d}/screen_')
script = f'''source "{HELPER}"
set -e; set {'-o' if pipefail else '+o'} pipefail
LOCAL_SCRIPTS="$TEST_ROOT/missing"
LOG_FILE="$TEST_ROOT/log"; screen_capture="$TEST_ROOT/screen"
FORMAT_TYPE=exfat; INTEL_GPU_PRESENT=true
INTEL_GPU_TOOLS_INSTALLED=false; PROXMENUX_UTILS=(fixturepkg:fixturepkg:fixture)
for fn in clear show_proxmenux_logo msg_title msg_info msg_info2 msg_info3 \\
msg_warn msg_success cleanup sleep pmx_journal_context detect_intel_gpus \\
check_intel_gpu_tools_installed download_common_functions source_install_functions; do
eval "$fn() {{ :; }}"
done
translate() {{ printf %s "$1"; }}
msg_error() {{ printf '%s\\n' "$1" >&2; }}
msg_ok() {{ printf '%s\\n' "$1"; }}
register_tool() {{ printf 'register %s\\n' "$*" >> "$TEST_ROOT/calls"; }}
chmod() {{ printf 'chmod\\n' >> "$TEST_ROOT/calls"; }}
hostname() {{ printf fixture; }}
pveversion() {{ echo pve-manager/9.0.0; }}
is_web_mode() {{ return 0; }}
hybrid_yesno() {{ printf 'prompt\\n' >> "$TEST_ROOT/calls"; }}
repository_policy() {{
printf '%s\\n' "$1" >> "$TEST_ROOT/calls"
[[ "$1" == plan ]] && echo offer || echo changed
}}
apt-get() {{ printf 'apt %s\\n' "$*" >> "$TEST_ROOT/calls"; return 100; }}
command() {{
if [[ "$1" == -v ]]; then return 1; fi
builtin command "$@"
}}
dialog() {{ printf fixturepkg >&2; }}
df() {{ printf 'Filesystem 1K-blocks Used Available Use%% Mounted on\\n';
printf 'fixture 9999999 1 9999999 1%% /\\n'; }}
dpkg() {{ return 1; }}
dpkg-query() {{ return 1; }}
install_single_package() {{ printf 'install\\n' >> "$TEST_ROOT/calls"; }}
install_intel_gpu_tools() {{ printf 'install\\n' >> "$TEST_ROOT/calls"; }}
pmx_install_pkg() {{ printf 'install\\n' >> "$TEST_ROOT/calls"; }}
{body}
if {name} {args}; then exit 0; else exit $?; fi
'''
result = subprocess.run(['/bin/bash', '-c', script],
env=dict(os.environ, PATH=str(bindir), TEST_ROOT=str(d)),
stdin=subprocess.DEVNULL, text=True,
capture_output=True, timeout=15)
calls = (d / 'calls').read_text().splitlines()
self.assertEqual(calls[:4], ['plan', 'prompt', 'apply', 'apt update'], result.stderr)
self.assertNotEqual(result.returncode, 0, result.stderr)
self.assertNotIn('install', calls)
self.assertNotIn('chmod', calls)
self.assertFalse(any(c.startswith('register proxmox_repos true') for c in calls))
self.assertIn('Repository sources changed, but APT', result.stderr + result.stdout)
self.assertNotIn('repositories configured', result.stdout)
self.assertNotIn('installation completed', result.stdout.lower())
self.assertNotIn('verified.', result.stdout)
if __name__ == '__main__':
unittest.main()
tests/test_repository_preservation.py
+377 -8
View File
@@ -68,6 +68,206 @@ class RepositoryPolicyTest(unittest.TestCase):
with patch.object(self.module, 'run', side_effect=run): with patch.object(self.module, 'run', side_effect=run):
return self.module.evaluate(suite, apply=apply) 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): def test_concurrent_inventory_edit_aborts_before_second_mutation(self):
for edit in ('replace', 'reorder', 'unrelated'): for edit in ('replace', 'reorder', 'unrelated'):
with self.subTest(edit=edit): with self.subTest(edit=edit):
@@ -133,6 +333,17 @@ class RepositoryPolicyTest(unittest.TestCase):
self.assertEqual(self.run_policy(), 'offer') self.assertEqual(self.run_policy(), 'offer')
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls)) 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): def test_consented_switch_disables_enterprise_then_adds_public_pve(self):
self.assertEqual(self.run_policy(apply=True), 'changed') self.assertEqual(self.run_policy(apply=True), 'changed')
writes = [c for c in self.calls if c[1] in ('create', 'set')] writes = [c for c in self.calls if c[1] in ('create', 'set')]
@@ -178,11 +389,29 @@ class RepositoryPolicyTest(unittest.TestCase):
self.run_policy(apply=True) self.run_policy(apply=True)
self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls)) self.assertFalse(any(c[1] in ('create', 'set') for c in self.calls))
def test_existing_public_or_test_preserved_without_switch(self): def test_existing_public_or_test_preserved_without_subscription_lookup(self):
for channel in ('pve-no-subscription', 'pve-test'): # Synthetic inventory only: this does not model real pvesh serialization.
with self.subTest(channel=channel): for channel in ('pve-no-subscription', 'pve-test', 'pvetest'):
self.repositories['files'][0]['repositories'] = [repo(channel)] for subscription in ('active', 'notfound', 'expired', 'new', 'suspended',
self.assertEqual(self.run_policy(), 'preserve') '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): def test_legacy_pvetest_list_on_pve8_is_preserved(self):
self.repositories = fixture( self.repositories = fixture(
@@ -197,7 +426,8 @@ class RepositoryPolicyTest(unittest.TestCase):
self.assertFalse(any(c[1] == 'create' for c in self.calls)) self.assertFalse(any(c[1] == 'create' for c in self.calls))
def test_unrecognized_and_failed_subscription_cannot_mutate(self): def test_unrecognized_and_failed_subscription_cannot_mutate(self):
for response in ('status: unknown\n', 'status: invalid\n', 'status: active\nstatus: notfound\n', 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')): 'key: secret\n', RuntimeError('service unavailable')):
with self.subTest(response=response): with self.subTest(response=response):
self.subscription = response self.subscription = response
@@ -334,6 +564,143 @@ class CallerPropagationTest(unittest.TestCase):
class ShellFlowTest(unittest.TestCase): 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): def test_web_consent_applies_only_after_prompt(self):
helper = ROOT / 'scripts/global/repository-functions.sh' helper = ROOT / 'scripts/global/repository-functions.sh'
with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d: with tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR')) as d:
@@ -343,13 +710,14 @@ class ShellFlowTest(unittest.TestCase):
translate() {{ printf %s "$1"; }} translate() {{ printf %s "$1"; }}
msg_error() {{ :; }} msg_error() {{ :; }}
is_web_mode() {{ return 0; }} is_web_mode() {{ return 0; }}
hybrid_yesno() {{ printf 'prompt\\n' >> "{log}"; [[ "$2" == *'This host has no subscription, switch to the no-subscription repository?'* ]]; }} 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; }} repository_policy() {{ printf '%s\\n' "$1" >> "{log}"; [[ "$1" == plan ]] && echo offer || echo changed; }}
apt-get() {{ printf 'apt %s\\n' "$*" >> "{log}"; }}
ensure_repositories ensure_repositories
''' '''
result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True) result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True)
self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(log.read_text().splitlines(), ['plan', 'prompt', 'apply']) self.assertEqual(log.read_text().splitlines(), ['plan', 'prompt', 'apply', 'apt update'])
def test_refusal_and_noninteractive_never_apply(self): def test_refusal_and_noninteractive_never_apply(self):
helper = ROOT / 'scripts/global/repository-functions.sh' helper = ROOT / 'scripts/global/repository-functions.sh'
@@ -370,6 +738,7 @@ class ShellFlowTest(unittest.TestCase):
is_web_mode() {{ {'return 0' if web else 'return 1'}; }} is_web_mode() {{ {'return 0' if web else 'return 1'}; }}
hybrid_yesno() {{ {'return 0' if confirm else 'return 1'}; }} hybrid_yesno() {{ {'return 0' if confirm else 'return 1'}; }}
repository_policy() {{ python3 "{mock}" "{log}" "$1"; }} repository_policy() {{ python3 "{mock}" "{log}" "$1"; }}
apt-get() {{ printf 'apt %s\\n' "$*" >> "{log}"; }}
ensure_repositories ensure_repositories
''' '''
result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True) result = subprocess.run(['bash', '-c', script], input='', text=True, capture_output=True)