New version 1.2.8

New version 1.2.8
This commit is contained in:
MacRimi
2026-11-02 08:11:10 +01:00
committed by GitHub
345 changed files with 36365 additions and 2550 deletions
@@ -102,6 +102,12 @@ PROTECTED_TECHNICAL_TERMS = (
"WebUI",
"Intel",
"CUDA",
"ROCm",
"Vulkan",
"SYCL",
"QSV",
"OpenVINO",
"OpenCL",
"KFD",
"GPU",
"CPU",
@@ -2,7 +2,8 @@
"""Build a verified ProxMenux LXC application-version tracking catalog.
The source of truth is a pinned snapshot of community-scripts/ProxmoxVE,
downloaded through the GitHub REST API. Only ct/*.sh launchers are considered.
plus community-scripts/core for its shared version helpers after the split.
Both are downloaded through the GitHub REST API. Only ct/*.sh launchers are considered.
An operational hint is emitted only when two independent pieces of the helper
scripts agree:
@@ -37,6 +38,7 @@ from typing import Any, Iterable
DEFAULT_REPOSITORY = "community-scripts/ProxmoxVE"
DEFAULT_CORE_REPOSITORY = "community-scripts/core"
DEFAULT_REF = "main"
API_VERSION = "2022-11-28"
USER_AGENT = "ProxMenux-app-tracking-catalog/1.0"
@@ -169,6 +171,7 @@ def obtain_snapshot(
repository: str,
ref: str,
cache_dir: Path,
required_paths: tuple[str, ...] = ("ct",),
) -> tuple[Path, str, str]:
commit = client.json(f"/repos/{repository}/commits/{ref}")
commit_sha = str(commit.get("sha") or "")
@@ -177,7 +180,7 @@ def obtain_snapshot(
snapshot_dir = cache_dir / repository.replace("/", "--") / commit_sha
marker = snapshot_dir / ".snapshot-complete"
if marker.exists() and (snapshot_dir / "ct").is_dir():
if marker.exists() and all((snapshot_dir / path).exists() for path in required_paths):
return snapshot_dir, commit_sha, marker.read_text(encoding="utf-8").strip()
payload = client.request(
@@ -190,6 +193,8 @@ def obtain_snapshot(
temp_dir = Path(tempfile.mkdtemp(prefix=f"{commit_sha}.tmp-", dir=temp_parent))
try:
safe_extract_tar(payload, temp_dir)
if not all((temp_dir / path).exists() for path in required_paths):
raise CatalogError(f"Invalid source snapshot for {repository}@{commit_sha}")
(temp_dir / ".snapshot-complete").write_text(archive_sha + "\n", encoding="utf-8")
if snapshot_dir.exists():
shutil.rmtree(snapshot_dir)
@@ -521,23 +526,66 @@ def repo_from_helper(item: dict[str, Any]) -> str:
return match.group(1) if match else ""
def shared_helper_path(source: Path, core_source: Path | None = None) -> Path:
"""Locate source text only; never source or execute upstream shell code."""
candidates = [source / "misc" / "tools.func"]
if core_source is not None:
candidates.extend(core_source / path for path in ("lib/forge.func", "lib/tools.func"))
for candidate in candidates:
if candidate.is_file():
return candidate
raise CatalogError(
"Shared version helpers not found. Older trees use misc/tools.func; "
"current trees require community-scripts/core (--core-source-dir for a local checkout)."
)
def shell_function(text: str, name: str) -> str:
match = re.search(
rf"(?ms)^(?:function\s+)?{re.escape(name)}\s*\(\)\s*\{{[^\n]*\n(.*?)^\}}",
text,
)
return match.group(1) if match else ""
def verify_shared_helper(path: Path) -> None:
"""Check both ends of the cache contract, including the new forge wrappers."""
text = read_text(path)
read_body = shell_function(text, "check_for_gh_release")
write_body = shell_function(text, "fetch_and_deploy_gh_release")
if re.search(r'\bcheck_for_release\s+github\s+"\$1"\s+"\$2"', read_body):
read_body = shell_function(text, "check_for_release")
if re.search(r'\bfetch_and_deploy_release\s+github\s+"\$1"\s+"\$2"', write_body):
write_body = shell_function(text, "fetch_and_deploy_release")
normalization = r'app_lc="\$\(echo "\$\{app,,\}" \| tr -d [\'\"] [\'\"]\)"'
valid = (
re.search(normalization, read_body)
and re.search(normalization, write_body)
and re.search(r'(?:local\s+)?current_file="\$HOME/\.\$\{app_lc\}"', read_body)
and re.search(r'(?:local\s+)?version_file="\$HOME/\.\$\{app_lc\}"', write_body)
and re.search(r'echo "\$version" >"\$version_file"', write_body)
and re.search(r'\$current_file', read_body)
)
if not valid:
raise CatalogError(f"Could not verify the shared GitHub release cache contract in {path}")
def build_catalog(
source: Path,
helpers: dict[str, dict[str, Any]],
*,
core_source: Path | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
catalog: dict[str, Any] = {}
v2_apps: dict[str, Any] = {}
records: list[dict[str, Any]] = []
launchers = sorted((source / "ct").glob("*.sh"))
tools_func = source / "misc" / "tools.func"
tools_text = read_text(tools_func)
shared_helper_verified = bool(
re.search(r'local version_file="\$HOME/\.\$\{app_lc\}"', tools_text)
and re.search(r'echo "\$version" >"\$version_file"', tools_text)
and re.search(r'local current_file="\$HOME/\.\$\{app_lc\}"', tools_text)
tools_func = shared_helper_path(source, core_source)
verify_shared_helper(tools_func)
helper_contract = (
"misc/tools.func" if tools_func == source / "misc" / "tools.func"
else f"core/{tools_func.relative_to(core_source).as_posix()}"
)
if not shared_helper_verified:
raise CatalogError("Could not verify the shared GitHub release cache contract in misc/tools.func")
for launcher_path in launchers:
slug = launcher_path.stem
@@ -696,7 +744,7 @@ def build_catalog(
"evidence": {
"install": f"install/{installer_path.name}: matching deploy helper",
"update": f"ct/{launcher_path.name}: matching release check",
"contract": "misc/tools.func: shared version-cache contract",
"contract": f"{helper_contract}: shared version-cache contract",
},
}
)
@@ -712,7 +760,7 @@ def build_catalog(
"evidence": {
"install": f"install/{installer_path.name}: fetch_and_deploy_gh_release({release.app}, {release.repo})",
"update": f"ct/{launcher_path.name}: check_for_gh_release({release.app}, {release.repo})",
"contract": "misc/tools.func writes and reads /root/.<normalized-app>",
"contract": f"{helper_contract} writes and reads /root/.<normalized-app>",
},
}
)
@@ -874,7 +922,7 @@ def build_catalog(
"status_counts": status_counts,
"method_counts": method_counts,
"reason_counts": reason_counts,
"shared_release_cache_contract_verified": shared_helper_verified,
"shared_release_cache_contract_verified": True,
"helper_upstream_tag_validation": tag_validation,
},
"records": records,
@@ -1307,6 +1355,12 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repository", default=DEFAULT_REPOSITORY)
parser.add_argument("--ref", default=DEFAULT_REF)
parser.add_argument("--core-repository", default=DEFAULT_CORE_REPOSITORY)
parser.add_argument("--core-ref", default=DEFAULT_REF)
parser.add_argument(
"--core-source-dir", type=Path,
help="Local community-scripts/core checkout for split source trees (no download)",
)
parser.add_argument(
"--source-dir",
type=Path,
@@ -1346,6 +1400,7 @@ def main(argv: list[str] | None = None) -> int:
client = GitHubClient()
commit_sha = "local-source"
archive_sha = "not-applicable"
core_provenance = None
try:
if args.source_dir:
source = args.source_dir.resolve()
@@ -1355,10 +1410,27 @@ def main(argv: list[str] | None = None) -> int:
source, commit_sha, archive_sha = obtain_snapshot(
client, args.repository, args.ref, args.cache_dir.expanduser().resolve()
)
if not (source / "ct").is_dir() or not (source / "misc" / "tools.func").is_file():
if not (source / "ct").is_dir() or not (source / "install").is_dir():
raise CatalogError(f"Not a valid ProxmoxVE source tree: {source}")
core_source = None
if not (source / "misc" / "tools.func").is_file():
if args.core_source_dir:
core_source = args.core_source_dir.resolve()
core_commit, core_archive = "local-source", "not-applicable"
elif args.source_dir:
raise CatalogError("Split local source tree: provide --core-source-dir for its shared helpers")
else:
core_source, core_commit, core_archive = obtain_snapshot(
client, args.core_repository, args.core_ref,
args.cache_dir.expanduser().resolve(), required_paths=("lib/forge.func",),
)
core_provenance = {
"repository": args.core_repository, "ref": args.core_ref,
"commit_sha": core_commit, "archive_sha256": core_archive,
"source_dir": str(core_source),
}
helpers = load_helper_catalog(args.helpers_cache)
catalog, audit, v2 = build_catalog(source, helpers)
catalog, audit, v2 = build_catalog(source, helpers, core_source=core_source)
audit["demoted_helper_markers"] = (
[] if args.include_helper_markers else demote_generic_helper_markers(catalog, v2, audit)
)
@@ -1376,6 +1448,7 @@ def main(argv: list[str] | None = None) -> int:
"ref": args.ref,
"commit_sha": commit_sha,
"archive_sha256": archive_sha,
"core": core_provenance,
"source_dir": str(source),
"generated_at_unix": int(time.time()),
"generator_sha256": sha256_file(Path(__file__).resolve()),
+15
View File
@@ -13,6 +13,15 @@ import sys
import tempfile
PYTHON_FILES = (
"tests/test_host_update_jobs.py",
"tests/test_host_update_transport.py",
"tests/test_host_update_shell.py",
"tests/test_host_update_auth.py",
"tests/test_kernel_reboot.py",
"tests/test_reboot_health_refresh.py",
"tests/test_host_update_controls.py",
"tests/test_host_update_packaging.py",
"tests/test_host_update_version.py",
"tests/test_lxc_update_jobs.py",
"tests/test_post_install_i18n.py",
"tests/test_remaining_cli_i18n.py",
@@ -34,6 +43,11 @@ PYTHON_FILES = (
"AppImage/scripts/tests/test_health_persistence_recurrence.py",
)
NODE_FILES = (
"tests/test_host_update_ui.cjs",
"tests/test_reboot_health_refresh.cjs",
"tests/test_host_update_controls.cjs",
"tests/test_reboot_recovery.cjs",
"tests/test_proxmenux_version_row.cjs",
"tests/test_lxc_update_jobs.cjs",
"tests/test_api_import_boundaries.cjs",
"tests/test_api_errors.cjs",
@@ -58,6 +72,7 @@ NODE_FILES = (
"tests/test_borg_ssh_guidance.cjs",
"tests/test_storage_messages.cjs",
"tests/test_health_text.cjs",
"tests/test_overview_chart_axis_gutters.cjs",
)
# unittest's CLI accepts a missing pattern as a successful zero-test run.
@@ -1,4 +1,5 @@
"""Inert infrastructure for actual endpoint/manual/queued/SQLite release seams."""
import ast
import datetime
import sqlite3
import tempfile
@@ -36,6 +37,10 @@ def restore_event(warnings):
def deliver(event_type, data, severity='INFO', language='en', manual=False, quiet=False, quiet_before=()):
"""Execute real dispatch/manual and optional real SQLite buffer+flush."""
captured = []
delivery_errors = []
def record_history(event_type, channel, title, body, severity, success, error, source):
if not success:
delivery_errors.append(f'{channel}: {error}')
channel = object.__new__(EmailChannel); channel.subject_prefix = '[ProxMenux]'
def sink(title, body, severity, data):
markup = channel._format_html(title, body, severity, data)
@@ -48,12 +53,25 @@ def deliver(event_type, data, severity='INFO', language='en', manual=False, quie
_claim_delivery=lambda event:'inert', _finish_delivery_claim=lambda *a,**kw:None,
_notification_language=lambda:language, _build_ai_config=lambda:{'ai_enabled':'false'},
_in_quiet_hours=lambda channel:quiet, _should_buffer_for_digest=lambda *a:False,
_record_history=lambda *a:None, _stats={'total_sent':0,'total_errors':0},
_record_history=record_history, _stats={'total_sent':0,'total_errors':0},
is_event_enabled=lambda event:True)
ns = dict(vars(typing), NotificationEvent=types.SimpleNamespace, TEMPLATES=templates.TEMPLATES, render_template=templates.render_template,
resolve_notification_hostname=lambda host, config:host or 'node-a',
enrich_with_emojis=templates.enrich_with_emojis, datetime=datetime.datetime,
_should_bypass_ai=lambda event:True, _AI_BYPASS_EVENTS=frozenset({'backup_complete','backup_fail'}))
format_backup_diagnostics_with_ai=templates.format_backup_diagnostics_with_ai,
format_with_ai_full=templates.format_with_ai_full,
# AI is disabled in this fixture. Execute the actual formatting helpers
# synchronously, without a worker pool, providers or operational imports.
_format_with_ai_bounded=lambda function,*args,**kwargs:function(*args,**kwargs),
enrich_context_for_ai=lambda **kwargs:'')
manager_path = SCRIPTS/'notification_manager.py'
for node in ast.parse(manager_path.read_text()).body:
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id in ('_AI_BYPASS_EVENTS', '_AI_NO_SUGGESTION_EVENTS'):
ns[target.id] = frozenset(ast.literal_eval(node.value.args[0]))
for name in ('_should_bypass_ai', '_format_backup_with_ai_bounded'):
extract(manager_path, name, None, ns)
for name in ('_dispatch_to_channels', '_dispatch_event', 'send_notification'):
setattr(manager, name, types.MethodType(extract(SCRIPTS/'notification_manager.py', name, 'NotificationManager', ns), manager))
with tempfile.TemporaryDirectory(prefix='notification-final-') as scratch:
@@ -73,10 +91,11 @@ def deliver(event_type, data, severity='INFO', language='en', manual=False, quie
if manual:
assert not quiet
result = manager.send_notification(event_type, severity, '', '', dict(data))
assert result['success']
assert result['success'], result.get('errors')
else:
event = types.SimpleNamespace(event_type=event_type, severity=severity, data=dict(data), source='inert', entity_type='node', entity_id='', event_id='inert', fingerprint='inert')
manager._dispatch_event(event)
assert not delivery_errors, '; '.join(delivery_errors)
if quiet:
conn = sqlite3.connect(db); rows = conn.execute('SELECT event_type,severity,title,body FROM quiet_pending').fetchall(); conn.close()
assert len(rows)==1+len(quiet_before) and not captured
@@ -0,0 +1,169 @@
"""Source layout regressions for the Helper-Scripts/core repository split."""
import contextlib
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch
SCRIPT = Path(__file__).resolve().parents[1] / "generate_app_tracking_catalog.py"
SPEC = importlib.util.spec_from_file_location("tracking_sources_under_test", SCRIPT)
generator = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = generator
SPEC.loader.exec_module(generator)
NORMALIZE = ''' app_lc="$(echo "${app,,}" | tr -d ' ')"'''
READ_BODY = NORMALIZE + '''
current_file="$HOME/.${app_lc}"
current="$(<"$current_file")"
'''
WRITE_BODY = NORMALIZE + '''
version_file="$HOME/.${app_lc}"
echo "$version" >"$version_file"
'''
def function(name, body):
return f"{name}() {{\n{body}\n}}\n"
LEGACY = (
function("check_for_gh_release", READ_BODY.replace(' current_file=', ' local current_file='))
+ function("fetch_and_deploy_gh_release", WRITE_BODY.replace(' version_file=', ' local version_file='))
)
SPLIT = (
function("check_for_gh_release", ' check_for_release github "$1" "$2" "${3:-}"')
+ function("fetch_and_deploy_gh_release", ' fetch_and_deploy_release github "$1" "$2" "${3:-tarball}"')
+ function("check_for_release", READ_BODY)
+ function("fetch_and_deploy_release", WRITE_BODY)
)
class TrackingSourcesTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name)
self.apps = self.root / "apps"
self.core = self.root / "core"
self.write(self.apps / "ct/example.sh", '''APP="Example"
check_for_gh_release "Example" "example/example"
''')
self.write(self.apps / "install/example-install.sh", '''
fetch_and_deploy_gh_release "Example" "example/example" "tarball" "latest"
''')
def write(self, path, content):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return path
def run_main(self, *extra):
output = self.root / "output.json"
args = ["--source-dir", str(self.apps), "--output", str(output),
"--include-helper-markers",
"--audit-output", str(self.root / "audit.json"),
"--v2-output", str(self.root / "v2.json"), *extra]
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
result = generator.main(args)
return result, output
def test_legacy_monolithic_contract_is_supported(self):
self.write(self.apps / "misc/tools.func", LEGACY)
result, output = self.run_main()
self.assertEqual(result, 0)
self.assertEqual(json.loads(output.read_text())["example"]["file_path"], "/root/.example")
def test_split_contract_builds_catalog_and_records_actual_evidence(self):
self.write(self.core / "lib/forge.func", SPLIT)
result, output = self.run_main("--core-source-dir", str(self.core))
self.assertEqual(result, 0)
self.assertEqual(json.loads(output.read_text())["example"]["repo"], "example/example")
audit = json.loads((self.root / "audit.json").read_text())
self.assertEqual(audit["provenance"]["core"]["source_dir"], str(self.core.resolve()))
v2 = (self.root / "v2.json").read_text()
self.assertIn("core/lib/forge.func", v2)
self.assertNotIn("misc/tools.func", v2)
def test_unsplit_core_tools_library_is_supported(self):
path = self.write(self.core / "lib/tools.func", LEGACY)
self.assertEqual(generator.shared_helper_path(self.apps, self.core), path)
generator.verify_shared_helper(path)
def test_changed_contract_never_overwrites_existing_outputs(self):
for old, new in [('echo "$version" >"$version_file"', 'echo "$version" >/tmp/version'),
('current_file="$HOME/.${app_lc}"', 'current_file="/tmp/version"'),
('tr -d \' \'', 'tr -d \'-\'')]:
with self.subTest(change=old):
self.write(self.core / "lib/forge.func", SPLIT.replace(old, new))
output = self.write(self.root / "output.json", '{"keep": true}\n')
result, _ = self.run_main("--core-source-dir", str(self.core))
self.assertEqual(result, 1)
self.assertEqual(output.read_text(), '{"keep": true}\n')
def test_matching_text_in_unrelated_functions_is_not_evidence(self):
path = self.write(self.core / "lib/forge.func", SPLIT.replace(
function("check_for_release", READ_BODY), function("unrelated", READ_BODY)))
with self.assertRaises(generator.CatalogError):
generator.verify_shared_helper(path)
def test_split_local_tree_requires_explicit_local_core_without_network(self):
with patch.object(generator.GitHubClient, "request", side_effect=AssertionError("Network")):
result, output = self.run_main()
self.assertEqual(result, 1)
self.assertFalse(output.exists())
def test_invalid_local_core_fails_without_writing_a_catalog(self):
result, output = self.run_main("--core-source-dir", str(self.core))
self.assertEqual(result, 1)
self.assertFalse(output.exists())
def test_snapshot_cache_accepts_core_without_ct_directory(self):
sha = "a" * 40
source = self.root / "cache/community-scripts--core" / sha
self.write(source / "lib/forge.func", SPLIT)
self.write(source / ".snapshot-complete", "archive-hash\n")
client = Mock()
client.json.return_value = {"sha": sha}
path, commit, archive = generator.obtain_snapshot(
client, "community-scripts/core", "main", self.root / "cache",
required_paths=("lib/forge.func",))
self.assertEqual((path, commit, archive), (source, sha, "archive-hash"))
client.request.assert_not_called()
def test_remote_split_source_pins_both_repositories(self):
self.write(self.core / "lib/forge.func", SPLIT)
output = self.root / "remote.json"
with patch.object(generator, "obtain_snapshot", side_effect=[
(self.apps, "a" * 40, "apps-archive"),
(self.core, "b" * 40, "core-archive"),
]) as snapshots, contextlib.redirect_stdout(io.StringIO()):
result = generator.main([
"--output", str(output), "--include-helper-markers",
"--audit-output", str(self.root / "remote-audit.json"),
"--v2-output", str(self.root / "remote-v2.json"),
])
self.assertEqual(result, 0)
provenance = json.loads((self.root / "remote-audit.json").read_text())["provenance"]
self.assertEqual(provenance["commit_sha"], "a" * 40)
self.assertEqual(provenance["core"]["commit_sha"], "b" * 40)
self.assertEqual(provenance["core"]["archive_sha256"], "core-archive")
self.assertEqual(snapshots.call_args_list[1].args[1], "community-scripts/core")
self.assertEqual(snapshots.call_args_list[1].kwargs["required_paths"], ("lib/forge.func",))
def test_invalid_download_is_not_marked_complete(self):
client = Mock()
client.json.return_value = {"sha": "b" * 40}
client.request.return_value = b"invalid-layout"
with patch.object(generator, "safe_extract_tar"):
with self.assertRaises(generator.CatalogError):
generator.obtain_snapshot(client, "community-scripts/core", "main", self.root / "cache",
required_paths=("lib/forge.func",))
self.assertEqual(list((self.root / "cache").rglob(".snapshot-complete")), [])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,102 @@
"""A schedule systemd rejects leaves a timer that never fires and reports no
error, so the wizard asks for it again instead of saving it."""
from pathlib import Path
import subprocess
import tempfile
import unittest
SCHEDULER = Path(__file__).resolve().parents[3] / 'scripts/backup_restore/backup_scheduler.sh'
ANALYZE = '#!/bin/sh\ncase "$2" in "daily 02:00") exit 1;; esac\nexit 0\n'
def function(name):
text = SCHEDULER.read_text()
start = text.index(name + '() {')
return text[start:text.index('\n}', start) + 2] + '\n'
class ScheduleValidation(unittest.TestCase):
def valid(self, expression, **environment):
with tempfile.TemporaryDirectory() as directory:
analyze = Path(directory) / 'systemd-analyze'
analyze.write_text(ANALYZE)
analyze.chmod(0o755)
return subprocess.run(
['/bin/bash', '-c', function('_calendar_is_valid') + '_calendar_is_valid "$1"', 'test', expression],
env={'PATH': f'{directory}:/usr/bin:/bin', **environment},
).returncode == 0
def test_systemd_decides_what_is_valid(self):
self.assertTrue(self.valid('02:30'))
self.assertFalse(self.valid('daily 02:00'))
def test_a_scheduler_without_systemctl_is_not_checked(self):
self.assertTrue(self.valid('daily 02:00', PMX_BACKUP_NO_SYSTEMCTL='1'))
def test_the_wizard_asks_again_before_saving_the_job(self):
text = SCHEDULER.read_text()
check = text.index('while ! _calendar_is_valid "$on_calendar"; do')
self.assertLess(check, text.index('"ON_CALENDAR=$on_calendar"'))
self.assertGreater(check, text.index('on_calendar=$(dialog'))
STUB_ANALYZE = '#!/bin/sh\ncase "$2" in daily\\ *|weekly\\ *|monthly\\ *) exit 1;; esac\nexit 0\n'
JOBS = {
'nas': 'ON_CALENDAR=daily\\ 02:00\nENABLED=1\n',
'week': 'ON_CALENDAR=weekly\\ 03:15\nENABLED=1\n',
'month': 'ON_CALENDAR=monthly\\ 04:00\nENABLED=0\n',
'good': 'ON_CALENDAR=Mon..Fri\\ 03:00\nENABLED=1\n',
'manual': 'ON_CALENDAR=daily\\ 05:00\nMANUAL_RUN=1\n',
'attached': 'ON_CALENDAR=daily\\ 06:00\nPVE_STORAGE=local\n',
'custom': 'ON_CALENDAR=every\\ night\nENABLED=1\n',
}
class ScheduleRepair(unittest.TestCase):
def run_repair(self):
names = ('_calendar_is_valid', '_scheduler_systemctl', '_job_env_get', '_job_is_attached',
'_write_job_units', '_repaired_calendar', '_repair_job_schedules')
self.folder = tempfile.TemporaryDirectory()
self.addCleanup(self.folder.cleanup)
root = Path(self.folder.name)
for name, body in (('systemd-analyze', STUB_ANALYZE),
('systemctl', f'#!/bin/sh\necho "$@" >> {root}/systemctl.log\n')):
(root / name).write_text(body)
(root / name).chmod(0o755)
jobs, units = root / 'jobs', root / 'units'
jobs.mkdir()
for job, body in JOBS.items():
(jobs / f'{job}.env').write_text(f'JOB_ID={job}\n{body}')
script = (f'JOBS_DIR={jobs}\nSYSTEMD_DIR={units}\nLOCAL_SCRIPTS={root}\nSCRIPT_DIR={root}\n'
'_job_file() { echo "${JOBS_DIR}/$1.env"; }\n'
'_service_file() { echo "${SYSTEMD_DIR}/proxmenux-backup-$1.service"; }\n'
'_timer_file() { echo "${SYSTEMD_DIR}/proxmenux-backup-$1.timer"; }\n'
+ ''.join(function(name) for name in names) + '_repair_job_schedules\n')
subprocess.run(['/bin/bash', '-c', script], check=True,
env={'PATH': f'{root}:/usr/bin:/bin'})
return jobs, units, root
def test_a_time_after_daily_weekly_or_monthly_is_repaired(self):
jobs, units, root = self.run_repair()
self.assertIn('ON_CALENDAR=02:00\n', (jobs / 'nas.env').read_text())
self.assertIn('OnCalendar=02:00\n', (units / 'proxmenux-backup-nas.timer').read_text())
self.assertIn('OnCalendar=Mon 03:15\n', (units / 'proxmenux-backup-week.timer').read_text())
self.assertIn('OnCalendar=*-*-01 04:00\n', (units / 'proxmenux-backup-month.timer').read_text())
log = (root / 'systemctl.log').read_text()
self.assertIn('enable --now proxmenux-backup-nas.timer', log)
self.assertNotIn('proxmenux-backup-month.timer', log)
def test_the_rest_of_the_jobs_are_left_as_they_are(self):
jobs, units, _ = self.run_repair()
for job in ('good', 'manual', 'attached', 'custom'):
self.assertEqual((jobs / f'{job}.env').read_text(), f'JOB_ID={job}\n{JOBS[job]}', job)
self.assertFalse((units / f'proxmenux-backup-{job}.timer').exists(), job)
def test_both_installers_repair_on_update(self):
root = SCHEDULER.parents[2]
for installer in ('install_proxmenux.sh', 'install_proxmenux_beta.sh'):
self.assertIn('backup_scheduler.sh" --repair-schedules', (root / installer).read_text())
if __name__ == '__main__':
unittest.main()
@@ -1,105 +0,0 @@
"""Log2RAM syncs only the archived journal files, so the active file it found
on disk when it was installed stays there and is restored at every boot.
journald then continues an old sequence and deletes its newest archives first.
The post-install makes the sync remove that file from the disk copy."""
from pathlib import Path
import re
import shutil
import stat
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[3]
SHARED = ROOT / "scripts/global/utils-install-functions.sh"
FLOWS = {"auto_post_install.sh": "install_log2ram_auto", "customizable_post_install.sh": "configure_log2ram"}
# The sync of Log2RAM in its journald-aware mode, as its project ships it.
JOURNALD_AWARE = '''#!/usr/bin/env bash
sync_to_disk() {
is_safe
local optional_params=()
if journald_logrotate; then
optional_params+=("--include=journal/*/*@*.journal")
optional_params+=("--exclude=journal/*/*")
fi
rsync -aXv --no-acls --delete-after "${optional_params[@]}" "${RAM_LOG}"/ "${HDD_LOG}"/
}
'''
NOT_AWARE = '''#!/usr/bin/env bash
sync_to_disk() {
rsync -aXv --delete-after "${RAM_LOG}"/ "${HDD_LOG}"/
}
'''
def gnu_sed():
return shutil.which("bash") and subprocess.run(["sed", "--version"], capture_output=True).returncode == 0
@unittest.skipUnless(gnu_sed(), "bash and GNU sed are needed")
class Log2ramJournalSyncTests(unittest.TestCase):
def correct(self, folder, content, times=1):
script = Path(folder) / "log2ram"
script.write_text(content)
script.chmod(0o755)
calls = "; ".join(['pmx_log2ram_drop_stale_journal "$1"; echo "rc=$?"'] * times)
# As the change journal does, the new content is read whole before it is written.
write = 'pmx_write_file() { local content; content="$(cat)"; printf "%s\\n" "$content" > "$1"; }'
run = subprocess.run(["bash", "-c", f'source "{SHARED}"; {write}; {calls}',
"bash", str(script)], capture_output=True, text=True, check=False)
return script, run.stdout.split()
def test_the_sync_removes_what_it_does_not_copy(self):
with tempfile.TemporaryDirectory() as folder:
script, results = self.correct(folder, JOURNALD_AWARE)
lines = script.read_text().splitlines()
position = lines.index(' optional_params+=("--exclude=journal/*/*")')
self.assertEqual(lines[position + 1], ' optional_params+=("--delete-excluded")')
self.assertEqual(lines[position + 2], " fi")
self.assertEqual(results, ["rc=0"])
self.assertEqual(subprocess.run(["bash", "-n", str(script)]).returncode, 0)
self.assertTrue(script.stat().st_mode & stat.S_IXUSR)
self.assertEqual(Path(str(script) + ".proxmenux.bak").read_text(), JOURNALD_AWARE)
def test_it_is_applied_once(self):
with tempfile.TemporaryDirectory() as folder:
script, results = self.correct(folder, JOURNALD_AWARE, times=2)
self.assertEqual(results, ["rc=0", "rc=1"])
self.assertEqual(script.read_text().count("--delete-excluded"), 1)
def test_a_sync_that_copies_every_file_is_left_as_it_is(self):
with tempfile.TemporaryDirectory() as folder:
script, results = self.correct(folder, NOT_AWARE)
self.assertEqual(results, ["rc=1"])
self.assertEqual(script.read_text(), NOT_AWARE)
self.assertFalse(Path(str(script) + ".proxmenux.bak").exists())
def test_both_flows_apply_it_to_new_and_existing_installations(self):
for name, function in FLOWS.items():
with self.subTest(flow=name):
text = (ROOT / "scripts/post_install" / name).read_text()
self.assertEqual(text.count("pmx_log2ram_drop_stale_journal"), 2)
# An existing installation is told that it needs a reboot.
self.assertIn('if pmx_log2ram_drop_stale_journal "$log2ram_bin"; then\n NECESSARY_REBOOT=1', text)
# The version is what offers the update to existing installations.
version = re.search(rf'{function}\(\) {{\n local FUNC_VERSION="([0-9.]+)"', text)
self.assertIsNotNone(version)
self.assertGreaterEqual(tuple(map(int, version.group(1).split("."))), (1, 7))
def test_rotated_logs_are_compressed_at_once_on_a_tmpfs_var_log(self):
# A rotated log kept uncompressed filled the 128M Log2RAM of a PBS.
for name, function in FLOWS.items():
with self.subTest(flow=name):
text = (ROOT / "scripts/post_install" / name).read_text()
self.assertNotIn("delaycompress", text)
version = re.search(rf'{function}\(\) {{\n local FUNC_VERSION="([0-9.]+)"', text)
self.assertGreaterEqual(tuple(map(int, version.group(1).split("."))), (1, 8))
if __name__ == "__main__":
unittest.main()
+125
View File
@@ -0,0 +1,125 @@
"""Log2RAM keeps /var/log in RAM while the journal and the task logs of
Proxmox VE and Proxmox Backup Server stay on disk behind a link. An update keeps the size
and interval the host already has."""
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[3]
KEEP = ROOT / "scripts/global/log2ram_keep_on_disk.sh"
SETUP = ROOT / "scripts/global/log2ram_setup.sh"
@unittest.skipUnless(shutil.which("bash"), "bash is needed")
class KeepOnDiskTests(unittest.TestCase):
def setUp(self):
folder = tempfile.TemporaryDirectory()
self.addCleanup(folder.cleanup)
self.root = Path(folder.name)
self.log = self.root / "var/log"
(self.log / "journal/abc").mkdir(parents=True)
(self.log / "journal/abc/system.journal").write_text("entries")
(self.log / "pveproxy").mkdir()
(self.log / "pveproxy/access.log").write_text("GET /")
def run_script(self, revert=False):
env = dict(os.environ, PMX_L2R_ROOT=str(self.root), PMX_L2R_REVERT="1" if revert else "0")
return subprocess.run(["bash", str(KEEP)], env=env, capture_output=True, text=True)
def test_the_journal_moves_to_disk_behind_a_link(self):
self.assertEqual(self.run_script().returncode, 0)
journal = self.log / "journal"
self.assertTrue(journal.is_symlink())
self.assertEqual(os.readlink(journal), "/var/hdd.journal")
self.assertEqual((self.root / "var/hdd.journal/abc/system.journal").read_text(), "entries")
self.assertTrue((self.log / "pveproxy/access.log").exists())
self.assertFalse((self.root / "var/hdd.proxmox-backup").exists())
def test_running_again_changes_nothing(self):
self.run_script()
self.assertEqual(self.run_script().returncode, 0)
self.assertEqual(os.readlink(self.log / "journal"), "/var/hdd.journal")
self.assertEqual(sorted(p.name for p in (self.root / "var/hdd.journal").iterdir()), ["abc"])
def test_the_task_logs_of_backup_server_stay_on_disk_too(self):
(self.log / "proxmox-backup/tasks").mkdir(parents=True)
(self.log / "proxmox-backup/tasks/active").write_text("task")
self.run_script()
self.assertEqual(os.readlink(self.log / "proxmox-backup"), "/var/hdd.proxmox-backup")
self.assertEqual((self.root / "var/hdd.proxmox-backup/tasks/active").read_text(), "task")
def test_files_written_back_to_the_folder_join_the_copy_on_disk(self):
self.run_script()
(self.log / "journal").unlink()
(self.log / "journal/def").mkdir(parents=True)
(self.log / "journal/def/system.journal").write_text("later")
self.run_script()
disk = self.root / "var/hdd.journal"
self.assertEqual(sorted(p.name for p in disk.iterdir()), ["abc", "def"])
self.assertTrue((self.log / "journal").is_symlink())
def test_a_host_without_a_journal_folder_gets_one_on_disk(self):
shutil.rmtree(self.log / "journal")
self.run_script()
self.assertTrue((self.root / "var/hdd.journal").is_dir())
self.assertTrue((self.log / "journal").is_symlink())
def test_the_task_logs_of_proxmox_ve_stay_on_disk(self):
(self.log / "pve/tasks").mkdir(parents=True)
(self.log / "pve/tasks/index").write_text("UPID")
self.run_script()
self.assertEqual(os.readlink(self.log / "pve"), "/var/hdd.pve")
self.assertEqual((self.root / "var/hdd.pve/tasks/index").read_text(), "UPID")
def test_removing_log2ram_puts_the_folders_back(self):
(self.log / "proxmox-backup").mkdir()
self.run_script()
self.assertEqual(self.run_script(revert=True).returncode, 0)
for name in ("journal", "proxmox-backup"):
self.assertFalse((self.log / name).is_symlink(), name)
self.assertTrue((self.log / name).is_dir(), name)
self.assertFalse((self.root / f"var/hdd.{name}").exists(), name)
self.assertEqual((self.log / "journal/abc/system.journal").read_text(), "entries")
@unittest.skipUnless(shutil.which("bash"), "bash is needed")
class KeptSettingsTests(unittest.TestCase):
def read(self, function, files):
with tempfile.TemporaryDirectory() as folder:
script = SETUP.read_text()
for path, content in files.items():
target = Path(folder) / path.lstrip("/")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content)
script = script.replace(path, str(target))
library = Path(folder) / "setup.sh"
library.write_text(script)
result = subprocess.run(["bash", "-c", f'source "{library}"; {function}'],
capture_output=True, text=True)
return result.stdout.strip()
def test_the_size_comes_from_the_installed_configuration(self):
self.assertEqual(self.read("pmx_log2ram_current_size", {"/etc/log2ram.conf": "SIZE=256M\n"}), "256")
def test_the_interval_comes_from_the_timer_of_this_version(self):
timer = "[Timer]\nOnCalendar=\nOnCalendar=*-*-* 00/6:55:00\n"
self.assertEqual(self.read("pmx_log2ram_current_hours",
{"/etc/systemd/system/log2ram-daily.timer.d/proxmenux.conf": timer}), "6")
def test_the_interval_of_an_earlier_version_is_kept(self):
cron = "MAILTO=\"\"\n0 */3 * * * root /usr/local/bin/log2ram write >/dev/null 2>&1\n"
self.assertEqual(self.read("pmx_log2ram_current_hours", {"/etc/cron.d/log2ram": cron}), "3")
def test_the_release_is_pinned_to_a_verified_commit(self):
text = SETUP.read_text()
self.assertIn('PMX_L2R_TAG="1.7.2"', text)
self.assertIn('PMX_L2R_COMMIT="f171449530f8984374c8f69fff3c92af813a694b"', text)
self.assertNotIn("install.sh", text.split("_pmx_log2ram_install_release() {", 1)[1].split("\n}\n", 1)[0])
if __name__ == "__main__":
unittest.main()
@@ -29,6 +29,10 @@ TRANSACTION = ROOT / 'oci/remote/oci_stack_transaction.py'
def extracted(path, name, scope):
if name == 'remove':
# Operational cache cleanup is exercised by the engine tests; keep wording tests inert.
scope.setdefault('compose_jobs', SimpleNamespace(forget_completed=lambda *args: None))
scope.setdefault('public_routes', SimpleNamespace(remove=lambda *args: None))
node = next(n for n in ast.parse(path.read_text()).body
if isinstance(n, ast.FunctionDef) and n.name == name)
exec(compile(ast.Module(body=[node], type_ignores=[]), str(path), 'exec'), scope)
@@ -71,6 +75,9 @@ class InventoryMessages(unittest.TestCase):
self.assertNotIn('Host directories that are kept, with their content:', text)
def test_removal_notice_qualifies_saved_record_source(self):
# Extracted operational functions must never inspect the real pmxcfs.
host = Path(self.enterContext(tempfile.TemporaryDirectory()))
inert_path = lambda value: host / Path(value).name if str(value).startswith('/etc/pve/') else Path(value)
events = []
scope = {'members_of': lambda root, vmid: (101, {}, [101]),
'instances': SimpleNamespace(read=lambda root, vmid: {'installation_id': 'owned'},
@@ -81,7 +88,7 @@ class InventoryMessages(unittest.TestCase):
'remove_owned_host_firewall': lambda primary: True,
'run': lambda *args: events.append(('run', args)),
'subprocess': SimpleNamespace(run=lambda *args, **kwargs: None),
'Path': Path, 'shutil': SimpleNamespace(rmtree=lambda path: None),
'Path': inert_path, 'shutil': SimpleNamespace(rmtree=lambda path: None),
'image_cache': SimpleNamespace(prune=lambda root, lock: []),
'work_backup': SimpleNamespace(forget=lambda root, vmid: None),
'oci_console': SimpleNamespace(remove_log=lambda vmid: events.append(('log', vmid))),
@@ -20,7 +20,7 @@ INSTALL = ROOT / 'oci/remote/install_oci.sh'
PREVIEW = 'Private network targeted for release if no other guest uses it:'
CONFIRM = 'Remove the application? Its container disks are deleted, and only a backup can bring them back.'
RESULT = 'Removal command finished; review any warnings above.'
BRIDGE = 'Private network release attempted:'
BRIDGE = 'Private network of the application released:'
BRIDGE_FAILURE = 'Could not complete private network release:'
SUCCESS = 'The application was removed'
FIREWALL = 'Could not verify removal of the managed host firewall rule.'
@@ -35,6 +35,10 @@ KEYS = (PREVIEW, CONFIRM, RESULT, BRIDGE, BRIDGE_FAILURE, SUCCESS, FIREWALL, POR
def extract(path, name, scope):
if name == 'remove':
# Operational cache cleanup is exercised by the engine tests; keep wording tests inert.
scope.setdefault('compose_jobs', SimpleNamespace(forget_completed=lambda *args: None))
scope.setdefault('public_routes', SimpleNamespace(remove=lambda *args: None))
node = next(n for n in ast.parse(path.read_text()).body if isinstance(n, ast.FunctionDef) and n.name == name)
exec(compile(ast.Module(body=[node], type_ignores=[]), str(path), 'exec'), scope)
return scope[name]
@@ -257,7 +261,7 @@ class RemovalWordings(unittest.TestCase):
self.assertIn(('warn', FIREWALL), events)
self.assertEqual(events[-1], ('ok', RESULT))
def test_ignored_bridge_command_failures_do_not_claim_release(self):
def test_a_failed_bridge_release_warns_instead_of_claiming_it(self):
events = []
scope = {'members_of': lambda root, vmid: (101, {}, [101]),
'instances': SimpleNamespace(read=lambda root, vmid: {'installation_id': 'owned'},
@@ -274,11 +278,12 @@ class RemovalWordings(unittest.TestCase):
'oci_console': SimpleNamespace(remove_log=lambda vmid: None),
'guest_node': lambda vmid: None, 'remove_host_state': lambda vmid: None, 'release_shared_host_files': lambda hookscripts: None, 're': re,
'translate': lambda text: text, 'msg_info': lambda text: None,
'msg_ok': lambda text: events.append(('ok', text)), 'msg_warn': lambda text: None}
'msg_ok': lambda text: events.append(('ok', text)),
'msg_warn': lambda text: events.append(('warn', text))}
extract(REMOVE, 'remove', scope)(Path('/inert'), 101)
self.assertIn(('attempt', 'vmbr9'), events)
self.assertIn(('ok', BRIDGE + ' vmbr9'), events)
self.assertNotIn(('ok', 'Private network of the application released: vmbr9'), events)
self.assertIn(('warn', BRIDGE_FAILURE + ' vmbr9'), events)
self.assertNotIn(('ok', BRIDGE + ' vmbr9'), events)
def test_invalid_managed_firewall_metadata_warns_instead_of_silent_clean(self):
events = []
@@ -133,7 +133,7 @@ class SelectionSetupWording(TestCase):
with self.subTest(app=app):
path = ROOT / f'oci/remote/install_{app}_stack.sh'
lines = path.read_text().splitlines()
start = next(i for i, line in enumerate(lines) if line.strip().startswith('curl -fsS "http://${APPLICATION_LAN_IP}' + port + '/" >/dev/null 2>>"$OCI_LOG"'))
start = next(i for i, line in enumerate(lines) if line.strip().startswith('oci_curl "$APPLICATION_ID" -fsS "http://${APPLICATION_LAN_IP}' + port + '/" >/dev/null 2>>"$OCI_LOG"'))
command = '\n'.join(lines[start:start + (3 if app == 'tandoor' else 2)])
# Actual Bash failure expression, never source the parent installer.
harness = '''set -u
@@ -141,6 +141,7 @@ APPLICATION_LAN_IP=192.0.2.1
APPLICATION_ID=101
OCI_LOG=/dev/null
curl() { return 7; }
oci_curl() { shift; curl "$@"; }
translate() { printf '%s' "$1"; }
print_first_boot_diagnostics() { :; }
die() { printf 'ERROR:%s\\n' "$1"; }
@@ -141,7 +141,11 @@ class UpdateCancellationTests(CancellationFixture):
self.write.assert_not_called()
def test_existing_unattended_review_age_busy_and_missing_codes_are_preserved(self):
with patch.object(reconcile, 'propose', return_value={'details': ['external change']}), redirect_stdout(io.StringIO()):
def declined(unattended, *_args, **_kwargs):
unattended.declined = 'review'
return False
with patch.object(reconcile, 'propose', return_value={'details': ['external change'], 'review': True}), \
patch.object(menu.UnattendedUI, 'review', declined), redirect_stdout(io.StringIO()):
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update', unattended=True), 4)
with patch.object(menu, '_image_too_recent', return_value=True), redirect_stdout(io.StringIO()):
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update', unattended=True, min_image_age_days=3), 5)
@@ -1,9 +1,13 @@
"""The watchdog of an OCI application is reachable from the Monitor and worded in every language."""
import ast
import copy
import json
from pathlib import Path
import re
from unittest import TestCase
from unittest.mock import Mock
from types import SimpleNamespace
from typing import Any
ROOT = Path(__file__).resolve().parents[3]
LOCALES = ('en', 'es', 'de', 'fr', 'it', 'pt', 'sk', 'sv')
@@ -65,9 +69,45 @@ class OciWatchdogInstaller(TestCase):
self.assertLess(install.index('deployment = build_deployment(candidate, wizard, mode)'), asked)
self.assertLess(asked, install.index('approved = wizard.review('))
self.assertLess(install.index('watchdog = bool(deployment.pop("watchdog", False))'), install.index('run_remote_install('))
self.assertIn('set_watchdog(PROJECT_ROOT, sorted(vmids), True)', install)
self._assert_jobs_are_excluded_from_installed_watchdog()
self.assertIn('row(translate("Watchdog"), _yes_no(plan["watchdog"]))', self.CLI)
def _assert_jobs_are_excluded_from_installed_watchdog(self):
node = next(item for item in ast.parse(self.CLI).body
if isinstance(item, ast.FunctionDef) and item.name == 'install_template')
class InertImports(ast.NodeTransformer):
def visit_ImportFrom(self, item):
return None
node = InertImports().visit(node)
deployment = {'services': [
{'name': 'migration', 'deployment': {'execution_mode': 'one-shot'}},
{'name': 'database', 'deployment': {}},
{'name': 'worker', 'deployment': {}}]}
result = {'vmid': 4800, 'stack_vmids': {'web': 4800, 'migration': 4801,
'database': 4802, 'worker': 4803}}
set_watchdog = Mock(return_value=True)
class UI:
def confirm(self, *args, **kwargs): return True
def review(self, *args, **kwargs): return True
def close(self): pass
def install(project, template, plan, target):
self.assertNotIn('watchdog', plan)
return result
scope = {'Any': Any, 'copy': copy, 'BacktrackUI': lambda ui: ui,
'RestartWizard': RuntimeError, 'InstallError': ValueError,
'build_deployment': lambda *args: copy.deepcopy(deployment),
'watchdog_default': lambda *args: False,
'translate': lambda value: value, '_deployment_summary_text': lambda *args: '',
'PROJECT_ROOT': Path('/inert'), 'source_text': lambda value: 'test', 'APP_TITLE': 'test',
'console': SimpleNamespace(**{name: lambda *args: None for name in
('show_logo', 'msg_title', 'msg_error', 'msg_ok', 'msg_warn', 'wait_for_enter')}),
'run_remote_install': install, 'carry_records': lambda *args, **kwargs: None,
'images': SimpleNamespace(offer_removal=lambda *args: (False, None)),
'_print_installation_summary': lambda *args: None, 'set_watchdog': set_watchdog}
exec(compile(ast.fix_missing_locations(ast.Module(body=[node], type_ignores=[])), 'cli.py', 'exec'), scope)
scope['install_template'](UI(), {'catalog_ui': {'title': {'en_US': 'test'}}}, 'test', 'advanced')
set_watchdog.assert_called_once_with(Path('/inert'), [4800, 4802, 4803], True)
def test_both_management_menus_offer_it(self):
self.assertEqual(self.MENU.count("('watchdog', "), 2)
self.assertEqual(self.MENU.count("if action == 'watchdog':"), 2)
+15 -11
View File
@@ -20,8 +20,7 @@ AFTER = 'The recovery did not complete. Review the log and choose "{choice}" aga
REGISTRY_ADVICE = ('Check the image reference and registry access. If the registry is unavailable, '
'try again later. For a private image, sign in from Registry accounts in the OCI menu.')
NEW_KEYS = (
'The Compose file describes several services:',
'Only one service at a time can be installed this way.',
'Containers of the application:',
'No docker run command was given',
'Image selection was cancelled',
'Could not inspect the image in its registry:',
@@ -77,15 +76,19 @@ class OciWordingTests(TestCase):
setattr(module, '_cache', {NEW_KEYS[0]: 'Several services (translated):'})
self.assertEqual(module.translate(NEW_KEYS[0]), 'Several services (translated):')
def test_multiple_services_named_not_images(self):
def test_multiple_services_keep_their_names_and_enter_stack_conversion(self):
# Reusing one image twice still creates two distinct Compose services.
document = {'services': {'alpha': {'image': 'same:latest'},
'beta': {'image': 'same:latest'}}}
converted = object()
stack = Mock(return_value=converted)
template = function(CUSTOM, 'template_from_compose',
_services=lambda doc: doc['services'], _check_mounts=lambda _: None)
with self.assertRaises(ConversionError) as caught:
template(json.dumps({'services': {'alpha': {'image': 'same:latest'},
'beta': {'image': 'same:latest'}}}))
self.assertEqual(str(caught.exception),
'The Compose file describes several services: alpha, beta. '
'Only one service at a time can be installed this way.')
_services=lambda doc: doc['services'], _template_from_stack=stack)
self.assertIs(template(json.dumps(document)), converted)
stack.assert_called_once()
self.assertEqual(stack.call_args.args, (document, None))
self.assertFalse(stack.call_args.kwargs['interpolated'])
self.assertIsNone(stack.call_args.kwargs.get('image_entrypoints'))
def test_empty_paste_is_specific_and_nonempty_unchanged(self):
for title, expected in ((None, 'No Compose file was given'),
@@ -108,7 +111,8 @@ class OciWordingTests(TestCase):
def test_registry_nonzero_and_success_command_unchanged(self):
result = SimpleNamespace(returncode=1, stdout='', stderr='connection timed out')
run = Mock(return_value=result)
report = function(CUSTOM, 'registry_report', subprocess=SimpleNamespace(run=run))
report = function(CUSTOM, 'registry_report', subprocess=SimpleNamespace(run=run),
_registry_lookup=lambda reference: reference)
self.assertEqual(report('example:latest'),
(False, 'Could not inspect the image in its registry: example:latest'))
run.assert_called_once_with(['skopeo', 'inspect', '--raw', 'docker://example:latest'],
@@ -167,7 +167,7 @@ class UpdateCliTests(unittest.TestCase):
with tempfile.TemporaryDirectory(prefix='pbs-update-i18n-') as folder:
body=function('scripts/global/update-pbs.sh','update_pbs')
for old,new in [('/var/log/',folder+'/log/'),('/tmp/proxmenux_screen_capture_',folder+'/tmp/proxmenux_screen_capture_'),('/var/run/',folder+'/run/'),('/boot/',folder+'/boot/')]: body=body.replace(old,new)
script=UPDATE_HARNESS+function('scripts/utils.sh','translate')+body+'\nupdate_pbs\n'
script=UPDATE_HARNESS+function('scripts/utils.sh','translate')+function('scripts/global/common-functions.sh','pmx_confirm_update')+body+'\nupdate_pbs\n'
result=subprocess.run(['bash','-c',script],capture_output=True,text=True,timeout=15,
env={**os.environ,'REPO':str(ROOT),'FIXTURE':folder,'CASE':case,'LANGUAGE':language,'SUITE':suite})
files={str(p.relative_to(folder)):p.read_text() for p in Path(folder).rglob('*') if p.is_file()}
@@ -69,13 +69,22 @@ class TerminalCatalogsTests(unittest.TestCase):
def test_shipped_active_catalogs_keep_literal_contracts(self):
result = subprocess.run(
[sys.executable, '-B', str(SCRIPTS / 'check_terminal_catalogs.py'),
'--root', str(ROOT), '--require-complete'],
'--root', str(ROOT)],
capture_output=True, text=True, timeout=60,
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
for locale in MODULE.DEFAULT_LANGUAGES:
self.assertIn(f'[{locale}] active=', result.stdout)
self.assertNotIn('missing=1', result.stdout)
# New strings are reviewed in Slovak. Other locales keep the runtime
# English fallback until the translation bot supplies their values;
# still validate every translation already present in those locales.
result = subprocess.run(
[sys.executable, '-B', str(SCRIPTS / 'check_terminal_catalogs.py'),
'--root', str(ROOT), '--languages', 'sk', '--require-complete'],
capture_output=True, text=True, timeout=60,
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn('missing=0', result.stdout)
def test_optional_completeness_still_rejects_invalid_present_translations(self):
with tempfile.TemporaryDirectory() as temporary:
@@ -0,0 +1,25 @@
"""A label made only of product and API names is right when it comes back
unchanged, so the translation bot keeps it instead of retrying and dropping it."""
from pathlib import Path
import sys
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import build_translation_cache as cache # noqa: E402
class ProtectedTermsTests(unittest.TestCase):
def test_acceleration_profile_labels_are_kept_unchanged(self):
for label in ("Intel/AMD (Vulkan)", "AMD (ROCm)", "Intel (SYCL)", "NVIDIA (CUDA)",
"Intel/AMD (VA-API)", "Intel (OpenVINO)"):
with self.subTest(label=label):
self.assertTrue(cache.is_fully_protected(label))
self.assertFalse(cache.looks_untranslated(label, label))
def test_a_sentence_coming_back_unchanged_is_still_a_failure(self):
sentence = "The GPU is used through Vulkan when it is available"
self.assertTrue(cache.looks_untranslated(sentence, sentence))
if __name__ == "__main__":
unittest.main()
@@ -33,7 +33,6 @@ jobs:
- uses: actions/checkout@v4
with:
ref: develop
fetch-depth: 0
- uses: actions/setup-python@v5
with:
python-version: '3.11'
+1 -1
View File
@@ -32,7 +32,6 @@ jobs:
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
persist-credentials: false
- uses: actions/setup-python@v5
with:
@@ -45,5 +44,6 @@ jobs:
AUTHOR: ${{ github.event.pull_request.user.login }}
BASE: ${{ github.event.pull_request.base.sha }}
run: |
git fetch --no-tags --depth=1 origin "$BASE"
git show "$BASE:oci/catalog/verification.json" > "$RUNNER_TEMP/base.json" 2>/dev/null || : > "$RUNNER_TEMP/base.json"
python3 .github/scripts/oci_validation.py authors --base "$RUNNER_TEMP/base.json" --author "$AUTHOR"
@@ -1,50 +0,0 @@
name: Apprise Slack and Matrix
on:
pull_request:
paths:
- '.github/workflows/test-apprise-services.yml'
- 'AppImage/scripts/notification_channels.py'
- 'AppImage/components/apprise-service-help.tsx'
- 'AppImage/components/notification-settings.tsx'
- 'tests/test_apprise_services.py'
- 'tests/test_apprise_service_help.cjs'
push:
branches: [main, develop]
paths:
- 'AppImage/scripts/notification_channels.py'
- 'AppImage/components/apprise-service-help.tsx'
- 'tests/test_apprise_services.py'
- 'tests/test_apprise_service_help.cjs'
workflow_dispatch:
permissions:
contents: read
jobs:
apprise-services:
runs-on: ubuntu-latest
timeout-minutes: 10
strategy:
matrix:
apprise: ['1.7.0', '2.0.1']
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'apprise==${{ matrix.apprise }}'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test real provider plugins with mocked HTTP
run: python -B -m unittest discover -s tests -p test_apprise_services.py -v
- name: Render provider help and locale fallback
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_apprise_service_help.cjs
@@ -1,50 +0,0 @@
name: Backup storage retention
on:
pull_request:
paths:
- '.github/workflows/test-backup-storage-retention.yml'
- 'AppImage/scripts/flask_vm_backup_routes.py'
- 'AppImage/components/backup-storage-retention.tsx'
- 'AppImage/components/backup-tab.tsx'
- 'AppImage/components/vm-backup-jobs.tsx'
- 'AppImage/lib/backup-retention.ts'
- 'tests/test_vm_backup_jobs_api.py'
- 'tests/test_backup_storage_retention.cjs'
push:
branches: [main, develop]
paths:
- 'AppImage/scripts/flask_vm_backup_routes.py'
- 'AppImage/components/backup-storage-retention.tsx'
- 'AppImage/lib/backup-retention.ts'
- 'tests/test_vm_backup_jobs_api.py'
- 'tests/test_backup_storage_retention.cjs'
workflow_dispatch:
permissions:
contents: read
jobs:
storage-retention:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'Flask==3.1.2'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test native job and storage configuration APIs
run: python -B -m unittest discover -s tests -p test_vm_backup_jobs_api.py -v
- name: Test shared retention properties
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_backup_storage_retention.cjs
+384
View File
@@ -0,0 +1,384 @@
name: Feature tests
on:
pull_request:
paths:
- '.github/workflows/test-features.yml'
- 'AppImage/scripts/notification_manager.py'
- 'AppImage/scripts/notification_templates.py'
- 'AppImage/scripts/ai_context_enrichment.py'
- 'AppImage/scripts/ai_providers/base.py'
- 'AppImage/scripts/tests/test_backup_ai_diagnostics.py'
- 'AppImage/scripts/notification_channels.py'
- 'AppImage/components/apprise-service-help.tsx'
- 'AppImage/components/notification-settings.tsx'
- 'tests/test_apprise_services.py'
- 'tests/test_apprise_service_help.cjs'
- 'AppImage/scripts/flask_vm_backup_routes.py'
- 'AppImage/components/backup-storage-retention.tsx'
- 'AppImage/components/backup-tab.tsx'
- 'AppImage/components/vm-backup-jobs.tsx'
- 'AppImage/lib/backup-retention.ts'
- 'tests/test_vm_backup_jobs_api.py'
- 'tests/test_backup_storage_retention.cjs'
- 'tests/test_figurine_installation.py'
- 'scripts/post_install/customizable_post_install.sh'
- 'AppImage/scripts/post_install_versions.py'
- 'AppImage/scripts/flask_proxmenux_routes.py'
- 'AppImage/scripts/guest_backup_archives.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/jwt_middleware.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/components/guest-backup-list.tsx'
- 'AppImage/components/virtual-machines.tsx'
- 'AppImage/lib/guest-backup-archives.ts'
- 'AppImage/messages/**'
- 'tests/test_guest_backup_archives.*'
- 'AppImage/scripts/lvm_device_scope.py'
- 'AppImage/scripts/disk_temperature_history.py'
- 'AppImage/scripts/health_monitor.py'
- 'AppImage/scripts/flask_notification_routes.py'
- 'AppImage/scripts/proxmox_storage_monitor.py'
- 'AppImage/scripts/tests/test_host_backup_node.py'
- 'AppImage/scripts/tests/test_proxmox_node_name.py'
- 'AppImage/scripts/tests/test_pve_notifications_cfg_writes.py'
- 'tests/test_lvm_disk_read_policy.py'
- 'oci/**'
- 'AppImage/scripts/oci_menu_bridge.py'
- 'AppImage/scripts/lxc_apps.py'
- 'AppImage/scripts/tests/test_oci_parallel_checks.py'
- 'AppImage/scripts/*pbs*.py'
- 'AppImage/scripts/tests/test_pbs.py'
- 'AppImage/scripts/tests/test_pbs_i18n.py'
- 'AppImage/scripts/auth_manager.py'
- 'AppImage/components/pbs-*.tsx'
- 'AppImage/components/proxmox-dashboard.tsx'
- 'AppImage/tests/pbs-navigation.cjs'
- 'scripts/global/proxmox_platform.py'
- 'scripts/global/platform-functions.sh'
- 'scripts/global/repository-functions.sh'
- 'scripts/menus/pbs_menu.sh'
- 'scripts/utilities/pbs_update.sh'
- 'scripts/global/update-pbs.sh'
- 'scripts/menus/menu_post_install.sh'
- 'scripts/post_install/*.sh'
push:
branches: [main, develop]
paths:
- '.github/workflows/test-features.yml'
- 'AppImage/scripts/notification_manager.py'
- 'AppImage/scripts/notification_templates.py'
- 'AppImage/scripts/ai_context_enrichment.py'
- 'AppImage/scripts/ai_providers/base.py'
- 'AppImage/scripts/tests/test_backup_ai_diagnostics.py'
- 'AppImage/scripts/notification_channels.py'
- 'AppImage/components/apprise-service-help.tsx'
- 'tests/test_apprise_services.py'
- 'tests/test_apprise_service_help.cjs'
- 'AppImage/scripts/flask_vm_backup_routes.py'
- 'AppImage/components/backup-storage-retention.tsx'
- 'AppImage/lib/backup-retention.ts'
- 'tests/test_vm_backup_jobs_api.py'
- 'tests/test_backup_storage_retention.cjs'
- 'tests/test_figurine_installation.py'
- 'scripts/post_install/customizable_post_install.sh'
- 'AppImage/scripts/post_install_versions.py'
- 'AppImage/scripts/flask_proxmenux_routes.py'
- 'AppImage/scripts/guest_backup_archives.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/jwt_middleware.py'
- 'AppImage/components/guest-backup-list.tsx'
- 'AppImage/components/virtual-machines.tsx'
- 'AppImage/lib/guest-backup-archives.ts'
- 'tests/test_guest_backup_archives.*'
- 'AppImage/scripts/lvm_device_scope.py'
- 'AppImage/scripts/disk_temperature_history.py'
- 'AppImage/scripts/health_monitor.py'
- 'AppImage/scripts/flask_notification_routes.py'
- 'AppImage/scripts/proxmox_storage_monitor.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/scripts/tests/test_host_backup_node.py'
- 'AppImage/scripts/tests/test_proxmox_node_name.py'
- 'AppImage/scripts/tests/test_pve_notifications_cfg_writes.py'
- 'tests/test_lvm_disk_read_policy.py'
- 'oci/**'
- 'AppImage/scripts/oci_menu_bridge.py'
- 'AppImage/scripts/lxc_apps.py'
- 'AppImage/scripts/tests/test_oci_parallel_checks.py'
- 'AppImage/scripts/*pbs*.py'
- 'scripts/global/proxmox_platform.py'
workflow_dispatch:
permissions:
contents: read
jobs:
apprise-services:
runs-on: ubuntu-latest
timeout-minutes: 10
strategy:
matrix:
apprise: ['1.7.0', '2.0.1']
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'apprise==${{ matrix.apprise }}'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test real provider plugins with mocked HTTP
run: python -B -m unittest discover -s tests -p test_apprise_services.py -v
- name: Render provider help and locale fallback
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_apprise_service_help.cjs
storage-retention:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'Flask==3.1.2'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test native job and storage configuration APIs
run: python -B -m unittest discover -s tests -p test_vm_backup_jobs_api.py -v
- name: Test shared retention properties
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_backup_storage_retention.cjs
installer-and-release:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Check shell syntax and isolated installer regressions
run: |
bash -n scripts/post_install/customizable_post_install.sh
python3 -B -m unittest discover -s tests -p test_figurine_installation.py -v
- name: Install ARM user-mode emulation for release smoke tests
run: |
sudo apt-get update -qq
sudo apt-get install -y --no-install-recommends qemu-user
- name: Verify and run the pinned AMD64, ARM64 and ARMv7 release binaries
env:
FIGURINE_VERSION: '2.1.1'
shell: bash
run: |
set -euo pipefail
scratch="$(mktemp -d)"
trap 'rm -rf "$scratch"' EXIT
while read -r arch sha; do
archive="figurine_linux_${arch}_v${FIGURINE_VERSION}.tar.gz"
target="$scratch/$arch"
mkdir -p "$target"
curl --fail --location --retry 3 --output "$target/$archive" \
"https://github.com/arsham/figurine/releases/download/v${FIGURINE_VERSION}/$archive"
printf '%s %s\n' "$sha" "$target/$archive" | sha256sum --check
tar -xf "$target/$archive" -C "$target"
runner=()
case "$arch" in
arm64) runner=(qemu-aarch64) ;;
arm) runner=(qemu-arm) ;;
esac
"${runner[@]}" "$target/deploy/figurine" --version | tee "$target/version"
grep -F "figurine version v$FIGURINE_VERSION" "$target/version"
"${runner[@]}" "$target/deploy/figurine" -f "3d.flf" "ProxMenux-test" > "$target/banner"
test -s "$target/banner"
printf 'PASS: %s version and login banner\n' "$arch"
done <<'ASSETS'
amd64 f2b1c076942c94009d133df0f304a234a0994bf590b40f2927668324d578306a
arm64 611550ddb31a75437be3f8587bcdd029d0d890f4d820b7f494a2c6615fecba10
arm 429e66bded486d49f730c4a6559e728cd1e9a163a0f56a85867e2cbec010171d
ASSETS
guest-backups:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'Flask==3.1.2'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test native deletion, protection, authorization and cache invalidation
run: python -B -m unittest discover -s tests -p test_guest_backup_archives.py -v
- name: Test frontend deletion and completion handling
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_guest_backup_archives.cjs
disk-policy:
runs-on: ubuntu-latest
timeout-minutes: 5
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Test automatic LVM readers with mocked inventory and policy
run: python -I -B -m unittest discover -s tests -p test_lvm_disk_read_policy.py -v
- name: Test backup node ownership and former jobs
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_host_backup_node.py -v
- name: Test local node lookup without Proxmox API commands
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_proxmox_node_name.py -v
- name: Test notification configuration writes against a cluster filesystem fixture
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_pve_notifications_cfg_writes.py -v
- name: Test AI diagnostic explanations and backup inventory integrity
run: |
python -I -B -m unittest discover -s AppImage/scripts/tests -p test_backup_ai_diagnostics.py -v
python -I -B -m unittest discover -s AppImage/scripts/tests -p test_vzdump_ai_integrity.py -v
registry-accounts:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.13'
- name: Install OCI test dependencies
run: python3 -m pip install -r oci/requirements.txt
- name: Run watchdog identity and lifecycle regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p test_watchdog.py -v
- name: Run OCI image reuse and resource edit regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_image_reuse.py -v
python3 -m unittest discover -s oci/tests -p test_native_modify.py -v
python3 -m unittest discover -s oci/tests -p test_modify_review.py -v
python3 -m unittest discover -s oci/tests -p test_modify_review_transaction.py -v
python3 -m unittest discover -s oci/tests -p test_stage_release.py -v
- name: Run carried record, recovery and configuration cache regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_carried*.py' -v
python3 -m unittest discover -s oci/tests -p test_restore_recovery.py -v
- name: Run read-only volume preparation regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
test -x /bin/bash
test -x /usr/bin/jq
python3 -m unittest discover -s oci/tests -p test_readonly_volume_preparation.py -v
- name: Run offline registry account tests
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p 'test_registry_*.py' -v
- name: Run OCI menu update checks and selection regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_update*.py' -v
python3 -m unittest discover -s AppImage/scripts/tests -p 'test_oci*.py' -v
python3 -m unittest discover -s oci/tests -p test_recreate_registry_digest.py -v
- name: Run OCI Modify latency regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_latency.py -v
python3 -m unittest discover -s oci/tests -p test_menu_performance.py -v
python3 -m unittest discover -s oci/tests -p test_dialog_escape.py -v
python3 -m unittest discover -s oci/tests -p test_work_backup.py -v
python3 -m unittest discover -s oci/tests -p test_recreation.py -v
- name: Run OCI modification menu regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_menu.py -v
python3 -m unittest discover -s oci/tests -p test_manage_sync_notice.py -v
python3 -m unittest discover -s oci/tests -p test_recreation.py -v
python3 -m unittest discover -s oci/tests -p test_shared_directories.py -v
python3 -m unittest discover -s oci/tests -p 'test_wizard*.py' -v
- name: Run external OCI settings regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_external_tags.py' -v
python3 -m unittest discover -s oci/tests -p 'test_instance_reconcile.py' -v
python3 -m unittest discover -s oci/tests -p 'test_guessed_endpoint_healthcheck.py' -v
python3 -m unittest discover -s oci/tests -p 'test_runtime_home_*.py' -v
- name: Run managed volume startup regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p 'test_managed_volume_*.py' -v
- name: Run catalog web-link regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p test_catalog_web_links.py -v
- name: Run full OCI engine test suite
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -v
pbs:
runs-on: ubuntu-latest
timeout-minutes: 10
strategy:
fail-fast: false
matrix:
python: ['3.11', '3.13']
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python }}
- name: Install test dependencies
run: python -m pip install Flask PyJWT cryptography pyotp segno
- name: Platform, real TLS, APIs, health and AppImage packaging
run: |
python -B -m unittest discover -s AppImage/scripts/tests -p test_pbs.py -v
python -B -m unittest discover -s AppImage/scripts/tests -p test_pbs_i18n.py -v
python -B -m unittest discover -s AppImage/scripts/tests -p test_auth_manager_ssl_reload.py -v
for script in scripts/global/platform-functions.sh scripts/global/repository-functions.sh scripts/menus/pbs_menu.sh scripts/utilities/pbs_update.sh scripts/menus/main_menu.sh scripts/post_install/auto_post_install.sh scripts/post_install/customizable_post_install.sh scripts/post_install/uninstall-tools.sh scripts/global/update-pbs.sh scripts/menus/menu_post_install.sh; do
bash -n "$script"
done
- uses: actions/setup-node@v6
if: matrix.python == '3.11'
with:
node-version: '22.14.0'
- name: Dashboard platform routing
if: matrix.python == '3.11'
run: |
npm ci --prefix .github/ci/offline-node --legacy-peer-deps --ignore-scripts --no-audit --no-fund
NODE_PATH="$PWD/.github/ci/offline-node/node_modules" node AppImage/tests/pbs-navigation.cjs
-73
View File
@@ -1,73 +0,0 @@
name: Figurine installation
on:
pull_request:
paths:
- '.github/workflows/test-figurine.yml'
- 'tests/test_figurine_installation.py'
- 'scripts/post_install/customizable_post_install.sh'
- 'AppImage/scripts/post_install_versions.py'
- 'AppImage/scripts/flask_proxmenux_routes.py'
push:
branches: [main, develop]
paths:
- '.github/workflows/test-figurine.yml'
- 'tests/test_figurine_installation.py'
- 'scripts/post_install/customizable_post_install.sh'
- 'AppImage/scripts/post_install_versions.py'
- 'AppImage/scripts/flask_proxmenux_routes.py'
workflow_dispatch:
permissions:
contents: read
jobs:
installer-and-release:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Check shell syntax and isolated installer regressions
run: |
bash -n scripts/post_install/customizable_post_install.sh
python3 -B -m unittest discover -s tests -p test_figurine_installation.py -v
- name: Install ARM user-mode emulation for release smoke tests
run: |
sudo apt-get update -qq
sudo apt-get install -y --no-install-recommends qemu-user
- name: Verify and run the pinned AMD64, ARM64 and ARMv7 release binaries
env:
FIGURINE_VERSION: '2.1.1'
shell: bash
run: |
set -euo pipefail
scratch="$(mktemp -d)"
trap 'rm -rf "$scratch"' EXIT
while read -r arch sha; do
archive="figurine_linux_${arch}_v${FIGURINE_VERSION}.tar.gz"
target="$scratch/$arch"
mkdir -p "$target"
curl --fail --location --retry 3 --output "$target/$archive" \
"https://github.com/arsham/figurine/releases/download/v${FIGURINE_VERSION}/$archive"
printf '%s %s\n' "$sha" "$target/$archive" | sha256sum --check
tar -xf "$target/$archive" -C "$target"
runner=()
case "$arch" in
arm64) runner=(qemu-aarch64) ;;
arm) runner=(qemu-arm) ;;
esac
"${runner[@]}" "$target/deploy/figurine" --version | tee "$target/version"
grep -F "figurine version v$FIGURINE_VERSION" "$target/version"
"${runner[@]}" "$target/deploy/figurine" -f "3d.flf" "ProxMenux-test" > "$target/banner"
test -s "$target/banner"
printf 'PASS: %s version and login banner\n' "$arch"
done <<'ASSETS'
amd64 f2b1c076942c94009d133df0f304a234a0994bf590b40f2927668324d578306a
arm64 611550ddb31a75437be3f8587bcdd029d0d890f4d820b7f494a2c6615fecba10
arm 429e66bded486d49f730c4a6559e728cd1e9a163a0f56a85867e2cbec010171d
ASSETS
-54
View File
@@ -1,54 +0,0 @@
name: Guest backup archives
on:
pull_request:
paths:
- '.github/workflows/test-guest-backups.yml'
- 'AppImage/scripts/guest_backup_archives.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/jwt_middleware.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/components/guest-backup-list.tsx'
- 'AppImage/components/virtual-machines.tsx'
- 'AppImage/lib/guest-backup-archives.ts'
- 'AppImage/messages/**'
- 'tests/test_guest_backup_archives.*'
push:
branches: [main, develop]
paths:
- 'AppImage/scripts/guest_backup_archives.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/jwt_middleware.py'
- 'AppImage/components/guest-backup-list.tsx'
- 'AppImage/components/virtual-machines.tsx'
- 'AppImage/lib/guest-backup-archives.ts'
- 'tests/test_guest_backup_archives.*'
workflow_dispatch:
permissions:
contents: read
jobs:
guest-backups:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- uses: actions/setup-node@v6
with:
node-version: '22.14.0'
- name: Install fixture dependencies
run: |
python -m pip install 'Flask==3.1.2'
npm ci --prefix .github/ci/offline-node --ignore-scripts --no-audit --no-fund
- name: Test native deletion, protection, authorization and cache invalidation
run: python -B -m unittest discover -s tests -p test_guest_backup_archives.py -v
- name: Test frontend deletion and completion handling
env:
NODE_PATH: ${{ github.workspace }}/.github/ci/offline-node/node_modules
run: node tests/test_guest_backup_archives.cjs
@@ -1,55 +0,0 @@
name: Disk policy and host monitor regressions
on:
pull_request:
paths:
- '.github/workflows/test-lvm-disk-policy.yml'
- 'AppImage/scripts/lvm_device_scope.py'
- 'AppImage/scripts/disk_temperature_history.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/health_monitor.py'
- 'AppImage/scripts/flask_notification_routes.py'
- 'AppImage/scripts/proxmox_storage_monitor.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/scripts/tests/test_host_backup_node.py'
- 'AppImage/scripts/tests/test_proxmox_node_name.py'
- 'AppImage/scripts/tests/test_pve_notifications_cfg_writes.py'
- 'tests/test_lvm_disk_read_policy.py'
push:
branches: [main, develop]
paths:
- 'AppImage/scripts/lvm_device_scope.py'
- 'AppImage/scripts/disk_temperature_history.py'
- 'AppImage/scripts/flask_server.py'
- 'AppImage/scripts/health_monitor.py'
- 'AppImage/scripts/flask_notification_routes.py'
- 'AppImage/scripts/proxmox_storage_monitor.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/scripts/tests/test_host_backup_node.py'
- 'AppImage/scripts/tests/test_proxmox_node_name.py'
- 'AppImage/scripts/tests/test_pve_notifications_cfg_writes.py'
- 'tests/test_lvm_disk_read_policy.py'
workflow_dispatch:
permissions:
contents: read
jobs:
disk-policy:
runs-on: ubuntu-latest
timeout-minutes: 5
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Test automatic LVM readers with mocked inventory and policy
run: python -I -B -m unittest discover -s tests -p test_lvm_disk_read_policy.py -v
- name: Test backup node ownership and former jobs
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_host_backup_node.py -v
- name: Test local node lookup without Proxmox API commands
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_proxmox_node_name.py -v
- name: Test notification configuration writes against a cluster filesystem fixture
run: python -I -B -m unittest discover -s AppImage/scripts/tests -p test_pve_notifications_cfg_writes.py -v
-109
View File
@@ -1,109 +0,0 @@
name: Offline OCI registry tests
on:
pull_request:
paths:
- 'oci/**'
- 'AppImage/scripts/oci_menu_bridge.py'
- 'AppImage/scripts/lxc_apps.py'
- 'AppImage/scripts/tests/test_oci_parallel_checks.py'
- 'AppImage/scripts/flask_server.py'
- '.github/workflows/test-oci-registries.yml'
push:
branches: [main, develop]
paths:
- 'oci/**'
- 'AppImage/scripts/oci_menu_bridge.py'
- 'AppImage/scripts/lxc_apps.py'
- 'AppImage/scripts/tests/test_oci_parallel_checks.py'
- 'AppImage/scripts/flask_server.py'
- '.github/workflows/test-oci-registries.yml'
workflow_dispatch:
permissions:
contents: read
jobs:
registry-accounts:
runs-on: ubuntu-latest
timeout-minutes: 5
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: '3.13'
- name: Install OCI test dependencies
run: python3 -m pip install -r oci/requirements.txt
- name: Run watchdog identity and lifecycle regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p test_watchdog.py -v
- name: Run OCI image reuse and resource edit regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_image_reuse.py -v
python3 -m unittest discover -s oci/tests -p test_native_modify.py -v
python3 -m unittest discover -s oci/tests -p test_modify_review.py -v
python3 -m unittest discover -s oci/tests -p test_modify_review_transaction.py -v
python3 -m unittest discover -s oci/tests -p test_stage_release.py -v
- name: Run carried record, recovery and configuration cache regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_carried*.py' -v
python3 -m unittest discover -s oci/tests -p test_restore_recovery.py -v
- name: Run read-only volume preparation regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
test -x /bin/bash
test -x /usr/bin/jq
python3 -m unittest discover -s oci/tests -p test_readonly_volume_preparation.py -v
- name: Run offline registry account tests
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p 'test_registry_*.py' -v
- name: Run OCI menu update checks and selection regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_update*.py' -v
python3 -m unittest discover -s AppImage/scripts/tests -p 'test_oci*.py' -v
python3 -m unittest discover -s oci/tests -p test_recreate_registry_digest.py -v
- name: Run OCI Modify latency regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_latency.py -v
python3 -m unittest discover -s oci/tests -p test_menu_performance.py -v
python3 -m unittest discover -s oci/tests -p test_dialog_escape.py -v
python3 -m unittest discover -s oci/tests -p test_work_backup.py -v
python3 -m unittest discover -s oci/tests -p test_recreation.py -v
- name: Run OCI modification menu regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p test_modify_menu.py -v
python3 -m unittest discover -s oci/tests -p test_manage_sync_notice.py -v
python3 -m unittest discover -s oci/tests -p test_recreation.py -v
python3 -m unittest discover -s oci/tests -p test_shared_directories.py -v
python3 -m unittest discover -s oci/tests -p 'test_wizard*.py' -v
- name: Run external OCI settings regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python3 -m unittest discover -s oci/tests -p 'test_external_tags.py' -v
python3 -m unittest discover -s oci/tests -p 'test_instance_reconcile.py' -v
python3 -m unittest discover -s oci/tests -p 'test_guessed_endpoint_healthcheck.py' -v
python3 -m unittest discover -s oci/tests -p 'test_runtime_home_*.py' -v
- name: Run managed volume startup regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p 'test_managed_volume_*.py' -v
- name: Run catalog web-link regressions
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python3 -m unittest discover -s oci/tests -p test_catalog_web_links.py -v
+1 -1
View File
@@ -69,7 +69,7 @@ jobs:
with:
python-version: '3.11'
- name: Install native discovery route fixture dependency
run: python3 -m pip install 'Flask>=3,<4' 'Pillow==12.3.0' 'defusedxml==0.7.1'
run: python3 -m pip install 'Flask>=3,<4' 'Pillow==12.3.0' 'defusedxml==0.7.1' 'PyJWT>=2,<3' 'gevent>=25,<27' 'flask-sock==0.7.0' requests
- name: Run qualified Python files
env:
PYTHONDONTWRITEBYTECODE: '1'
-68
View File
@@ -1,68 +0,0 @@
name: PBS support
on:
pull_request:
paths:
- '.github/workflows/test-pbs.yml'
- 'AppImage/scripts/*pbs*.py'
- 'AppImage/scripts/tests/test_pbs.py'
- 'AppImage/scripts/tests/test_pbs_i18n.py'
- 'AppImage/scripts/health_monitor.py'
- 'AppImage/scripts/auth_manager.py'
- 'AppImage/scripts/build_appimage.sh'
- 'AppImage/components/pbs-*.tsx'
- 'AppImage/components/proxmox-dashboard.tsx'
- 'AppImage/tests/pbs-navigation.cjs'
- 'scripts/global/proxmox_platform.py'
- 'scripts/global/platform-functions.sh'
- 'scripts/global/repository-functions.sh'
- 'scripts/menus/pbs_menu.sh'
- 'scripts/utilities/pbs_update.sh'
- 'scripts/global/update-pbs.sh'
- 'scripts/menus/menu_post_install.sh'
- 'scripts/post_install/*.sh'
push:
branches: [main, develop]
paths:
- 'AppImage/scripts/*pbs*.py'
- 'scripts/global/proxmox_platform.py'
- '.github/workflows/test-pbs.yml'
workflow_dispatch:
permissions:
contents: read
jobs:
pbs:
runs-on: ubuntu-latest
timeout-minutes: 10
strategy:
fail-fast: false
matrix:
python: ['3.11', '3.13']
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python }}
- name: Install test dependencies
run: python -m pip install Flask PyJWT cryptography pyotp segno
- name: Platform, real TLS, APIs, health and AppImage packaging
run: |
python -B -m unittest discover -s AppImage/scripts/tests -p test_pbs.py -v
python -B -m unittest discover -s AppImage/scripts/tests -p test_pbs_i18n.py -v
python -B -m unittest discover -s AppImage/scripts/tests -p test_auth_manager_ssl_reload.py -v
for script in scripts/global/platform-functions.sh scripts/global/repository-functions.sh scripts/menus/pbs_menu.sh scripts/utilities/pbs_update.sh scripts/menus/main_menu.sh scripts/post_install/auto_post_install.sh scripts/post_install/customizable_post_install.sh scripts/post_install/uninstall-tools.sh scripts/global/update-pbs.sh scripts/menus/menu_post_install.sh; do
bash -n "$script"
done
- uses: actions/setup-node@v6
if: matrix.python == '3.11'
with:
node-version: '22.14.0'
- name: Dashboard platform routing
if: matrix.python == '3.11'
run: |
npm ci --prefix .github/ci/offline-node --legacy-peer-deps --ignore-scripts --no-audit --no-fund
NODE_PATH="$PWD/.github/ci/offline-node/node_modules" node AppImage/tests/pbs-navigation.cjs
@@ -14,6 +14,7 @@ on:
branches: [main]
paths:
- ".github/scripts/generate_app_tracking_catalog.py"
- ".github/scripts/tests/test_app_tracking_sources.py"
- ".github/workflows/update-app-tracking-hints.yml"
- "json/runtime_verified_overrides.json"
@@ -37,6 +38,9 @@ jobs:
with:
python-version: "3.11"
- name: Test Helper-Scripts source layouts
run: python -B -m unittest discover -s .github/scripts/tests -p test_app_tracking_sources.py -v
- name: ⚙️ Generate app_tracking_hints.generated.json (intermediate)
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}