[PATCH 3/7] tuna: Add comprehensive tests for save/apply functionality

John Kacur <[email protected]> Thu, 16 Jul 2026 13:45:46 -0400
Newsgroups org.kernel.vger.linux-rt-users
Message-ID <[email protected]>
Add Python test suite for cpuset save/apply subcommands with 12 test
cases covering all functionality and edge cases.

Test coverage (tests/test_save_apply.py):
- test_save_creates_yaml_file: Verify save creates output file
- test_save_captures_all_cpusets: Verify all user cpusets are saved
- test_save_captures_partition_types: Verify isolated vs member preserved
- test_save_captures_cpu_ranges: Verify CPU assignments preserved
- test_apply_recreates_cpusets: Verify apply recreates from YAML
- test_apply_preserves_partition_types: Verify partition types restored
- test_apply_preserves_cpu_assignments: Verify CPU assignments restored
- test_apply_with_existing_cpusets_warns: Verify idempotent operation
- test_apply_missing_file_returns_error: Error handling for missing files
- test_roundtrip_preserves_configuration: Full save→destroy→apply→verify
- test_save_with_no_cpusets: Handle empty cpuset case gracefully
- test_save_skips_system_cpusets: Verify system cpuset filtering

Test structure:
- Uses unittest framework (consistent with existing tests)
- Requires root permissions (@skipUnless decorator)
- Requires cgroup v2 support (@skipUnless decorator)
- setUp/tearDown for cleanup (creates temp files, cleans cpusets)
- Cleans up test cpusets from both automated and manual testing
- Helper method create_test_cpusets() for test data
- Uses tempfile module for safe YAML file handling

All tests follow the same pattern as existing cpuset tests in
test_cpuset.py and test_cpuset_cli.py, ensuring consistency
across the test suite.

Tests can be run with:
  sudo python3 -m unittest tests.test_save_apply -v

Assisted-by: Claude:claude-sonnet-4-5
Signed-off-by: John Kacur <[email protected]>
---
 tests/README.md          |  37 +++--
 tests/test_save_apply.py | 300 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 327 insertions(+), 10 deletions(-)
 create mode 100644 tests/test_save_apply.py

diff --git a/tests/README.md b/tests/README.md
index 3e8b84a81fd4..1b96a54807f9 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -12,7 +12,7 @@ This directory contains the test suite for tuna, using Python's `unittest` frame
 # Run all non-root tests only (45 tests)
 make tests
 
-# Run ALL tests (all 146 tests, requires sudo for root-required tests)
+# Run ALL tests (all 158 tests, requires sudo for root-required tests)
 make test-all
 
 # Run specific test module
@@ -37,7 +37,7 @@ make tests
 # or
 make unit-tests
 
-# Using make (ALL tests including root-required tests, 146 tests)
+# Using make (ALL tests including root-required tests, 158 tests)
 make test-all
 
 # Using the test runner directly (non-root tests only)
@@ -168,7 +168,7 @@ Tests are organized using Python's `unittest` framework. Each test file contains
   - Tests edge cases: empty string and None inputs
   - Run: `python3 -m unittest tests.test_show_threads_cgroups.TestExtractCpusetName -v`
 
-#### Root-Required Tests (102 tests)
+#### Root-Required Tests (114 tests)
 
 - **test_process_blocklist.py** (2 tests) - Process blocklist root tests
   - Tests actual blocking of systemd from being moved to cpusets
@@ -226,7 +226,23 @@ Tests are organized using Python's `unittest` framework. Each test file contains
     - test_cgroups_displays_system_slice_shortened: Verifies system.slice shows shortened name
   - Run: `sudo python3 -m unittest tests.test_show_threads_cgroups.TestShowThreadsCgroupsDisplay -v`
 
-**Total: 146 tests** (45 non-root + 101 root-required)
+- **test_save_apply.py** (12 tests) - save/apply cpuset configuration tests
+  - TestSaveApply: Tests cpuset save and apply functionality
+    - test_save_creates_yaml_file: Verify save creates output file
+    - test_save_captures_all_cpusets: Verify all user cpusets are saved
+    - test_save_captures_partition_types: Verify isolated vs member preserved
+    - test_save_captures_cpu_ranges: Verify CPU assignments preserved
+    - test_apply_recreates_cpusets: Verify apply recreates from YAML
+    - test_apply_preserves_partition_types: Verify partition types restored
+    - test_apply_preserves_cpu_assignments: Verify CPU assignments restored
+    - test_apply_with_existing_cpusets_warns: Verify idempotent operation
+    - test_apply_missing_file_returns_error: Error handling for missing files
+    - test_roundtrip_preserves_configuration: Full save→destroy→apply→verify
+    - test_save_with_no_cpusets: Handle empty cpuset case gracefully
+    - test_save_skips_system_cpusets: Verify system cpuset filtering
+  - Run: `sudo python3 -m unittest tests.test_save_apply -v`
+
+**Total: 158 tests** (45 non-root + 113 root-required)
 
 ### Test Requirements: Root vs Non-Root
 
@@ -275,7 +291,7 @@ These tests run without root privileges:
 - Tests are fast and don't affect the running system
 - Tests verify both happy path (with mocks) and error path (with real EPERM)
 
-#### Root-Required Tests (102 tests)
+#### Root-Required Tests (114 tests)
 
 These tests require root to create/manipulate cgroups:
 
@@ -446,9 +462,9 @@ Ran 45 tests in 0.XXXs
 OK (skipped=102)
 ```
 
-### All Tests Including Root-Required (146 tests)
+### All Tests Including Root-Required (158 tests)
 
-Running with sudo will run all 146 tests:
+Running with sudo will run all 158 tests:
 
 ```bash
 $ sudo python3 -m unittest discover -s tests -p "test_*.py" -v
@@ -463,7 +479,7 @@ test_sequential_naming (test_cpuset_cli.TestGetNextTunaCpusetName) ... ok
 ... (4 show_threads -G display tests)
 
 ----------------------------------------------------------------------
-Ran 146 tests in X.XXXs
+Ran 158 tests in X.XXXs
 
 OK
 ```
@@ -480,12 +496,13 @@ The test suite includes:
 - **show_threads --cpuset help test**: CLI documentation verification (read-only)
 - **Cpuset name extraction tests**: Cgroup path parsing logic (pure function testing)
 
-### Integration Tests (102 tests - require root)
+### Integration Tests (114 tests - require root)
 - **Process blocklist tests**: Test blocklist protection (2 root-required tests)
 - **Cpuset module tests**: Test interaction with cgroup v2 subsystem
 - **Cpuset CLI tests**: Test command-line interface integration
 - **show_threads --cpuset tests**: Test process filtering by cpuset membership
 - **show_threads -G display tests**: Test cpuset name display in cgroups output
+- **Save/apply tests**: Test cpuset configuration save and restore functionality
 
 Future test categories might include:
 - **System tests** - Full end-to-end tests requiring CPU affinity operations
@@ -510,7 +527,7 @@ python3 -m unittest discover -s tests -p "test_*.py" -v
 
 ### Root-Required Tests (Optional for CI)
 
-The cpuset tests (102 tests) can run in CI with special setup:
+The cpuset tests (114 tests) can run in CI with special setup:
 - Requires root access or privileged containers
 - Requires cgroup v2 support
 - May need dedicated test runners with appropriate permissions
diff --git a/tests/test_save_apply.py b/tests/test_save_apply.py
new file mode 100644
index 000000000000..30630d2a0b92
--- /dev/null
+++ b/tests/test_save_apply.py
@@ -0,0 +1,300 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# SPDX-License-Identifier: GPL-2.0-only
+"""
+Test suite for tuna cpuset save/apply commands
+
+Tests the cpuset save and apply functionality:
+- save: Saving cpuset configuration to YAML
+- apply: Restoring cpuset configuration from YAML
+- Round-trip: save → destroy → apply → verify
+- Edge cases: existing cpusets, missing files, empty profiles, etc.
+
+All tests clean up created files and cpusets afterwards.
+"""
+
+import unittest
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+# Add tuna to path if running from tests directory
+if Path(__file__).parent.name == 'tests':
+    sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from tuna import cpuset, profile
+
+
+# Import the CLI functions from tuna-cmd.py
+tuna_cmd_path = Path(__file__).parent.parent / 'tuna-cmd.py'
+spec = __import__('importlib.util').util.spec_from_file_location("tuna_cmd", tuna_cmd_path)
+tuna_cmd = __import__('importlib.util').util.module_from_spec(spec)
+spec.loader.exec_module(tuna_cmd)
+
+# Import the functions we want to test
+save_cpusets_cmd = tuna_cmd.save_cpusets_cmd
+apply_cpusets_cmd = tuna_cmd.apply_cpusets_cmd
+get_all_cpusets_info = tuna_cmd.get_all_cpusets_info
+
+
[email protected](os.geteuid() == 0, "Requires root permissions")
[email protected](cpuset.CpusetsInit().supported, "Requires cgroup v2 support")
+class TestSaveApply(unittest.TestCase):
+    """Test save and apply commands for cpuset configuration"""
+
+    def setUp(self):
+        """Create temporary file and test cpusets before each test"""
+        self.ci = cpuset.CpusetsInit()
+
+        # Create temporary YAML file
+        self.temp_fd, self.temp_file = tempfile.mkstemp(suffix='.yaml', prefix='tuna-test-')
+        os.close(self.temp_fd)  # Close fd, we'll use the filename
+
+        # Clean up any existing test cpusets (including from manual tests)
+        for pattern in ['test_save_*', 'test_rt', 'test_housekeeping']:
+            try:
+                cpuset.cleanup_cpusets(pattern, force=True, recursive=False)
+            except:
+                pass
+
+    def tearDown(self):
+        """Clean up test cpusets and temporary file after each test"""
+        # Clean up test cpusets (including from manual tests)
+        for pattern in ['test_save_*', 'test_rt', 'test_housekeeping']:
+            try:
+                cpuset.cleanup_cpusets(pattern, force=True, recursive=False)
+            except:
+                pass
+
+        # Remove temporary YAML file
+        if os.path.exists(self.temp_file):
+            os.unlink(self.temp_file)
+
+    def create_test_cpusets(self):
+        """Helper: Create two test cpusets for testing"""
+        # Create test_save_rt (isolated)
+        cs_rt = cpuset.Cpuset('test_save_rt')
+        cs_rt.write_memnode('0')
+        cs_rt.assign_cpus('2-3')
+        cs_rt.set_cpu_exclusive()
+
+        # Create test_save_hk (member)
+        cs_hk = cpuset.Cpuset('test_save_hk')
+        cs_hk.write_memnode('0')
+        cs_hk.assign_cpus('0-1')
+
+        return cs_rt, cs_hk
+
+    def test_save_creates_yaml_file(self):
+        """Test that save command creates a YAML file"""
+        self.create_test_cpusets()
+
+        # Save cpusets
+        save_cpusets_cmd(self.temp_file)
+
+        # Verify file exists and has content
+        self.assertTrue(os.path.exists(self.temp_file))
+        self.assertGreater(os.path.getsize(self.temp_file), 0)
+
+    def test_save_captures_all_cpusets(self):
+        """Test that save captures all user cpusets"""
+        self.create_test_cpusets()
+
+        # Save cpusets
+        save_cpusets_cmd(self.temp_file)
+
+        # Read and verify YAML content
+        from ruamel.yaml import YAML
+        yaml = YAML()
+        with open(self.temp_file, 'r') as f:
+            data = yaml.load(f)
+
+        self.assertIn('cpusets', data)
+        self.assertEqual(len(data['cpusets']), 2)
+
+        # Check cpuset names
+        names = {cs['name'] for cs in data['cpusets']}
+        self.assertEqual(names, {'test_save_rt', 'test_save_hk'})
+
+    def test_save_captures_partition_types(self):
+        """Test that save preserves partition types (isolated vs member)"""
+        self.create_test_cpusets()
+
+        # Save cpusets
+        save_cpusets_cmd(self.temp_file)
+
+        # Read YAML
+        from ruamel.yaml import YAML
+        yaml = YAML()
+        with open(self.temp_file, 'r') as f:
+            data = yaml.load(f)
+
+        # Find each cpuset in saved data
+        cpusets_by_name = {cs['name']: cs for cs in data['cpusets']}
+
+        self.assertEqual(cpusets_by_name['test_save_rt']['partition'], 'isolated')
+        self.assertEqual(cpusets_by_name['test_save_hk']['partition'], 'member')
+
+    def test_save_captures_cpu_ranges(self):
+        """Test that save captures CPU assignments correctly"""
+        self.create_test_cpusets()
+
+        # Save cpusets
+        save_cpusets_cmd(self.temp_file)
+
+        # Read YAML
+        from ruamel.yaml import YAML
+        yaml = YAML()
+        with open(self.temp_file, 'r') as f:
+            data = yaml.load(f)
+
+        # Find each cpuset in saved data
+        cpusets_by_name = {cs['name']: cs for cs in data['cpusets']}
+
+        self.assertEqual(cpusets_by_name['test_save_rt']['cpus'], '2-3')
+        self.assertEqual(cpusets_by_name['test_save_hk']['cpus'], '0-1')
+
+    def test_apply_recreates_cpusets(self):
+        """Test that apply recreates cpusets from YAML"""
+        # Create and save cpusets
+        self.create_test_cpusets()
+        save_cpusets_cmd(self.temp_file)
+
+        # Destroy cpusets
+        cpuset.cleanup_cpusets('test_save_*', force=True, recursive=False)
+
+        # Verify they're gone
+        remaining = cpuset.list_cpusets(pattern='test_save_*', recursive=False)
+        self.assertEqual(len(remaining), 0)
+
+        # Apply profile to recreate
+        apply_cpusets_cmd(self.temp_file, verbose=False)
+
+        # Verify cpusets were recreated
+        recreated = cpuset.list_cpusets(pattern='test_save_*', recursive=False)
+        self.assertEqual(len(recreated), 2)
+        self.assertIn('test_save_rt', recreated)
+        self.assertIn('test_save_hk', recreated)
+
+    def test_apply_preserves_partition_types(self):
+        """Test that apply restores partition types correctly"""
+        # Create, save, destroy, apply
+        self.create_test_cpusets()
+        save_cpusets_cmd(self.temp_file)
+        cpuset.cleanup_cpusets('test_save_*', force=True, recursive=False)
+        apply_cpusets_cmd(self.temp_file, verbose=False)
+
+        # Verify partition types
+        cs_rt = cpuset.Cpuset('test_save_rt', existing_ok=True)
+        cs_hk = cpuset.Cpuset('test_save_hk', existing_ok=True)
+
+        self.assertEqual(cs_rt.get_partition_type(), 'isolated')
+        self.assertEqual(cs_hk.get_partition_type(), 'member')
+
+    def test_apply_preserves_cpu_assignments(self):
+        """Test that apply restores CPU assignments correctly"""
+        # Create, save, destroy, apply
+        self.create_test_cpusets()
+        save_cpusets_cmd(self.temp_file)
+        cpuset.cleanup_cpusets('test_save_*', force=True, recursive=False)
+        apply_cpusets_cmd(self.temp_file, verbose=False)
+
+        # Verify CPU assignments
+        cs_rt = cpuset.Cpuset('test_save_rt', existing_ok=True)
+        cs_hk = cpuset.Cpuset('test_save_hk', existing_ok=True)
+
+        self.assertEqual(cs_rt.get_cpus(), '2-3')
+        self.assertEqual(cs_hk.get_cpus(), '0-1')
+
+    def test_apply_with_existing_cpusets_warns(self):
+        """Test that apply warns when cpusets already exist"""
+        # Create and save cpusets
+        self.create_test_cpusets()
+        save_cpusets_cmd(self.temp_file)
+
+        # Apply without destroying (cpusets already exist)
+        # This should return warnings
+        num_applied, num_warnings, num_errors = profile.apply_cpusets(
+            self.temp_file, cpuset, verbose=False
+        )
+
+        self.assertEqual(num_applied, 0)  # Nothing new created
+        self.assertEqual(num_warnings, 2)  # Both cpusets already exist
+        self.assertEqual(num_errors, 0)
+
+    def test_apply_missing_file_returns_error(self):
+        """Test that apply returns error for missing file"""
+        missing_file = '/tmp/nonexistent-tuna-test.yaml'
+
+        num_applied, num_warnings, num_errors = profile.apply_cpusets(
+            missing_file, cpuset, verbose=False
+        )
+
+        self.assertEqual(num_applied, 0)
+        self.assertEqual(num_warnings, 0)
+        self.assertEqual(num_errors, 1)
+
+    def test_roundtrip_preserves_configuration(self):
+        """Test full round-trip: save → destroy → apply → verify"""
+        # Step 1: Create test cpusets
+        cs_rt, cs_hk = self.create_test_cpusets()
+        original_rt_cpus = cs_rt.get_cpus()
+        original_hk_cpus = cs_hk.get_cpus()
+        original_rt_partition = cs_rt.get_partition_type()
+        original_hk_partition = cs_hk.get_partition_type()
+
+        # Step 2: Save configuration
+        save_cpusets_cmd(self.temp_file)
+
+        # Step 3: Destroy cpusets
+        cpuset.cleanup_cpusets('test_save_*', force=True, recursive=False)
+
+        # Step 4: Apply configuration
+        apply_cpusets_cmd(self.temp_file, verbose=False)
+
+        # Step 5: Verify configuration matches original
+        cs_rt_restored = cpuset.Cpuset('test_save_rt', existing_ok=True)
+        cs_hk_restored = cpuset.Cpuset('test_save_hk', existing_ok=True)
+
+        self.assertEqual(cs_rt_restored.get_cpus(), original_rt_cpus)
+        self.assertEqual(cs_hk_restored.get_cpus(), original_hk_cpus)
+        self.assertEqual(cs_rt_restored.get_partition_type(), original_rt_partition)
+        self.assertEqual(cs_hk_restored.get_partition_type(), original_hk_partition)
+
+    def test_save_with_no_cpusets(self):
+        """Test that save handles case with no cpusets gracefully"""
+        # Don't create any cpusets
+        # save_cpusets_cmd will print to stderr but shouldn't crash
+
+        # This should complete without error but save 0 cpusets
+        num_saved = profile.save_cpusets(self.temp_file, get_all_cpusets_info)
+        self.assertEqual(num_saved, 0)
+
+    def test_save_skips_system_cpusets(self):
+        """Test that save automatically filters out system cpusets"""
+        # Create one test cpuset
+        cs = cpuset.Cpuset('test_save_single')
+        cs.write_memnode('0')
+        cs.assign_cpus('0')
+
+        # Save (should only save test_save_single, not any .slice/.scope/.service cpusets)
+        save_cpusets_cmd(self.temp_file)
+
+        # Read YAML
+        from ruamel.yaml import YAML
+        yaml = YAML()
+        with open(self.temp_file, 'r') as f:
+            data = yaml.load(f)
+
+        # Verify only our test cpuset was saved (no system cpusets)
+        saved_names = [cs['name'] for cs in data['cpusets']]
+        for name in saved_names:
+            self.assertFalse(name.endswith(('.slice', '.scope', '.mount', '.service')))
+
+        # Cleanup
+        cpuset.destroy_cpuset('test_save_single', force=True)
+
+
+if __name__ == '__main__':
+    unittest.main()
-- 
2.55.0