TaskDistantMixin.py

# ClusterShell (distant) test suite
# Written by S. Thiell

"""Unit test for ClusterShell Task (distant)"""

import pwd
import unittest
import warnings

from TLib import HOSTNAME, make_temp_filename, make_temp_dir
from ClusterShell.Event import EventHandler
from ClusterShell.Task import *
from ClusterShell.Worker.Ssh import WorkerSsh
from ClusterShell.Worker.EngineClient import *
from ClusterShell.Worker.Worker import FANOUT_UNLIMITED, WorkerBadArgumentError

import socket

# TEventHandlerChecker 'received event' flags
EV_START = 0x01
EV_PICKUP = 0x02
EV_READ = 0x04
EV_WRITTEN = 0x08
EV_HUP = 0x10
EV_TIMEOUT = 0x20
EV_CLOSE = 0x40


class TaskDistantMixin(object):

    def setUp(self):
        self._task = task_self()

    def testLocalhostCommand(self):
        # init worker
        worker = self._task.shell("/bin/hostname", nodes=HOSTNAME)
        # run task
        self._task.resume()

    def testLocalhostCommand2(self):
        # init worker
        worker = self._task.shell("/bin/hostname", nodes=HOSTNAME)

        worker = self._task.shell("/bin/uname -r", nodes=HOSTNAME)
        # run task
        self._task.resume()

    def testTaskShellWorkerGetCommand(self):
        worker1 = self._task.shell("/bin/hostname", nodes=HOSTNAME)
        worker2 = self._task.shell("/bin/uname -r", nodes=HOSTNAME)
        self._task.resume()
        self.assertTrue(hasattr(worker1, 'command'))
        self.assertTrue(hasattr(worker2, 'command'))
        self.assertEqual(worker1.command, "/bin/hostname")
        self.assertEqual(worker2.command, "/bin/uname -r")

    def testTaskShellRunDistant(self):
        wrk = task_self().run("false", nodes=HOSTNAME)
        self.assertEqual(wrk.node_retcode(HOSTNAME), 1)

    def testLocalhostCopy(self):
        dests = []
        try:
            for i in range(5):
                dest = make_temp_filename(suffix='LocalhostCopy')
                dests.append(dest)
                worker = self._task.copy("/etc/hosts", dest, nodes=HOSTNAME)
            self._task.resume()
        finally:
            for dest in dests:
                os.unlink(dest)

    def testCopyNodeFailure(self):
        # == stderr merged ==
        self._task.set_default("stderr", False)
        dest = make_temp_filename(suffix='LocalhostCopyF')
        worker = self._task.copy("/etc/hosts", dest,
                                 nodes='unlikely-node,%s' % HOSTNAME)
        self._task.resume()
        self.assertEqual(worker.node_error_buffer("unlikely-node"), None)
        self.assertTrue(len(worker.node_buffer("unlikely-node")) > 2)
        os.unlink(dest)

        # == stderr separated ==
        self._task.set_default("stderr", True)
        try:
            dest = make_temp_filename(suffix='LocalhostCopyF2')
            worker = self._task.copy("/etc/hosts", dest,
                                     nodes='unlikely-node,%s' % HOSTNAME)
            # run task
            self._task.resume()
            self.assertTrue(worker.node_buffer("unlikely-node") is None)
            self.assertTrue(len(worker.node_error_buffer("unlikely-node")) > 2)
            os.unlink(dest)
        finally:
            self._task.set_default("stderr", False)

    def testLocalhostCopyDir(self):
        dtmp_src = make_temp_dir('src')
        dtmp_dst = make_temp_dir('testLocalhostCopyDir')
        try:
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_b"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a", "lev2"))
            worker = self._task.copy(dtmp_src.name, dtmp_dst.name,
                                     nodes=HOSTNAME)
            self._task.resume()
            self.assertTrue(os.path.exists(os.path.join(dtmp_dst.name,
                            os.path.basename(dtmp_src.name), "lev1_a", "lev2")))
        finally:
            dtmp_dst.cleanup()
            dtmp_src.cleanup()

    def testLocalhostExplicitSshCopy(self):
        dest = make_temp_filename('testLocalhostExplicitSshCopy')
        srcsz = os.path.getsize("/etc/hosts")
        try:
            worker = WorkerSsh(HOSTNAME, source="/etc/hosts", dest=dest,
                               handler=None, timeout=10)
            self._task.schedule(worker)
            self._task.resume()
            self.assertEqual(srcsz, os.path.getsize(dest))
        finally:
            os.remove(dest)

    def testLocalhostExplicitSshCopyWithOptions(self):
        dest = make_temp_dir('testLocalhostExplicitSshCopyWithOptions')
        self._task.set_info("scp_path", "/usr/bin/scp -l 10")
        self._task.set_info("scp_options", "-oLogLevel=QUIET")
        try:
            worker = WorkerSsh(HOSTNAME, source="/etc/hosts", dest=dest.name,
                               handler=None)
            self._task.schedule(worker)
            self._task.resume()
            self.assertEqual(self._task.max_retcode(), 0)
            self.assertTrue(os.path.exists(os.path.join(dest.name, "hosts")))
        finally:
            os.unlink(os.path.join(dest.name, "hosts"))
            dest.cleanup()
        # clear options after test
        task_cleanup()
        self.assertEqual(task_self().info("scp_path"), None)

    def testLocalhostExplicitSshCopyDir(self):
        dtmp_src = make_temp_dir('src')
        dtmp_dst = make_temp_dir('testLocalhostExplicitSshCopyDir')
        try:
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_b"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a", "lev2"))
            worker = WorkerSsh(HOSTNAME, source=dtmp_src.name,
                               dest=dtmp_dst.name, handler=None)
            self._task.schedule(worker)
            self._task.resume()
            path = os.path.join(dtmp_dst.name, os.path.basename(dtmp_src.name),
                                "lev1_a", "lev2")
            self.assertTrue(os.path.exists(path))
        finally:
            dtmp_dst.cleanup()
            dtmp_src.cleanup()

    def testLocalhostExplicitSshCopyDirPreserve(self):
        dtmp_src = make_temp_dir('src')
        dtmp_dst = make_temp_dir('testLocalhostExplicitSshCopyDirPreserve')
        try:
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_b"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a", "lev2"))
            worker = WorkerSsh(HOSTNAME, source=dtmp_src.name,
                               dest=dtmp_dst.name, handler=None, timeout=10,
                               preserve=True)
            self._task.schedule(worker)
            self._task.resume()
            self.assertTrue(os.path.exists(os.path.join(dtmp_dst.name,
                            os.path.basename(dtmp_src.name), "lev1_a", "lev2")))
        finally:
            dtmp_dst.cleanup()
            dtmp_src.cleanup()

    def testExplicitSshWorker(self):
        # init worker
        worker = WorkerSsh(HOSTNAME, command="/bin/echo alright", handler=None)
        self._task.schedule(worker)
        # run task
        self._task.resume()
        # test output
        self.assertEqual(worker.node_buffer(HOSTNAME), b"alright")

    def testExplicitSshWorkerWithOptions(self):
        self._task.set_info("ssh_path", "/usr/bin/ssh -C")
        self._task.set_info("ssh_options", "-oLogLevel=QUIET")
        worker = WorkerSsh(HOSTNAME, command="/bin/echo alright", handler=None)
        self._task.schedule(worker)
        # run task
        self._task.resume()
        # test output
        self.assertEqual(worker.node_buffer(HOSTNAME), b"alright")
        # clear options after test
        task_cleanup()
        self.assertEqual(task_self().info("ssh_path"), None)

    def testExplicitSshWorkerStdErr(self):
        # init worker
        worker = WorkerSsh(HOSTNAME, command="/bin/echo alright 1>&2",
                           handler=None, stderr=True)
        self._task.schedule(worker)
        # run task
        self._task.resume()
        # test output
        self.assertEqual(worker.node_error_buffer(HOSTNAME), b"alright")

        # Re-test with stderr=False
        worker = WorkerSsh(HOSTNAME, command="/bin/echo alright 1>&2",
                           handler=None, stderr=False)
        self._task.schedule(worker)
        # run task
        self._task.resume()
        # test output
        self.assertEqual(worker.node_error_buffer(HOSTNAME), None)

    class TEventHandlerChecker(EventHandler):
        """simple event trigger validator"""
        def __init__(self, test):
            self.test = test
            self.flags = 0
            self.read_count = 0
            self.written_count = 0
        def ev_start(self, worker):
            self.test.assertEqual(self.flags, 0)
            self.flags |= EV_START
        def ev_pickup(self, worker, node):
            self.test.assertTrue(self.flags & EV_START)
            self.flags |= EV_PICKUP
            self.last_node = node
        def ev_read(self, worker, node, sname, msg):
            self.test.assertEqual(self.flags, EV_START | EV_PICKUP)
            self.flags |= EV_READ
            self.last_node = node
            self.last_read = msg
        def ev_written(self, worker, node, sname, size):
            self.test.assertTrue(self.flags & (EV_START | EV_PICKUP))
            self.flags |= EV_WRITTEN
        def ev_hup(self, worker, node, rc):
            self.test.assertTrue(self.flags & (EV_START | EV_PICKUP))
            self.flags |= EV_HUP
            self.last_node = node
            self.last_rc = rc
        def ev_close(self, worker, timedout):
            self.test.assertTrue(self.flags & EV_START)
            self.test.assertTrue(self.flags & EV_CLOSE == 0)
            if timedout:
                self.flags |= EV_TIMEOUT
            self.flags |= EV_CLOSE

    def testShellEvents(self):
        # init worker
        test_eh = self.__class__.TEventHandlerChecker(self)
        worker = self._task.shell("/bin/hostname", nodes=HOSTNAME, handler=test_eh)
        # run task
        self._task.resume()
        # test events received: start, read, hup, close
        self.assertEqual(test_eh.flags, EV_START | EV_PICKUP | EV_READ | EV_HUP | EV_CLOSE)

    def testShellEventsWithTimeout(self):
        # init worker
        test_eh = self.__class__.TEventHandlerChecker(self)
        worker = self._task.shell("/bin/echo alright && /bin/sleep 10",
                                  nodes=HOSTNAME, handler=test_eh, timeout=2)
        self.assertTrue(worker != None)
        # run task
        self._task.resume()
        # test events received: start, read, timeout, close
        self.assertEqual(test_eh.flags, EV_START | EV_PICKUP | EV_READ | EV_TIMEOUT | EV_CLOSE)
        self.assertEqual(worker.node_buffer(HOSTNAME), b"alright")
        self.assertEqual(worker.num_timeout(), 1)
        self.assertEqual(self._task.num_timeout(), 1)
        count = 0
        for node in self._task.iter_keys_timeout():
            count += 1
            self.assertEqual(node, HOSTNAME)
        self.assertEqual(count, 1)
        count = 0
        for node in worker.iter_keys_timeout():
            count += 1
            self.assertEqual(node, HOSTNAME)
        self.assertEqual(count, 1)

    def testShellEventsWithTimeout2(self):
        # init worker
        test_eh1 = self.__class__.TEventHandlerChecker(self)
        worker1 = self._task.shell("/bin/echo alright && /bin/sleep 10",
                                   nodes=HOSTNAME, handler=test_eh1, timeout=2)
        test_eh2 = self.__class__.TEventHandlerChecker(self)
        worker2 = self._task.shell("/bin/echo okay && /bin/sleep 10",
                                   nodes=HOSTNAME, handler=test_eh2, timeout=3)
        # run task
        self._task.resume()
        # test events received: start, read, timeout, close
        self.assertEqual(test_eh1.flags, EV_START | EV_PICKUP | EV_READ | EV_TIMEOUT | EV_CLOSE)
        self.assertEqual(test_eh2.flags, EV_START | EV_PICKUP | EV_READ | EV_TIMEOUT | EV_CLOSE)
        self.assertEqual(worker1.node_buffer(HOSTNAME), b"alright")
        self.assertEqual(worker2.node_buffer(HOSTNAME), b"okay")
        self.assertEqual(worker1.num_timeout(), 1)
        self.assertEqual(worker2.num_timeout(), 1)
        self.assertEqual(self._task.num_timeout(), 2)

    def testShellEventsReadNoEOL(self):
        # init worker
        test_eh = self.__class__.TEventHandlerChecker(self)
        worker = self._task.shell("/bin/echo -n okay", nodes=HOSTNAME, handler=test_eh)
        # run task
        self._task.resume()
        # test events received: start, close
        self.assertEqual(test_eh.flags, EV_START | EV_PICKUP | EV_READ | EV_HUP | EV_CLOSE)
        self.assertEqual(worker.node_buffer(HOSTNAME), b"okay")

    def testShellEventsNoReadNoTimeout(self):
        # init worker
        test_eh = self.__class__.TEventHandlerChecker(self)
        worker = self._task.shell("/bin/sleep 2", nodes=HOSTNAME, handler=test_eh)
        # run task
        self._task.resume()
        # test events received: start, close
        self.assertEqual(test_eh.flags, EV_START | EV_PICKUP | EV_HUP | EV_CLOSE)
        self.assertEqual(worker.node_buffer(HOSTNAME), None)

    def testLocalhostCommandFanout(self):
        fanout = self._task.info("fanout")
        self._task.set_info("fanout", 2)
        # init worker
        for i in range(0, 10):
            worker = self._task.shell("/bin/echo %d" % i, nodes=HOSTNAME)
        # run task
        self._task.resume()
        # restore fanout value
        self._task.set_info("fanout", fanout)

    def testWorkerBuffers(self):
        # Warning: if you modify this test, please also modify testWorkerErrorBuffers()
        task = task_self()
        worker = task.shell("/usr/bin/printf 'foo\nbar\nxxx\n'",
                            nodes=HOSTNAME)
        task.resume()
        # test iter_buffers() by worker...
        cnt = 2
        for buf, nodes in worker.iter_buffers():
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(len(nodes), 1)
                self.assertEqual(str(nodes), HOSTNAME)
        self.assertEqual(cnt, 1)
        # new check in 1.7 to ensure match_keys is not a string
        testgen = worker.iter_buffers(HOSTNAME)
        # cast to list to effectively iterate
        self.assertRaises(TypeError, list, testgen)
        # and also fixed an issue when match_keys was an empty list
        for buf, nodes in worker.iter_buffers([]):
            self.assertFalse("Found buffer with empty match_keys?!")
        for buf, nodes in worker.iter_buffers([HOSTNAME]):
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(len(nodes), 1)
                self.assertEqual(str(nodes), HOSTNAME)
        self.assertEqual(cnt, 0)
        # test flushing buffers by worker
        worker.flush_buffers()
        self.assertEqual(list(worker.iter_buffers()), [])

    def testWorkerErrorBuffers(self):
        # Warning: if you modify this test, please also modify testWorkerBuffers()
        task = task_self()
        worker = task.shell("/usr/bin/printf 'foo\nbar\nxxx\n' 1>&2",
                            nodes=HOSTNAME, stderr=True)
        task.resume()
        # test iter_errors() by worker...
        cnt = 2
        for buf, nodes in worker.iter_errors():
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(len(nodes), 1)
                self.assertEqual(str(nodes), HOSTNAME)
        self.assertEqual(cnt, 1)
        # new check in 1.7 to ensure match_keys is not a string
        testgen = worker.iter_errors(HOSTNAME)
        # cast to list to effectively iterate
        self.assertRaises(TypeError, list, testgen)
        # and also fixed an issue when match_keys was an empty list
        for buf, nodes in worker.iter_errors([]):
            self.assertFalse("Found error buffer with empty match_keys?!")
        for buf, nodes in worker.iter_errors([HOSTNAME]):
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(len(nodes), 1)
                self.assertEqual(str(nodes), HOSTNAME)
        self.assertEqual(cnt, 0)
        # test flushing error buffers by worker
        worker.flush_errors()
        self.assertEqual(list(worker.iter_errors()), [])

    def testWorkerNodeBuffers(self):
        task = task_self()

        worker = task.shell("/usr/bin/printf 'foo\nbar\nxxx\n'",
                            nodes=HOSTNAME)

        task.resume()

        cnt = 1
        for node, buf in worker.iter_node_buffers():
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(node, HOSTNAME)
        self.assertEqual(cnt, 0)

    def testWorkerNodeErrors(self):
        task = task_self()

        worker = task.shell("/usr/bin/printf 'foo\nbar\nxxx\n' 1>&2",
                            nodes=HOSTNAME, stderr=True)

        task.resume()

        cnt = 1
        for node, buf in worker.iter_node_errors():
            cnt -= 1
            if buf == b"foo\nbar\nxxx\n":
                self.assertEqual(node, HOSTNAME)
        self.assertEqual(cnt, 0)

    def testWorkerRetcodes(self):
        task = task_self()

        worker = task.shell("/bin/sh -c 'exit 3'", nodes=HOSTNAME)

        task.resume()

        cnt = 2
        for rc, keys in worker.iter_retcodes():
            cnt -= 1
            self.assertEqual(rc, 3)
            self.assertEqual(len(keys), 1)
            self.assertEqual(keys[0], HOSTNAME)

        self.assertEqual(cnt, 1)

        for rc, keys in worker.iter_retcodes(HOSTNAME):
            cnt -= 1
            self.assertEqual(rc, 3)
            self.assertEqual(len(keys), 1)
            self.assertEqual(keys[0], HOSTNAME)

        self.assertEqual(cnt, 0)

        # test node_retcode
        self.assertEqual(worker.node_retcode(HOSTNAME), 3)   # 1.2.91+
        self.assertEqual(worker.node_rc(HOSTNAME), 3)

        # test node_retcode failure
        self.assertRaises(KeyError, worker.node_retcode, "dummy")

        # test max retcode API
        self.assertEqual(task.max_retcode(), 3)

    def testWorkerNodeRetcodes(self):
        task = task_self()

        worker = task.shell("/bin/sh -c 'exit 3'", nodes=HOSTNAME)

        task.resume()

        cnt = 1
        for node, rc in worker.iter_node_retcodes():
            cnt -= 1
            self.assertEqual(rc, 3)
            self.assertEqual(node, HOSTNAME)

        self.assertEqual(cnt, 0)

    def testEscape(self):
        cmd = r"export CSTEST=foobar; /bin/echo \$CSTEST | sed 's/\ foo/bar/'"
        worker = self._task.shell(cmd, nodes=HOSTNAME)
        # execute
        self._task.resume()
        # read result
        self.assertEqual(worker.node_buffer(HOSTNAME), b"$CSTEST")

    def testEscape2(self):
        cmd = r"export CSTEST=foobar; /bin/echo $CSTEST | sed 's/\ foo/bar/'"
        worker = self._task.shell(cmd, nodes=HOSTNAME)
        # execute
        self._task.resume()
        # read result
        self.assertEqual(worker.node_buffer(HOSTNAME), b"foobar")

    def testSshUserOption(self):
        ssh_user_orig = self._task.info("ssh_user")
        self._task.set_info("ssh_user", pwd.getpwuid(os.getuid())[0])
        worker = self._task.shell("/bin/echo foobar", nodes=HOSTNAME)
        self._task.resume()
        # restore original ssh_user (None)
        self.assertEqual(ssh_user_orig, None)
        self._task.set_info("ssh_user", ssh_user_orig)

    def testSshUserOptionForScp(self):
        ssh_user_orig = self._task.info("ssh_user")
        self._task.set_info("ssh_user", pwd.getpwuid(os.getuid())[0])
        dest = make_temp_filename('testLocalhostCopyU')
        worker = self._task.copy("/etc/hosts", dest, nodes=HOSTNAME)
        self._task.resume()
        # restore original ssh_user (None)
        self.assertEqual(ssh_user_orig, None)
        self._task.set_info("ssh_user", ssh_user_orig)
        os.unlink(dest)

    def testSshOptionsOption(self):
        ssh_options_orig = self._task.info("ssh_options")
        try:
            self._task.set_info("ssh_options", "-oLogLevel=QUIET")
            worker = self._task.shell("/bin/echo foobar", nodes=HOSTNAME)
            self._task.resume()
            self.assertEqual(worker.node_buffer(HOSTNAME), b"foobar")
            # test 3 options
            self._task.set_info("ssh_options", \
                "-oLogLevel=QUIET -oStrictHostKeyChecking=no -oVerifyHostKeyDNS=no")
            worker = self._task.shell("/bin/echo foobar3", nodes=HOSTNAME)
            self._task.resume()
            self.assertEqual(worker.node_buffer(HOSTNAME), b"foobar3")
        finally:
            # restore original ssh_user (None)
            self.assertEqual(ssh_options_orig, None)
            self._task.set_info("ssh_options", ssh_options_orig)

    def testSshOptionsOptionForScp(self):
        ssh_options_orig = self._task.info("ssh_options")
        testfile = None
        try:
            testfile = make_temp_filename('testLocalhostCopyO')
            if os.path.exists(testfile):
                os.remove(testfile)
            self._task.set_info("ssh_options", \
                "-oLogLevel=QUIET -oStrictHostKeyChecking=no -oVerifyHostKeyDNS=no")
            worker = self._task.copy("/etc/hosts", testfile, nodes=HOSTNAME)
            self._task.resume()
            self.assertTrue(os.path.exists(testfile))
        finally:
            os.unlink(testfile)
            # restore original ssh_user (None)
            self.assertEqual(ssh_options_orig, None)
            self._task.set_info("ssh_options", ssh_options_orig)

    def testShellStderrWithHandler(self):
        class StdErrHandler(EventHandler):
            def ev_read(self, worker, node, sname, msg):
                if sname == worker.SNAME_STDERR:
                    assert msg == b"something wrong"

        worker = self._task.shell("echo something wrong 1>&2", nodes=HOSTNAME,
                                  handler=StdErrHandler(), stderr=True)
        self._task.resume()
        for buf, nodes in worker.iter_errors():
            self.assertEqual(buf, b"something wrong")
        for buf, nodes in worker.iter_errors([HOSTNAME]):
            self.assertEqual(buf, b"something wrong")

    def testShellWriteSimple(self):
        worker = self._task.shell("cat", nodes=HOSTNAME)
        worker.write(b"this is a test\n")
        worker.set_write_eof()
        self._task.resume()
        self.assertEqual(worker.node_buffer(HOSTNAME), b"this is a test")

    def testShellWriteHandler(self):
        class WriteOnReadHandler(EventHandler):
            def __init__(self, target_worker):
                self.target_worker = target_worker
            def ev_read(self, worker, node, sname, msg):
                self.target_worker.write(node.encode() + b':' + msg + b'\n')
                self.target_worker.set_write_eof()

        reader = self._task.shell("cat", nodes=HOSTNAME)
        worker = self._task.shell("sleep 1; echo foobar", nodes=HOSTNAME,
                                  handler=WriteOnReadHandler(reader))
        self._task.resume()
        res = "%s:foobar" % HOSTNAME
        self.assertEqual(reader.node_buffer(HOSTNAME), res.encode())

    def testSshBadArgumentOption(self):
        # Check code < 1.4 compatibility
        self.assertRaises(WorkerBadArgumentError, WorkerSsh, HOSTNAME, None,
                          None)
        # As of 1.4, ValueError is raised for missing parameter
        self.assertRaises(ValueError, WorkerSsh, HOSTNAME, None, None) # 1.4+

    def testCopyEvents(self):
        test_eh = self.__class__.TEventHandlerChecker(self)
        dest = make_temp_filename('testLocalhostCopyEvents')
        worker = self._task.copy("/etc/hosts", dest, nodes=HOSTNAME,
                                 handler=test_eh)
        # run task
        self._task.resume()
        os.unlink(dest)
        self.assertEqual(test_eh.flags, EV_START | EV_PICKUP | EV_HUP | EV_CLOSE)

    def testWorkerAbort(self):
        task = task_self()

        # Test worker.abort() in an event handler.
        class AbortOnTimer(EventHandler):
            def __init__(self, worker):
                EventHandler.__init__(self)
                self.ext_worker = worker
                self.testtimer = False
            def ev_timer(self, timer):
                self.ext_worker.abort()
                self.ext_worker.abort()  # safe but no effect
                self.testtimer = True

        aot = AbortOnTimer(task.shell("sleep 10", nodes=HOSTNAME))
        self.assertEqual(aot.testtimer, False)
        task.timer(1.5, handler=aot)
        task.resume()
        self.assertEqual(aot.testtimer, True)

    def testWorkerAbortSanity(self):
        task = task_self()
        worker = task.shell("sleep 1", nodes=HOSTNAME)
        worker.abort()

        # test noop abort() on unscheduled worker
        worker = WorkerSsh(HOSTNAME, command="sleep 1", handler=None,
                           timeout=None)
        worker.abort()

    def testLocalhostRCopy(self):
        try:
            dest = make_temp_dir('testLocalhostRCopy')
            # use fake node 'aaa' to test rank > 0
            worker = self._task.rcopy("/etc/hosts", dest.name, "aaa,%s" % HOSTNAME,
                                      handler=None, timeout=10)
            self._task.resume()
            self.assertEqual(worker.source, "/etc/hosts")
            self.assertEqual(worker.dest, dest.name)
            self.assertTrue(os.path.exists(os.path.join(dest.name,
                                                        "hosts.%s" % HOSTNAME)))
        finally:
            dest.cleanup()

    def testLocalhostExplicitSshReverseCopy(self):
        dest = make_temp_dir('testLocalhostExplicitSshRCopy')
        try:
            worker = WorkerSsh(HOSTNAME, source="/etc/hosts", dest=dest.name,
                               handler=None, timeout=10, reverse=True)
            self._task.schedule(worker)
            self._task.resume()
            self.assertEqual(worker.source, "/etc/hosts")
            self.assertEqual(worker.dest, dest.name)
            self.assertTrue(os.path.exists(os.path.join(dest.name,
                                                        "hosts.%s" % HOSTNAME)))
        finally:
            dest.cleanup()

    def testLocalhostExplicitSshReverseCopyDir(self):
        dtmp_src = make_temp_dir('src')
        dtmp_dst = make_temp_dir('testLocalhostExplicitSshReverseCopyDir')
        try:
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_b"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a", "lev2"))
            worker = WorkerSsh(HOSTNAME, source=dtmp_src.name,
                               dest=dtmp_dst.name, handler=None, timeout=30,
                               reverse=True)
            self._task.schedule(worker)
            self._task.resume()
            self.assertTrue(os.path.exists(os.path.join(dtmp_dst.name, "%s.%s" % \
                    (os.path.basename(dtmp_src.name), HOSTNAME), "lev1_a", "lev2")))
        finally:
            dtmp_dst.cleanup()
            dtmp_src.cleanup()

    def testLocalhostExplicitSshReverseCopyDirPreserve(self):
        dtmp_src = make_temp_dir('src')
        dtmp_dst = make_temp_dir('testLocalhostExplicitSshReverseCpDirPreserve')
        try:
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_b"))
            os.mkdir(os.path.join(dtmp_src.name, "lev1_a", "lev2"))
            worker = WorkerSsh(HOSTNAME, source=dtmp_src.name,
                               dest=dtmp_dst.name, handler=None, timeout=30,
                               reverse=True)
            self._task.schedule(worker)
            self._task.resume()
            self.assertTrue(os.path.exists(os.path.join(dtmp_dst.name, "%s.%s" % \
                    (os.path.basename(dtmp_src.name), HOSTNAME), "lev1_a", "lev2")))
        finally:
            dtmp_dst.cleanup()
            dtmp_src.cleanup()

    def testErroneousSshPath(self):
        try:
            self._task.set_info("ssh_path", "/wrong/path/to/ssh")
            # init worker
            worker = self._task.shell("/bin/echo ok", nodes=HOSTNAME)
            # run task
            self._task.resume()
            self.assertEqual(self._task.max_retcode(), 255)
        finally:
            # restore fanout value
            self._task.set_info("ssh_path", None)

    class TEventHandlerEvCountChecker(EventHandler):
        """simple event count validator"""

        def __init__(self):
            self.start_count = 0
            self.pickup_count = 0
            self.hup_count = 0
            self.close_count = 0

        def ev_start(self, worker):
            self.start_count += 1

        def ev_pickup(self, worker, node):
            self.pickup_count += 1

        def ev_hup(self, worker, node, rc):
            self.hup_count += 1

        def ev_close(self, worker, timedout):
            self.close_count += 1

    @unittest.skipIf(HOSTNAME == 'localhost', "does not work with hostname set to 'localhost'")
    def testWorkerEventCount(self):
        test_eh = self.__class__.TEventHandlerEvCountChecker()
        nodes = "localhost,%s" % HOSTNAME
        worker = self._task.shell("/bin/hostname", nodes=nodes, handler=test_eh)
        self._task.resume()
        # test event count
        self.assertEqual(test_eh.pickup_count, 2)
        self.assertEqual(test_eh.hup_count, 2)
        self.assertEqual(test_eh.start_count, 1)
        self.assertEqual(test_eh.close_count, 1)