Synnefo

import __main__

import prctl

import signal

from multiprocessing import Process, Queue

API = None

TOKEN = None

DEFAULT_API = "http://dev67.dev.grnet.gr:8000/api/v1.1"

DEFAULT_TOKEN = "46e427d657b20defe352804f0eb6f8a2"

TEST_RUN_ID = datetime.datetime.strftime(datetime.datetime.now(),

                                         "%Y%m%d%H%M%S")

SNF_TEST_PREFIX = "snf-test-"

        if os in ("ubuntu", "kubuntu", "fedora"):

            return "user"

        elif os == "windows":

            return "Administrator"

            return "root"

        if server["status"] not in (current_status, new_status):

            return None  # Do not raise exception, return so the test fails

                            "operation `%s' timed out" % opmsg)

        familystr = {socket.AF_INET: "IPv4", socket.AF_INET6: "IPv6",

                     socket.AF_UNSPEC: "Unspecified-IPv4/6"}

        # Ensure the status is actually the expected one

        server = self.client.get_server_details(self.serverid)

        self.assertEquals(server["status"], new_status)

        self.assertNotIn(self.serverid, [s["id"] for s in servers])

class TestRunnerProcess(Process):

    """A distinct process used to execute part of the tests in parallel"""

    def __init__(self, **kw):

        Process.__init__(self, **kw)

        kwargs = kw["kwargs"]

        self.testq = kwargs["testq"]

        self.runner = kwargs["runner"]

    def run(self):

        # Make sure this test runner process dies with the parent

        # and is not left behind.

        #

        # WARNING: This uses the prctl(2) call and is

        # Linux-specific.

        prctl.set_pdeathsig(signal.SIGHUP)

        while True:

            log.debug("I am process %d, GETting from queue is %s",

                     os.getpid(), self.testq)

            msg = self.testq.get()

            log.debug("Dequeued msg: %s", msg)

            if msg == "TEST_RUNNER_TERMINATE":

                raise SystemExit

            elif issubclass(msg, unittest.TestCase):

                # Assemble a TestSuite, and run it

                suite = unittest.TestLoader().loadTestsFromTestCase(msg)

                self.runner.run(suite)

            else:

                raise Exception("Cannot handle msg: %s" % msg)

def _run_cases_in_parallel(cases, fanout=1, runner=None):

    """Run instances of TestCase in parallel, in a number of distinct processes

    The cases iterable specifies the TestCases to be executed in parallel,

    by test runners running in distinct processes.

    The fanout parameter specifies the number of processes to spawn,

    and defaults to 1.

    The runner argument specifies the test runner class to use inside each

    runner process.

    """

    if runner is None:

        runner = unittest.TextTestRunner()

    # testq: The master process enqueues TestCase objects into this queue,

    #        test runner processes pick them up for execution, in parallel.

    testq = Queue()

    runners = []

    for i in xrange(0, fanout):

        kwargs = dict(testq=testq, runner=runner)

        runners.append(TestRunnerProcess(kwargs=kwargs))

    log.info("Spawning %d test runner processes", len(runners))

    for p in runners:

        p.start()

    log.debug("Spawned %d test runners, PIDs are %s",

              len(runners), [p.pid for p in runners])

    # Enqueue test cases

    map(testq.put, cases)

    map(testq.put, ["TEST_RUNNER_TERMINATE"] * len(runners))

    log.debug("Joining %d processes", len(runners))

    for p in runners:

        p.join()

    log.debug("Done joining %d processes", len(runners))

    name = "SpawnServerTestCase_%d" % kwargs["imageid"]

    setattr(__main__,name,cls)

          "\n    ".join(["%d: %s" % (s["id"], s["name"]) for s in stale])

            c.delete_server(server["id"])

    parser.add_option("--api",

                      action="store", type="string", dest="api",

                      help="The API URI to use to reach the Synnefo API",

                      default=DEFAULT_API)

    parser.add_option("--token",

                      action="store", type="string", dest="token",

                      help="The token to use for authentication to the API",

                      default=DEFAULT_TOKEN)

    parser.add_option("--fanout",

                      action="store", type="int", dest="fanout",

                      help="Spawn up to COUNT child processes to execute " \

                           "in parallel, essentially have up to COUNT " \

                           "server build requests outstanding",

                           "name starts with `%s'" % SNF_TEST_PREFIX,

                           "name starts with `%s'" % SNF_TEST_PREFIX,

    corresponding TestCase class (shared fixtures). Distinct subclasses of

    TestCase MAY SHARE NO DATA, since they are run in parallel, in distinct

    test runner processes.

    global API, TOKEN

    API = opts.api

    TOKEN = opts.token

    # FIXME: logging, log, LOG PID, TEST_RUN_ID, arguments

    # FIXME: Network testing? Create, destroy, connect, ping, disconnect VMs?

    runner = unittest.TextTestRunner(verbosity=2, failfast=opts.failfast)

    # The following cases run sequentially

    seq_cases = [UnauthorizedTestCase, FlavorsTestCase, ImagesTestCase]

    _run_cases_in_parallel(seq_cases, fanout=3, runner=runner)

    # The following cases run in parallel

    par_cases = []

    for image in DIMAGES:

        imageid = image["id"]

        imagename = image["name"]

        if opts.force_flavorid:

            flavorid = opts.force_flavorid

        else:

            flavorid = choice([f["id"] for f in DFLAVORS if f["disk"] >= 20])

        personality = None   # FIXME

        servername = "%s%s for %s" % (SNF_TEST_PREFIX, TEST_RUN_ID, imagename)

        is_windows = imagename.lower().find("windows") >= 0

        case = _spawn_server_test_case(imageid=imageid, flavorid=flavorid,

                                       imagename=imagename,

                                       personality=personality,

                                       servername=servername,

                                       is_windows=is_windows,

                                       action_timeout=opts.action_timeout,

                                       build_warning=opts.build_warning,

                                       build_fail=opts.build_fail,

                                       query_interval=opts.query_interval)

        par_cases.append(case)

    print "%s" % FlavorsTestCase

    print "dict", __main__.__dict__

    _run_cases_in_parallel(par_cases, fanout=opts.fanout, runner=runner)