--- /dev/null
+# 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
extra_args=['-U"$USERNAME%$PASSWORD"'],
name="samba.tests.prefork_restart",
py3_compatible=True)
+planoldpythontestsuite("preforkrestartdc:local",
+ "samba.tests.blackbox.smbcontrol_process",
+ extra_path=[
+ os.path.join(srcdir(), 'python/samba/tests')],
+ extra_args=['-U"$USERNAME%$PASSWORD"'],
+ name="samba.tests.blackbox.smbcontrol_process",
+ py3_compatible=True)
planoldpythontestsuite("proclimitdc:local",
"samba.tests.process_limits",
extra_path=[