Attachment #5949 for bug #34593

View | Details | Raw Unified | Return to bug 34593
Collapse All | Expand All




            print("Created process with pid '%s' was not terminated, "
                  "forcing kill" % self.Proc.pid)
            self.Proc.kill()
            _pid, exit_status, _res_usage = wait4(self.Proc.pid, WNOHANG)
            if exit_status != 9:
                print("The exit status while force kill is '%s' "
                      "instead of 'SIGKILL' code 9." % exit_status)

            pid = self.create_process()
            if self.query_process_exists(pid):
                self.make_kill_request('SIGTERM', [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if exit_status != 15:
                    utils.fail("Process exit status is '%s' instead of "
                               "'SIGTERM' code 15." % exit_status)

            pid = self.create_process()
            if self.query_process_exists(pid):
                self.make_kill_request('SIGKILL', [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if exit_status != 9:
                    utils.fail("Process exit status is '%s' instead of "
                               "'SIGKILL' code 9." % exit_status)
- 
--
.../test/ucs-test/tests/60_umc-system/22_umc-service-proc-killing   | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)




                    if 'python' in result['command']:
                        return True
            except KeyError as exc:
                utils.fail("Failed to find key '%s' in"
                           "the 'top/query' request respone from UMC: %r"
                           % (exc, result))

    def force_process_kill(self):
        """
- 
--
.../60_umc-system/22_umc-service-proc-killing      | 22 +++++++++++-----------
1 file changed, 11 insertions(+), 11 deletions(-)




from TestUMCSystemModule import TestUMCSystem

from psutil import Process
from os import wait4, WNOHANG, WTERMSIG
import signal

from univention.config_registry import ConfigRegistry
import univention.testing.utils as utils

                  "forcing kill" % self.Proc.pid)
            self.Proc.kill()
            _pid, exit_status, _res_usage = wait4(self.Proc.pid, WNOHANG)
            if WTERMSIG(exit_status) != signal.SIGKILL:
                print("The exit status while force kill is 0x%x "
                      "instead of 'SIGKILL'(%d)." % (exit_status, signal.SIGKILL))

    def create_process(self):
        """

            if self.query_process_exists(pid):
                self.make_kill_request('SIGTERM', [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if WTERMSIG(exit_status) != signal.SIGTERM:
                    utils.fail("Process exit status is 0x%x instead of "
                               "SIGTERM(%d)." % (exit_status, signal.SIGTERM))
                if self.query_process_exists(pid):
                    utils.fail("Process did not terminate after request "
                               "with 'SIGTERM' signal")

            if self.query_process_exists(pid):
                self.make_kill_request('SIGKILL', [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if WTERMSIG(exit_status) != signal.SIGKILL:
                    utils.fail("Process exit status is 0x%x instead of "
                               "SIGKILL(%d)." % (exit_status, signal.SIGKILL))
                if self.query_process_exists(pid):
                    utils.fail("Process did not terminate after request "
                               "with 'SIGKILL' signal")
- 
--
.../60_umc-system/22_umc-service-proc-killing      | 49 +++++++++-------------
1 file changed, 19 insertions(+), 30 deletions(-)




        self.Proc = Process(pid)
        return pid

    def test(self, signame):
        print "Testing UMC process killing with signal '%s'" % signame
        signum = getattr(signal, signame)
        process killing functionality.
        """
        self.get_ucr_credentials()
        self.create_connection_authenticate()

        # case 1: killing process using signal 'SIGTERM'
        try:
            print "Testing UMC process killing with 'SIGTERM' signal"
            pid = self.create_process()
            if self.query_process_exists(pid):
                self.make_kill_request(signame, [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if WTERMSIG(exit_status) != signum:
                    utils.fail("Process exit status is 0x%x instead of "
                               "%s(%d)." % (exit_status, signame, signum))
                if self.query_process_exists(pid):
                    utils.fail("Process did not terminate after request "
                               "with signal %s" % (signame))
            else:
                utils.fail("The process is not running right after creation")
        finally:
            self.force_process_kill()

    def main(self):
        """
        Method to start the test of the UMC backend
        process killing functionality.
        """
        self.get_ucr_credentials()
        self.create_connection_authenticate()

        # case 1: killing process using signal 'SIGTERM'
        self.test('SIGTERM')

        # case 2: killing process using signal 'SIGKILL'
        self.test('SIGKILL')
            print "Testing UMC process killing with 'SIGKILL' signal"
            pid = self.create_process()
            if self.query_process_exists(pid):
                self.make_kill_request('SIGKILL', [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if WTERMSIG(exit_status) != signal.SIGKILL:
                    utils.fail("Process exit status is 0x%x instead of "
                               "SIGKILL(%d)." % (exit_status, signal.SIGKILL))
                if self.query_process_exists(pid):
                    utils.fail("Process did not terminate after request "
                               "with 'SIGKILL' signal")
            else:
                utils.fail("The process is not running right after creation")
        finally:
            self.force_process_kill()


if __name__ == '__main__':
- 
--
.../60_umc-system/22_umc-service-proc-killing      | 22 +++++++++++++---------
1 file changed, 13 insertions(+), 9 deletions(-)




from TestUMCSystemModule import TestUMCSystem

from psutil import Process
from os import wait4, WNOHANG, WTERMSIG, fork
import signal
from time import sleep

from univention.config_registry import ConfigRegistry
import univention.testing.utils as utils

        for result in request_result:
            try:
                if result['pid'] == pid:
                    if sys.executable in result['command']:
                        return True
            except KeyError as exc:
                utils.fail("Failed to find key '%s' in"

                print("The exit status while force kill is 0x%x "
                      "instead of 'SIGKILL'(%d)." % (exit_status, signal.SIGKILL))

    def create_process(self, ignore_sigterm=False):
        """
        Initiates a simple test process that should be killed after.
        Creates a psutil Process class to check running state
        before terminating. Also returns process id (pid).
        """
        pid = fork()
        if pid:  # parent
            self.Proc = Process(pid)
            return pid
        else:  # child under test
            if ignore_sigterm:
                signal.signal(signal.SIGTERM, signal.SIG_IGN)
            sleep(30)
            sys.exit(0)

    def test(self, signame):
        print "Testing UMC process killing with signal '%s'" % signame
- 
--
.../ucs-test/tests/60_umc-system/22_umc-service-proc-killing | 12 ++++++++----
1 file changed, 8 insertions(+), 4 deletions(-)




                utils.fail("Failed to find key '%s' in"
                           "the 'top/query' request respone from UMC: %r"
                           % (exc, result))
        return False

    def force_process_kill(self):
        """

            sleep(30)
            sys.exit(0)

    def test(self, signame, ignore_signal=False):
        print "Testing UMC process killing with signal '%s'" % signame
        signum = getattr(signal, signame)
        try:
            pid = self.create_process(ignore_signal)
            if self.query_process_exists(pid):
                self.make_kill_request(signame, [pid])
                _pid, exit_status, _res_usage = wait4(pid, WNOHANG)
                if ignore_signal != (WTERMSIG(exit_status) != signum):
                    utils.fail("Process exit status is 0x%x instead of "
                               "%s(%d)." % (exit_status, signame, signum))
                if ignore_signal != self.query_process_exists(pid):
                    utils.fail("Process did not terminate after request "
                               "with signal %s" % (signame))
            else:

        # case 2: killing process using signal 'SIGKILL'
        self.test('SIGKILL')

        # case 3: killing unwilling process using signal 'SIGTERM'
        self.test('SIGTERM', ignore_signal=True)


if __name__ == '__main__':
    TestUMC = TestUMCProcessKilling()
- 

Return to bug 34593