fix(post-install): prioritize ZRAM ahead of disk swap

AppImage/scripts/flask_proxmenux_routes.py

@@ -26,7 +26,7 @@ TOOL_METADATA = {
     'system_limits':        {'name': 'System Limits Increase',                'function': 'increase_system_limits',       'version': '1.1'},
     # entropy removed — modern kernels 5.6+ have built-in entropy generation, haveged no longer needed
     'memory_settings':      {'name': 'Memory Settings Optimization',          'function': 'optimize_memory_settings',     'version': '1.1'},
-    'zram_swap':            {'name': 'ZRAM-only Swap Profile',                 'function': 'configure_zram_swap',         'version': '1.0'},
+    'zram_swap':            {'name': 'ZRAM Swap Profile',                      'function': 'configure_zram_swap',         'version': '1.0'},
     'kernel_panic':         {'name': 'Kernel Panic Configuration',            'function': 'configure_kernel_panic',       'version': '1.0'},
     'apt_ipv4':             {'name': 'APT IPv4 Force',                        'function': 'force_apt_ipv4',               'version': '1.0'},
     'kexec':                {'name': 'kexec for quick reboots',               'function': 'enable_kexec',                 'version': '1.0'},

AppImage/scripts/tests/test_zram_profile_helpers.py

@@ -1,5 +1,6 @@
 """Contract tests for the opt-in ZRAM profile's pure Bash helpers."""
 
+import os
 import subprocess
 import tempfile
 import unittest
@@ -14,6 +15,16 @@ class HelperAvailabilityTests(unittest.TestCase):
     def test_helper_file_is_available(self):
         self.assertTrue(HELPERS.is_file(), f"ZRAM helper is missing: {HELPERS}")
 
+    def test_helper_can_be_sourced_twice(self):
+        result = subprocess.run(
+            ["bash", "-c", 'source "$1" && source "$1"', "bash", str(HELPERS)],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+
 
 def call_helper(function, *arguments):
     return subprocess.run(
@@ -44,7 +55,7 @@ class ZramGeneratorConfigTests(unittest.TestCase):
         }
         self.assertIn("[zram0]", result.stdout.splitlines())
         self.assertEqual(settings.get("zram-size", "").strip(), "ram / 10")
-        self.assertEqual(settings.get("swap-priority", "").strip(), "100")
+        self.assertEqual(settings.get("swap-priority", "").strip(), "32767")
         self.assertEqual(settings.get("compression-algorithm", "").strip(), expected_algorithm)
         self.assertNotIn("writeback", result.stdout.lower())
 
@@ -55,44 +66,107 @@ class ZramGeneratorConfigTests(unittest.TestCase):
         self.assert_config("true", "zstd(level=9)")
 
 
-class NonZramSwapDetectionTests(unittest.TestCase):
+class ForeignActiveZramSwapTests(unittest.TestCase):
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
         directory = Path(self.temporary.name)
+        self.root = directory
         self.swaps_file = directory / "swaps"
-        self.fstab_file = directory / "fstab"
 
-    def detect(self, swaps, fstab):
+    def detect(self, swaps, owned_device=""):
         self.swaps_file.write_text(swaps, encoding="utf-8")
-        self.fstab_file.write_text(fstab, encoding="utf-8")
-        return call_helper("zram_has_non_zram_swap", self.swaps_file, self.fstab_file)
+        return call_helper(
+            "zram_list_foreign_active_zram_swap", self.swaps_file, owned_device
+        )
 
-    def test_active_disk_swap_is_detected(self):
+    def test_active_disk_swap_is_allowed(self):
         result = self.detect(
             SWAPS_HEADER + "/dev/nvme0n1p3\tpartition\t1048576\t0\t-2\n",
-            "",
         )
         self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
 
-    def test_persistent_swapfile_is_detected(self):
-        result = self.detect(SWAPS_HEADER, "/swapfile none swap sw 0 0\n")
+    def test_active_unowned_zram0_is_detected(self):
+        result = self.detect(SWAPS_HEADER + "/dev/zram0\tpartition\t1048576\t0\t100\n")
         self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "active zram: /dev/zram0\n")
 
-    def test_commented_disk_swap_is_ignored(self):
-        result = self.detect(SWAPS_HEADER, "  # /swapfile none swap sw 0 0\n")
-        self.assertEqual(result.returncode, 1, result.stderr)
-
-    def test_active_and_persistent_zram_are_ignored(self):
+    def test_active_owned_zram0_is_allowed_alongside_disk_swap(self):
         result = self.detect(
-            SWAPS_HEADER + "/dev/zram0\tpartition\t1048576\t0\t100\n",
-            "/dev/zram0 none swap defaults 0 0\n",
+            SWAPS_HEADER
+            + "/dev/nvme0n1p3\tpartition\t1048576\t0\t-2\n"
+            + "/dev/zram0\tpartition\t1048576\t0\t100\n",
+            "/dev/zram0",
         )
-        self.assertEqual(result.returncode, 1, result.stderr)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
 
-    def test_regular_fstab_entry_does_not_count_as_swap(self):
-        result = self.detect(SWAPS_HEADER, "/dev/nvme0n1p2 / ext4 defaults 0 1\n")
-        self.assertEqual(result.returncode, 1, result.stderr)
+    def test_another_active_zram_device_is_rejected(self):
+        result = self.detect(
+            SWAPS_HEADER
+            + "/dev/zram0\tpartition\t1048576\t0\t100\n"
+            + "/dev/zram1\tpartition\t1048576\t0\t90\n",
+            "/dev/zram0",
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "active zram: /dev/zram1\n")
+
+    def test_swap_table_read_failure_is_reported(self):
+        fake_bin = self.root / "bin"
+        fake_bin.mkdir()
+        fake_awk = fake_bin / "awk"
+        fake_awk.write_text("#!/bin/sh\nexit 2\n", encoding="utf-8")
+        fake_awk.chmod(0o755)
+        env = os.environ.copy()
+        env["PATH"] = f"{fake_bin}:{env.get('PATH', '')}"
+
+        result = subprocess.run(
+            [
+                "bash",
+                "-c",
+                'source "$1" || exit; shift; "$@"',
+                "bash",
+                str(HELPERS),
+                "zram_list_foreign_active_zram_swap",
+                str(self.swaps_file),
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+            env=env,
+        )
+
+        self.assertEqual(result.returncode, 2, result.stderr)
+
+
+class ForeignFstabZramSwapTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.fstab_file = Path(self.temporary.name) / "fstab"
+
+    def detect(self, text):
+        self.fstab_file.write_text(text, encoding="utf-8")
+        return call_helper("zram_list_foreign_fstab_zram_swap", self.fstab_file)
+
+    def test_disk_swap_entries_are_allowed(self):
+        result = self.detect(
+            "UUID=swap-partition none swap defaults 0 0\n"
+            "/swapfile none swap sw 0 0\n"
+        )
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+
+    def test_zram_swap_entry_is_reported(self):
+        result = self.detect(
+            "UUID=swap-partition none swap defaults 0 0\n"
+            "/dev/zram1 none swap pri=100 0 0\n"
+        )
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "fstab zram: /dev/zram1\n")
 
 
 class ForeignZramConfigTests(unittest.TestCase):
@@ -132,7 +206,7 @@ class ForeignZramConfigTests(unittest.TestCase):
         self.assertEqual(self.foreign_configs(), [])
 
 
-class SystemdSwapDetectionTests(unittest.TestCase):
+class ForeignSystemdZramUnitTests(unittest.TestCase):
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary.cleanup)
@@ -140,27 +214,143 @@ class SystemdSwapDetectionTests(unittest.TestCase):
         self.system_dir = self.root / "etc/systemd/system"
         self.system_dir.mkdir(parents=True)
 
-    def test_administrator_defined_zram_and_other_swap_units_are_reported(self):
-        (self.system_dir / "dev-zram0.swap").write_text("", encoding="utf-8")
-        (self.system_dir / "custom.swap").write_text("", encoding="utf-8")
+    def test_disk_swap_unit_is_allowed_and_zram_units_are_reported(self):
+        (self.system_dir / "pve-swap.swap").write_text(
+            "[Swap]\nWhat=/dev/mapper/pve-swap\n", encoding="utf-8"
+        )
+        (self.system_dir / "dev-zram0.swap").write_text(
+            "[Swap]\nWhat=/dev/zram0\n", encoding="utf-8"
+        )
+        external = self.system_dir / "external.swap"
+        external.write_text("[Swap]\nWhat=/dev/zram1\n", encoding="utf-8")
+        vendor_dir = self.root / "usr/lib/systemd/system"
+        vendor_dir.mkdir(parents=True)
+        vendor = vendor_dir / "vendor-zram.swap"
+        vendor.write_text("[Swap]\nWhat=/dev/zram2\n", encoding="utf-8")
 
-        result = call_helper("zram_list_defined_systemd_swap_units", self.root)
+        result = call_helper("zram_list_foreign_systemd_zram_units", self.root)
 
         self.assertEqual(result.returncode, 0, result.stderr)
         self.assertEqual(
             set(result.stdout.splitlines()),
             {
-                f"systemd: {self.system_dir / 'dev-zram0.swap'}",
-                f"systemd: {self.system_dir / 'custom.swap'}",
+                f"systemd zram unit: {self.system_dir / 'dev-zram0.swap'}",
+                f"systemd zram unit: {external}",
+                f"systemd zram unit: {vendor}",
             },
         )
 
-    def test_no_non_zram_systemd_swap_has_empty_output(self):
-        result = call_helper("zram_list_defined_systemd_swap_units", self.root)
+    def test_unrelated_custom_swap_unit_is_allowed(self):
+        (self.system_dir / "custom.swap").write_text(
+            "[Swap]\nWhat=/swapfile\n", encoding="utf-8"
+        )
+
+        result = call_helper("zram_list_foreign_systemd_zram_units", self.root)
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+
+    def test_reserved_zram0_unit_masked_with_dev_null_is_reported(self):
+        (self.system_dir / "dev-zram0.swap").symlink_to("/dev/null")
+
+        result = call_helper("zram_list_foreign_systemd_zram_units", self.root)
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(
+            result.stdout,
+            f"systemd zram unit masked: {self.system_dir / 'dev-zram0.swap'}\n",
+        )
+
+    def test_zram_swap_drop_in_is_reported(self):
+        dropins = self.system_dir / "custom.swap.d"
+        dropins.mkdir()
+        dropin = dropins / "10-zram.conf"
+        dropin.write_text("[Swap]\nWhat=/dev/zram1\n", encoding="utf-8")
+
+        result = call_helper("zram_list_foreign_systemd_zram_units", self.root)
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, f"systemd zram unit: {dropin}\n")
+
+    def test_generator_wants_symlink_is_not_a_foreign_unit_file(self):
+        generator = self.root / "run/systemd/generator"
+        generator.mkdir(parents=True)
+        (generator / "dev-zram0.swap").write_text(
+            "[Swap]\nWhat=/dev/zram0\n", encoding="utf-8"
+        )
+        wants = self.root / "run/systemd/system/swap.target.wants"
+        wants.mkdir(parents=True)
+        (wants / "dev-zram0.swap").symlink_to("../../generator/dev-zram0.swap")
+
+        result = call_helper("zram_list_foreign_systemd_zram_units", self.root)
 
         self.assertEqual(result.returncode, 0, result.stderr)
         self.assertEqual(result.stdout, "")
 
 
+class ZramProfileActivationVerifierTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.block_root = self.root / "sys/block"
+        zram0 = self.block_root / "zram0"
+        zram0.mkdir(parents=True)
+        (zram0 / "disksize").write_text("3435973836\n", encoding="utf-8")
+        (zram0 / "comp_algorithm").write_text(
+            "lzo [zstd] lz4\n", encoding="utf-8"
+        )
+        self.meminfo = self.root / "proc/meminfo"
+        self.meminfo.parent.mkdir(parents=True)
+        self.meminfo.write_text("MemTotal:       33554432 kB\n", encoding="utf-8")
+        self.swaps = self.root / "proc/swaps"
+        self.swaps.write_text(SWAPS_HEADER, encoding="utf-8")
+
+    def verify(self):
+        return call_helper(
+            "zram_profile_verify_active",
+            "zstd",
+            "32767",
+            self.block_root,
+            self.meminfo,
+            self.swaps,
+        )
+
+    def test_profile_is_valid_with_lower_priority_disk_swap_active(self):
+        self.swaps.write_text(
+            SWAPS_HEADER
+            + "/dev/zram0\tpartition\t3355443\t0\t32767\n"
+            + "/dev/nvme0n1p3\tpartition\t1048576\t32\t-2\n",
+            encoding="utf-8",
+        )
+
+        result = self.verify()
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_profile_rejects_a_second_active_zram_device(self):
+        self.swaps.write_text(
+            SWAPS_HEADER
+            + "/dev/zram0\tpartition\t3355443\t0\t32767\n"
+            + "/dev/zram1\tpartition\t1048576\t0\t90\n",
+            encoding="utf-8",
+        )
+
+        result = self.verify()
+
+        self.assertEqual(result.returncode, 1, result.stderr)
+
+    def test_profile_rejects_zram_device_without_requested_priority(self):
+        self.swaps.write_text(
+            SWAPS_HEADER
+            + "/dev/zram0\tpartition\t3355443\t0\t100\n",
+            encoding="utf-8",
+        )
+
+        result = self.verify()
+
+        self.assertEqual(result.returncode, 1, result.stderr)
+
+
 if __name__ == "__main__":
     unittest.main()