s4 messaging tests: Add inject fault command

[martins/samba.git] / python / samba / tests / blackbox / smbcontrol_process.py

diff --git a/python/samba/tests/blackbox/smbcontrol_process.py b/python/samba/tests/blackbox/smbcontrol_process.py
new file mode 100644 (file)
index 0000000..342bace
--- /dev/null
+++ b/python/samba/tests/blackbox/smbcontrol_process.py
@@ -0,0 +1,85 @@
+# Blackbox tests for the smbcontrol fault injection commands
+#
+# Copyright (C) Andrew Bartlett <abartlet@samba.org> 2018
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# As the test terminates and sleeps samba processes these tests need to run
+# in the preforkrestartdc test environment to prevent them impacting other
+# tests.
+#
+from __future__ import print_function
+from samba.tests import BlackboxTestCase, BlackboxProcessError
+from samba.messaging import Messaging
+
+COMMAND = "bin/smbcontrol"
+PING = "ping"
+
+
+class SmbcontrolProcessBlockboxTests(BlackboxTestCase):
+
+    def setUp(self):
+        super(SmbcontrolProcessBlockboxTests, self).setUp()
+        lp_ctx = self.get_loadparm()
+        self.msg_ctx = Messaging(lp_ctx=lp_ctx)
+
+    def get_process_data(self):
+        services = self.msg_ctx.irpc_all_servers()
+
+        processes = []
+        for service in services:
+            for id in service.ids:
+                processes.append((service.name, id.pid))
+        return processes
+
+    def get_process(self, name):
+        processes = self.get_process_data()
+        for pname, pid in processes:
+            if name == pname:
+                return pid
+        return None
+
+    def test_inject_fault(self):
+        INJECT = "inject"
+        FAULT = "segv"
+        pid = self.get_process("rpc_server")
+
+        #
+        # Ensure we can ping the process before injecting a fault.
+        #
+        try:
+            self.check_run("%s %s %s" % (COMMAND, pid, PING),
+                           msg="trying to ping rpc_server")
+        except BlackboxProcessError as e:
+            self.fail("Unable to ping rpc_server process")
+
+        #
+        # Now inject a fault.
+        #
+        try:
+            self.check_run("%s %s %s %s" % (COMMAND, pid, INJECT, FAULT),
+                           msg="injecting fault into rpc_server")
+        except BlackboxProcessError as e:
+            print(e)
+            self.fail("Unable to inject a fault into the rpc_server process")
+
+        #
+        # The process should have died, so we should not be able to ping it
+        #
+        try:
+            self.check_run("%s %s %s" % (COMMAND, pid, PING),
+                           msg="trying to ping rpc_server")
+            self.fail("Could ping rpc_server process")
+        except BlackboxProcessError as e:
+            pass