mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-09-29 18:16:43 +00:00
ProxMenux 1.2.6.2-beta: OCI containers in the Monitor, docs and fixes
OCI manager Apps - App tab: containers installed from an OCI image are identified from their installation record; the application and image versions are shown and an update is detected by image digest; repository link; Refresh data. - Updates tab for OCI containers: Update and Recreate run the same flow as the OCI menu in the Monitor terminal; the pre-update backup can be kept in a backup storage; scheduled image updates with an optional minimum age. - Logs tab: console output of the application, kept on the host (lxc.console.logfile + logrotate) and followed live. - The Proxmox console opens a shell (cmode: shell) when the image has one. - A damaged image download is fetched again before failing. - Multi-container applications open at their LAN address; volume mount points on block storage report their usage. Monitor - Proxmox notifications are delivered to a loopback-only HTTP listener when HTTPS is enabled, so they no longer fail certificate verification. - Log persistence counts recurring patterns only; an ended burst is not reported as persistent and its warning clears on its own (#386). - Proxmox notification config backups are deduplicated and capped at three. - The update icon on the Apps page opens the container on its Updates tab. - Version 1.2.6.2-beta and its release notes in every Monitor language. Docs - OCI manager Apps and Audit & Report rebuilt as per-page message files, with a new page for OCI containers in the Monitor. - Seven pages fixed where rich-text tags were missing from t.rich. Translations - Spanish fixes across the OCI engine, the Monitor and the TUI menus. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
MacRimi
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
386d33df6e
commit
4437a671d2
@@ -0,0 +1,217 @@
|
||||
import importlib.util
|
||||
import json
|
||||
import shutil
|
||||
import socket
|
||||
import ssl
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
|
||||
MODULE_PATH = Path(__file__).resolve().parents[1] / "auth_manager.py"
|
||||
SPEC = importlib.util.spec_from_file_location("auth_manager_ssl_under_test", MODULE_PATH)
|
||||
auth_manager = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(auth_manager)
|
||||
|
||||
|
||||
class _FakeSslSocket:
|
||||
def __init__(self, context):
|
||||
self.context = context
|
||||
|
||||
|
||||
class ProxmoxCertificateHotReloadTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if shutil.which("openssl") is None:
|
||||
raise unittest.SkipTest("openssl is required for TLS fixture generation")
|
||||
cls.fixture_dir = tempfile.TemporaryDirectory()
|
||||
fixture_path = Path(cls.fixture_dir.name)
|
||||
cls.pairs = []
|
||||
for name in ("original", "renewed"):
|
||||
cert_path = fixture_path / f"{name}.pem"
|
||||
key_path = fixture_path / f"{name}.key"
|
||||
subprocess.run(
|
||||
[
|
||||
"openssl", "req", "-x509", "-newkey", "rsa:2048",
|
||||
"-nodes", "-days", "1", "-subj", f"/CN={name}.test",
|
||||
"-keyout", str(key_path), "-out", str(cert_path),
|
||||
],
|
||||
check=True,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
cls.pairs.append((cert_path, key_path))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.fixture_dir.cleanup()
|
||||
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp_dir.cleanup)
|
||||
temp_path = Path(self.temp_dir.name)
|
||||
self.active_cert = temp_path / "pveproxy-ssl.pem"
|
||||
self.active_key = temp_path / "pveproxy-ssl.key"
|
||||
self.ssl_config = temp_path / "ssl_config.json"
|
||||
self._install_pair(0)
|
||||
self._write_config("proxmox")
|
||||
|
||||
self.patch = mock.patch.multiple(
|
||||
auth_manager,
|
||||
SSL_CONFIG_FILE=self.ssl_config,
|
||||
PROXMOX_CUSTOM_CERT_PATH=str(self.active_cert),
|
||||
PROXMOX_CUSTOM_KEY_PATH=str(self.active_key),
|
||||
PROXMOX_CERT_PATH=str(temp_path / "missing-pve-ssl.pem"),
|
||||
PROXMOX_KEY_PATH=str(temp_path / "missing-pve-ssl.key"),
|
||||
)
|
||||
self.patch.start()
|
||||
self.addCleanup(self.patch.stop)
|
||||
self.addCleanup(self._reset_runtime)
|
||||
|
||||
def _reset_runtime(self):
|
||||
with auth_manager._SSL_RUNTIME_LOCK:
|
||||
auth_manager._SSL_RUNTIME_CONTEXT = None
|
||||
auth_manager._SSL_RUNTIME_FINGERPRINT = ""
|
||||
auth_manager._SSL_RUNTIME_CERT_PATH = ""
|
||||
auth_manager._SSL_RUNTIME_KEY_PATH = ""
|
||||
auth_manager._SSL_RUNTIME_SOURCE = "none"
|
||||
auth_manager._SSL_RUNTIME_LAST_REFRESH_ERROR = ""
|
||||
|
||||
def _install_pair(self, index):
|
||||
cert_path, key_path = self.pairs[index]
|
||||
shutil.copyfile(cert_path, self.active_cert)
|
||||
shutil.copyfile(key_path, self.active_key)
|
||||
|
||||
def _write_config(self, source):
|
||||
self.ssl_config.write_text(json.dumps({
|
||||
"enabled": True,
|
||||
"cert_path": str(self.active_cert),
|
||||
"key_path": str(self.active_key),
|
||||
"source": source,
|
||||
}))
|
||||
|
||||
def _create_context(self):
|
||||
return auth_manager.create_reloadable_ssl_context(
|
||||
str(self.active_cert), str(self.active_key)
|
||||
)
|
||||
|
||||
def test_unchanged_pair_does_not_rebuild_the_context(self):
|
||||
context = self._create_context()
|
||||
ssl_socket = _FakeSslSocket(context)
|
||||
|
||||
with mock.patch.object(
|
||||
auth_manager,
|
||||
"reload_server_ssl_context",
|
||||
wraps=auth_manager.reload_server_ssl_context,
|
||||
) as reload_mock:
|
||||
context.sni_callback(ssl_socket, "proxmenux.test", context)
|
||||
|
||||
reload_mock.assert_not_called()
|
||||
self.assertIs(ssl_socket.context, context)
|
||||
|
||||
def test_valid_renewed_pair_is_activated_during_the_handshake(self):
|
||||
context = self._create_context()
|
||||
previous_fingerprint = auth_manager._SSL_RUNTIME_FINGERPRINT
|
||||
self._install_pair(1)
|
||||
ssl_socket = _FakeSslSocket(context)
|
||||
|
||||
context.sni_callback(ssl_socket, "proxmenux.test", context)
|
||||
|
||||
self.assertNotEqual(previous_fingerprint, auth_manager._SSL_RUNTIME_FINGERPRINT)
|
||||
self.assertIs(ssl_socket.context, auth_manager._SSL_RUNTIME_CONTEXT)
|
||||
self.assertIsNot(ssl_socket.context, context)
|
||||
self.assertEqual(auth_manager._SSL_RUNTIME_LAST_REFRESH_ERROR, "")
|
||||
|
||||
def test_mismatched_pair_keeps_the_previous_context(self):
|
||||
context = self._create_context()
|
||||
previous_fingerprint = auth_manager._SSL_RUNTIME_FINGERPRINT
|
||||
shutil.copyfile(self.pairs[1][0], self.active_cert)
|
||||
ssl_socket = _FakeSslSocket(context)
|
||||
|
||||
context.sni_callback(ssl_socket, "proxmenux.test", context)
|
||||
|
||||
self.assertEqual(previous_fingerprint, auth_manager._SSL_RUNTIME_FINGERPRINT)
|
||||
self.assertIs(auth_manager._SSL_RUNTIME_CONTEXT, context)
|
||||
self.assertIs(ssl_socket.context, context)
|
||||
self.assertTrue(auth_manager._SSL_RUNTIME_LAST_REFRESH_ERROR)
|
||||
|
||||
def test_custom_certificate_source_is_not_examined_automatically(self):
|
||||
self._write_config("custom")
|
||||
context = self._create_context()
|
||||
previous_fingerprint = auth_manager._SSL_RUNTIME_FINGERPRINT
|
||||
self._install_pair(1)
|
||||
ssl_socket = _FakeSslSocket(context)
|
||||
|
||||
context.sni_callback(ssl_socket, "proxmenux.test", context)
|
||||
|
||||
self.assertEqual(previous_fingerprint, auth_manager._SSL_RUNTIME_FINGERPRINT)
|
||||
self.assertIs(ssl_socket.context, context)
|
||||
|
||||
def test_first_real_tls_connection_receives_the_renewed_certificate(self):
|
||||
server_context = self._create_context()
|
||||
self._install_pair(1)
|
||||
server_socket, client_socket = socket.socketpair()
|
||||
server_socket.settimeout(5)
|
||||
client_socket.settimeout(5)
|
||||
server_error = []
|
||||
|
||||
def serve_once():
|
||||
try:
|
||||
with server_context.wrap_socket(server_socket, server_side=True) as tls_socket:
|
||||
tls_socket.recv(1)
|
||||
except Exception as error: # pragma: no cover - asserted below
|
||||
server_error.append(error)
|
||||
|
||||
thread = threading.Thread(target=serve_once)
|
||||
thread.start()
|
||||
client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
||||
client_context.check_hostname = False
|
||||
client_context.verify_mode = ssl.CERT_NONE
|
||||
with client_context.wrap_socket(
|
||||
client_socket, server_hostname="proxmenux.test"
|
||||
) as tls_client:
|
||||
received_der = tls_client.getpeercert(binary_form=True)
|
||||
tls_client.sendall(b"x")
|
||||
thread.join(timeout=5)
|
||||
|
||||
self.assertFalse(thread.is_alive())
|
||||
self.assertEqual(server_error, [])
|
||||
renewed_pem = self.pairs[1][0].read_text()
|
||||
self.assertEqual(received_der, ssl.PEM_cert_to_DER_cert(renewed_pem))
|
||||
|
||||
def test_tls_connection_without_sni_also_receives_the_renewed_certificate(self):
|
||||
server_context = self._create_context()
|
||||
self._install_pair(1)
|
||||
server_socket, client_socket = socket.socketpair()
|
||||
server_socket.settimeout(5)
|
||||
client_socket.settimeout(5)
|
||||
server_error = []
|
||||
|
||||
def serve_once():
|
||||
try:
|
||||
with server_context.wrap_socket(server_socket, server_side=True) as tls_socket:
|
||||
tls_socket.recv(1)
|
||||
except Exception as error: # pragma: no cover - asserted below
|
||||
server_error.append(error)
|
||||
|
||||
thread = threading.Thread(target=serve_once)
|
||||
thread.start()
|
||||
client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
||||
client_context.check_hostname = False
|
||||
client_context.verify_mode = ssl.CERT_NONE
|
||||
with client_context.wrap_socket(client_socket) as tls_client:
|
||||
received_der = tls_client.getpeercert(binary_form=True)
|
||||
tls_client.sendall(b"x")
|
||||
thread.join(timeout=5)
|
||||
|
||||
self.assertFalse(thread.is_alive())
|
||||
self.assertEqual(server_error, [])
|
||||
renewed_pem = self.pairs[1][0].read_text()
|
||||
self.assertEqual(received_der, ssl.PEM_cert_to_DER_cert(renewed_pem))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,69 @@
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
import lxc_apps # noqa: E402
|
||||
|
||||
|
||||
class ScheduledUpdateRecordTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp_dir.cleanup)
|
||||
self.apps_dir_patch = mock.patch.object(
|
||||
lxc_apps, '_APPS_DIR', self.temp_dir.name,
|
||||
)
|
||||
self.apps_dir_patch.start()
|
||||
self.addCleanup(self.apps_dir_patch.stop)
|
||||
self.vmid = 9911
|
||||
self.assertTrue(lxc_apps._write_sidecar(self.vmid, {
|
||||
'vmid': self.vmid,
|
||||
'apps': [],
|
||||
'schedule': {'enabled': True, 'cron': '0 3 * * *'},
|
||||
}))
|
||||
|
||||
def test_record_keeps_log_and_reboot_evidence(self):
|
||||
self.assertTrue(lxc_apps.record_schedule_run(
|
||||
self.vmid,
|
||||
'success',
|
||||
'both',
|
||||
log_name='../9911-scheduled-' + ('a' * 32) + '.log',
|
||||
reboot_required=True,
|
||||
reboot_packages=['linux-image-amd64', 'libc6'],
|
||||
))
|
||||
schedule = lxc_apps._read_sidecar(self.vmid)['schedule']
|
||||
self.assertEqual(
|
||||
schedule['last_run_log'],
|
||||
'9911-scheduled-' + ('a' * 32) + '.log',
|
||||
)
|
||||
self.assertTrue(schedule['last_run_reboot_required'])
|
||||
self.assertEqual(
|
||||
schedule['last_run_reboot_packages'],
|
||||
['linux-image-amd64', 'libc6'],
|
||||
)
|
||||
|
||||
def test_lifecycle_clear_preserves_run_and_log(self):
|
||||
lxc_apps.record_schedule_run(
|
||||
self.vmid,
|
||||
'success',
|
||||
'os',
|
||||
log_name='9911-scheduled-' + ('b' * 32) + '.log',
|
||||
reboot_required=True,
|
||||
reboot_packages=['linux-image-amd64'],
|
||||
)
|
||||
self.assertTrue(lxc_apps.clear_schedule_reboot_required(self.vmid))
|
||||
schedule = lxc_apps._read_sidecar(self.vmid)['schedule']
|
||||
self.assertFalse(schedule['last_run_reboot_required'])
|
||||
self.assertNotIn('last_run_reboot_packages', schedule)
|
||||
self.assertEqual(schedule['last_run_status'], 'success')
|
||||
self.assertIn('last_run_log', schedule)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,160 @@
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
import notification_manager # noqa: E402
|
||||
from notification_events import NotificationEvent # noqa: E402
|
||||
|
||||
|
||||
class RecordingChannel:
|
||||
def __init__(self, results=None):
|
||||
self.results = list(results or [True])
|
||||
self.calls = 0
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def send(self, title, body, severity, data):
|
||||
with self.lock:
|
||||
self.calls += 1
|
||||
success = self.results.pop(0) if self.results else True
|
||||
time.sleep(0.1)
|
||||
return {'success': success, 'error': '' if success else 'temporary failure'}
|
||||
|
||||
|
||||
class NotificationDeliveryDedupTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp_dir.cleanup)
|
||||
self.db_path = Path(self.temp_dir.name) / 'health_monitor.db'
|
||||
conn = sqlite3.connect(str(self.db_path))
|
||||
conn.execute('''
|
||||
CREATE TABLE notification_last_sent (
|
||||
fingerprint TEXT PRIMARY KEY,
|
||||
last_sent_ts INTEGER NOT NULL,
|
||||
count INTEGER DEFAULT 1
|
||||
)
|
||||
''')
|
||||
conn.execute('''
|
||||
CREATE TABLE notification_history (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
event_type TEXT NOT NULL,
|
||||
channel TEXT NOT NULL,
|
||||
title TEXT,
|
||||
message TEXT,
|
||||
severity TEXT,
|
||||
sent_at TEXT NOT NULL,
|
||||
success INTEGER DEFAULT 1,
|
||||
error_message TEXT,
|
||||
source TEXT DEFAULT 'server'
|
||||
)
|
||||
''')
|
||||
conn.commit()
|
||||
conn.close()
|
||||
self.db_patch = mock.patch.object(
|
||||
notification_manager, 'DB_PATH', self.db_path,
|
||||
)
|
||||
self.db_patch.start()
|
||||
self.addCleanup(self.db_patch.stop)
|
||||
self.ai_context_patch = mock.patch.object(
|
||||
notification_manager, 'enrich_context_for_ai', return_value='',
|
||||
)
|
||||
self.ai_context_patch.start()
|
||||
self.addCleanup(self.ai_context_patch.stop)
|
||||
self.ai_rewrite_patch = mock.patch.object(
|
||||
notification_manager, '_format_with_ai_bounded', return_value=None,
|
||||
)
|
||||
self.ai_rewrite_patch.start()
|
||||
self.addCleanup(self.ai_rewrite_patch.stop)
|
||||
|
||||
def _manager(self, channel):
|
||||
manager = notification_manager.NotificationManager()
|
||||
manager._enabled = True
|
||||
manager._config = {
|
||||
'email.enabled': 'true',
|
||||
'email.rich_format': 'false',
|
||||
'ai_enabled': 'false',
|
||||
}
|
||||
manager._channels = {'email': channel}
|
||||
return manager
|
||||
|
||||
@staticmethod
|
||||
def _event():
|
||||
return NotificationEvent(
|
||||
event_type='lxc_update_applied',
|
||||
severity='INFO',
|
||||
data={
|
||||
'hostname': 'pve-test',
|
||||
'vmid': 210,
|
||||
'ct_name': 'docker-frontend',
|
||||
'target': 'Docker Engine',
|
||||
'result': 'succeeded',
|
||||
'duration': '10s',
|
||||
'details': 'Update completed',
|
||||
},
|
||||
source='manual',
|
||||
entity='ct',
|
||||
entity_id='210:same-run',
|
||||
)
|
||||
|
||||
def test_concurrent_managers_deliver_same_fingerprint_once(self):
|
||||
channel = RecordingChannel()
|
||||
managers = [self._manager(channel), self._manager(channel)]
|
||||
barrier = threading.Barrier(3)
|
||||
|
||||
def dispatch(manager):
|
||||
barrier.wait()
|
||||
manager._dispatch_event(self._event())
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=dispatch, args=(manager,))
|
||||
for manager in managers
|
||||
]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
barrier.wait()
|
||||
for thread in threads:
|
||||
thread.join(timeout=5)
|
||||
|
||||
self.assertEqual(channel.calls, 1)
|
||||
conn = sqlite3.connect(str(self.db_path))
|
||||
history_count = conn.execute(
|
||||
'SELECT COUNT(*) FROM notification_history WHERE success = 1'
|
||||
).fetchone()[0]
|
||||
claim_count = conn.execute(
|
||||
'SELECT COUNT(*) FROM notification_delivery_claims'
|
||||
).fetchone()[0]
|
||||
conn.close()
|
||||
self.assertEqual(history_count, 1)
|
||||
self.assertEqual(claim_count, 0)
|
||||
|
||||
def test_failed_delivery_releases_claim_for_retry(self):
|
||||
channel = RecordingChannel([False, True])
|
||||
manager = self._manager(channel)
|
||||
|
||||
manager._dispatch_event(self._event())
|
||||
manager._dispatch_event(self._event())
|
||||
|
||||
self.assertEqual(channel.calls, 2)
|
||||
conn = sqlite3.connect(str(self.db_path))
|
||||
successes = conn.execute(
|
||||
'SELECT success FROM notification_history ORDER BY id'
|
||||
).fetchall()
|
||||
claim_count = conn.execute(
|
||||
'SELECT COUNT(*) FROM notification_delivery_claims'
|
||||
).fetchone()[0]
|
||||
conn.close()
|
||||
self.assertEqual(successes, [(0,), (1,)])
|
||||
self.assertEqual(claim_count, 0)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,114 @@
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from queue import Queue
|
||||
from unittest import mock
|
||||
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
import notification_events # noqa: E402
|
||||
import notification_templates # noqa: E402
|
||||
import post_install_versions # noqa: E402
|
||||
|
||||
|
||||
class NotificationUpdatePolicyTests(unittest.TestCase):
|
||||
def test_apt_listchanges_system_mail_has_its_own_update_event(self):
|
||||
watcher = notification_events.ProxmoxHookWatcher(Queue())
|
||||
classified = watcher._classify_pve(
|
||||
"system-mail",
|
||||
"info",
|
||||
"Novedades de apt-listchanges para amd",
|
||||
"zfs-linux recommends that all users update absolute paths",
|
||||
)
|
||||
self.assertEqual(classified, ("apt_listchanges", "node", ""))
|
||||
|
||||
def test_regular_system_mail_remains_available(self):
|
||||
watcher = notification_events.ProxmoxHookWatcher(Queue())
|
||||
with mock.patch.object(notification_events, "_record_smartd_observation_impl"):
|
||||
classified = watcher._classify_pve(
|
||||
"system-mail",
|
||||
"warning",
|
||||
"SMART error (CurrentPendingSector) detected on host",
|
||||
"Device: /dev/sda",
|
||||
)
|
||||
self.assertEqual(classified, ("system_mail", "node", ""))
|
||||
|
||||
def test_apt_listchanges_body_is_preserved_and_attributed(self):
|
||||
watcher = notification_events.ProxmoxHookWatcher(Queue())
|
||||
upstream = (
|
||||
"zfs-linux (2.2.4-2) unstable; urgency=medium\n\n"
|
||||
" Package-maintainer recommendation.\n\n"
|
||||
" -- Maintainer <maintainer@example.com>"
|
||||
)
|
||||
result = watcher.process_webhook({
|
||||
"title": "apt-listchanges: News for host",
|
||||
"message": upstream,
|
||||
"severity": "info",
|
||||
"fields": {"type": "system-mail", "hostname": "pve-test"},
|
||||
})
|
||||
self.assertTrue(result["accepted"])
|
||||
event = watcher._queue.get_nowait()
|
||||
self.assertEqual(event.event_type, "apt_listchanges")
|
||||
self.assertEqual(event.data["reason"], upstream)
|
||||
|
||||
rendered = notification_templates.render_template(
|
||||
event.event_type,
|
||||
event.data,
|
||||
)
|
||||
self.assertIn("not a ProxMenux recommendation", rendered["body_text"])
|
||||
self.assertIn(upstream, rendered["body_text"])
|
||||
|
||||
def test_smaller_pending_subset_does_not_notify_again(self):
|
||||
updates = [
|
||||
{"key": "persistent_network", "available_version": "1.2"},
|
||||
]
|
||||
notified = {
|
||||
"log2ram": {"1.4"},
|
||||
"persistent_network": {"1.2"},
|
||||
}
|
||||
self.assertEqual(
|
||||
notification_events._new_post_install_update_versions(updates, notified),
|
||||
{},
|
||||
)
|
||||
|
||||
def test_new_version_of_existing_tool_is_detected(self):
|
||||
updates = [
|
||||
{"key": "persistent_network", "available_version": "1.3"},
|
||||
]
|
||||
notified = {"persistent_network": {"1.2"}}
|
||||
self.assertEqual(
|
||||
notification_events._new_post_install_update_versions(updates, notified),
|
||||
{"persistent_network": "1.3"},
|
||||
)
|
||||
|
||||
def test_notified_versions_share_the_existing_snapshot_file(self):
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
snapshot_path = Path(temporary) / "updates_available.json"
|
||||
cache = {
|
||||
"scanned_at": 123.0,
|
||||
"updates": [
|
||||
{"key": "log2ram", "available_version": "1.4"},
|
||||
],
|
||||
}
|
||||
with mock.patch.object(post_install_versions, "_UPDATES_JSON", snapshot_path), \
|
||||
mock.patch.object(post_install_versions, "_cache", cache):
|
||||
post_install_versions.save_notified_versions(
|
||||
{"log2ram": {"1.3", "1.4"}}
|
||||
)
|
||||
self.assertEqual(
|
||||
post_install_versions.load_notified_versions(),
|
||||
{"log2ram": {"1.3", "1.4"}},
|
||||
)
|
||||
payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
||||
self.assertEqual(payload["scanned_at"], 123.0)
|
||||
self.assertEqual(payload["updates"], cache["updates"])
|
||||
self.assertEqual(payload["notified_versions"]["log2ram"], ["1.3", "1.4"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,37 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
import lxc_apps
|
||||
|
||||
|
||||
class AdguardSetupTests(unittest.TestCase):
|
||||
def test_non_adguard_is_not_probed(self):
|
||||
with patch.object(lxc_apps, '_oci_instance_meta', return_value={'template_id': 'image-frigate'}), \
|
||||
patch.object(lxc_apps.subprocess, 'run') as run:
|
||||
self.assertFalse(lxc_apps.oci_adguard_setup_available(190))
|
||||
run.assert_not_called()
|
||||
|
||||
def test_setup_returns_true_only_for_http_response(self):
|
||||
process = MagicMock(stdout='192.168.0.42\n')
|
||||
connection = MagicMock()
|
||||
connection.__enter__.return_value.recv.return_value = b'HTTP/1.1 302 Found\r\n'
|
||||
with patch.object(lxc_apps, '_oci_instance_meta', return_value={'template_id': 'image-adguard-home'}), \
|
||||
patch.object(lxc_apps.subprocess, 'run', return_value=process), \
|
||||
patch.object(lxc_apps.socket, 'create_connection', return_value=connection) as connect:
|
||||
self.assertTrue(lxc_apps.oci_adguard_setup_available(190))
|
||||
connect.assert_called_once_with(('192.168.0.42', 3000), timeout=1)
|
||||
|
||||
def test_closed_setup_port_is_not_offered(self):
|
||||
process = MagicMock(stdout='192.168.0.42\n')
|
||||
with patch.object(lxc_apps, '_oci_instance_meta', return_value={'template_id': 'image-adguard-home'}), \
|
||||
patch.object(lxc_apps.subprocess, 'run', return_value=process), \
|
||||
patch.object(lxc_apps.socket, 'create_connection', side_effect=OSError):
|
||||
self.assertFalse(lxc_apps.oci_adguard_setup_available(190))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,55 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression tests based on LeidenSpain's real LXC OOM block."""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
from proxmox_known_errors import ( # noqa: E402
|
||||
analyze_oom_event,
|
||||
format_oom_diagnosis,
|
||||
get_error_context,
|
||||
)
|
||||
|
||||
|
||||
LXC_OOM = """
|
||||
apt-get invoked oom-killer: gfp_mask=0x101cca, order=3, oom_score_adj=0
|
||||
memory: usage 131056kB, limit 131072kB, failcnt 1922
|
||||
swap: usage 0kB, limit 0kB, failcnt 0
|
||||
Memory cgroup stats for /lxc/108:
|
||||
oom-kill:constraint=CONSTRAINT_MEMCG,nodemask=(null),cpuset=ns,mems_allowed=0,oom_memcg=/lxc/108,task_memcg=/lxc/108/ns/.lxc,task=apt-get,pid=1183877,uid=100000
|
||||
Memory cgroup out of memory: Killed process 1183877 (apt-get) total-vm:79700kB, anon-rss:65056kB
|
||||
"""
|
||||
|
||||
|
||||
class OomDiagnosticsTest(unittest.TestCase):
|
||||
def test_lxc_memcg_scope_and_limits(self):
|
||||
result = analyze_oom_event(LXC_OOM)
|
||||
self.assertIsNotNone(result)
|
||||
self.assertEqual(result['scope'], 'lxc')
|
||||
self.assertEqual(result['ctid'], '108')
|
||||
self.assertEqual(result['constraint'], 'CONSTRAINT_MEMCG')
|
||||
self.assertEqual(result['memory_usage_kib'], 131056)
|
||||
self.assertEqual(result['memory_limit_kib'], 131072)
|
||||
self.assertEqual(result['swap_limit_kib'], 0)
|
||||
self.assertEqual(result['victim_process'], 'apt-get')
|
||||
self.assertEqual(result['victim_pid'], '1183877')
|
||||
|
||||
def test_diagnosis_does_not_blame_host(self):
|
||||
diagnosis = format_oom_diagnosis(analyze_oom_event(LXC_OOM))
|
||||
self.assertIn('LXC 108', diagnosis)
|
||||
self.assertIn('not a host-wide OOM', diagnosis)
|
||||
self.assertIn('128.0 MiB used of 128.0 MiB', diagnosis)
|
||||
self.assertIn('Killed process: apt-get', diagnosis)
|
||||
|
||||
def test_known_error_context_includes_event_evidence(self):
|
||||
context = get_error_context(LXC_OOM, category='memory', detail_level='detailed')
|
||||
self.assertIn('Event analysis:', context)
|
||||
self.assertIn('LXC 108 memory cgroup', context)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user