1 files changed, 79 insertions, 0 deletions
diff --git a/tests/functional/test_memlock.py b/tests/functional/test_memlock.py
new file mode 100755
index 0000000..2b515ff
--- /dev/null
+++ b/tests/functional/test_memlock.py
@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+#
+# Functional test that check overcommit memlock options
+#
+# Copyright (c) Yandex Technologies LLC, 2025
+#
+# Author:
+#  Alexandr Moshkov <dtalexundeer@yandex-team.ru>
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+import re
+
+from typing import Dict
+
+from qemu_test import QemuSystemTest
+from qemu_test import skipLockedMemoryTest
+
+
+STATUS_VALUE_PATTERN = re.compile(r'^(\w+):\s+(\d+) kB', re.MULTILINE)
+
+
+@skipLockedMemoryTest(2_097_152)  # 2GB
+class MemlockTest(QemuSystemTest):
+    """
+    Runs a guest with memlock options.
+    Then verify, that this options is working correctly
+    by checking the status file of the QEMU process.
+    """
+
+    def common_vm_setup_with_memlock(self, memlock):
+        self.vm.add_args('-overcommit', f'mem-lock={memlock}')
+        self.vm.launch()
+
+    def test_memlock_off(self):
+        self.common_vm_setup_with_memlock('off')
+
+        status = self.get_process_status_values(self.vm.get_pid())
+
+        self.assertTrue(status['VmLck'] == 0)
+
+    def test_memlock_on(self):
+        self.common_vm_setup_with_memlock('on')
+
+        status = self.get_process_status_values(self.vm.get_pid())
+
+        # VmLck > 0 kB and almost all memory is resident
+        self.assertTrue(status['VmLck'] > 0)
+        self.assertTrue(status['VmRSS'] >= status['VmSize'] * 0.70)
+
+    def test_memlock_onfault(self):
+        self.common_vm_setup_with_memlock('on-fault')
+
+        status = self.get_process_status_values(self.vm.get_pid())
+
+        # VmLck > 0 kB and only few memory is resident
+        self.assertTrue(status['VmLck'] > 0)
+        self.assertTrue(status['VmRSS'] <= status['VmSize'] * 0.30)
+
+    def get_process_status_values(self, pid: int) -> Dict[str, int]:
+        result = {}
+        raw_status = self._get_raw_process_status(pid)
+
+        for line in raw_status.split('\n'):
+            if m := STATUS_VALUE_PATTERN.match(line):
+                result[m.group(1)] = int(m.group(2))
+
+        return result
+
+    def _get_raw_process_status(self, pid: int) -> str:
+        try:
+            with open(f'/proc/{pid}/status', 'r') as f:
+                return f.read()
+        except FileNotFoundError:
+            self.skipTest("Can't open status file of the process")
+
+
+if __name__ == '__main__':
+    MemlockTest.main()