"""Parallel testing, supporting arbitrary collection ordering

The Workflow
------------

- Master py.test process starts up, inspects config to decide how many slave to start, if at all

  - env['parallel_base_urls'] is inspected first
  - py.test config.option.appliances and the related --appliance cmdline flag are used
    if env['parallel
_base_urls'] isn't set
  - if neither are set, no parallelization happens

- Slaves are started
- Master runs collection, blocks until slaves report their collections
- Slaves each run collection and submit them to the master, then block inside their runtest loop,
  waiting for tests to run
- Master
 diffs slave collections against its own; the test ids are verified to match
  across all nodes
- Master enters main runtest loop, uses a generator to build lists of test groups which are then
  sent to slaves, one group at a time
- For each phase of each test, the slave serializes test reports, which are then unserialized on
  the master and handed to the normal pytest reporting hooks, which is able to deal with test
  reports arriving out of order
- Before running the last test in a group, the slave will request more tests from the master

  - If more tests are received, they are run
  - If no tests are received, the slave will shut down after running its final test

- After all slaves are shut down, the master will do its end-of-session reporting as usual, and
  shut down

"""

import collections
import difflib
import json
import os
import signal
import subprocess
from collections import OrderedDict, defaultdict, deque, namedtuple
from datetime import datetime
from itertools import count
from threading import Lock, RLock, Thread, Timer
from time import sleep, time
from urlparse import urlparse

import pytest
import zmq
from _pytest import runner
from functools32 import wraps

from fixtures import terminalreporter
from fixtures.parallelizer import remote
from fixtures.pytest_store import store
from utils import at_exit, conf
from utils.appliance import IPAppliance, stack as appliance_stack
from utils.log import create_sublogger
from utils.net import random_port
from utils.path import conf_path, project_path
from utils.sprout import SproutClient, SproutException
from utils.wait import wait_for


_appliance_help = '''specify appliance URLs to use for distributed testing.
this option can be specified more than once, and must be specified at least two times'''

env_base_urls = conf.env.get('parallel_base_urls', [])
if env_base_urls:
    conf.runtime['env']['base_url'] = env_base_urls[0]

# Initialize slaveid to None, indicating this as the master process
# slaves will set this to a unique string when they're initialized
conf.runtime['env']['slaveid'] = None

# lock for protecting mutation of recv queue
recv_lock = Lock()
# lock for protecting zmq socket access
zmq_lock = Lock()


def pytest_addoption(parser):
    group = parser.getgroup("cfme")
    group._addoption('--appliance', dest='appliances', action='append',
        default=env_base_urls, metavar='base_url', help=_appliance_help)
    group._addoption('--use-sprout', dest='use_sprout', action='store_true',
        default=False, help="Use Sprout for provisioning appliances.")
    group._addoption('--sprout-appliances', dest='sprout_appliances', type=int,
        default=1, help="How many Sprout appliances to use?.")
    group._addoption('--sprout-timeout', dest='sprout_timeout', type=int,
        default=60, help="How many minutes is the lease timeout.")
    group._addoption('--sprout-provision-timeout', dest='sprout_provision_timeout', type=int,
        default=60, help="How many minutes to wait for appliances provisioned.")
    group._addoption(
        '--sprout-group', dest='sprout_group', default=None, help="Which stream to use.")
    group._addoption(
        '--sprout-version', dest='sprout_version', default=None, help="Which version to use.")
    group._addoption(
        '--sprout-date', dest='sprout_date', default=None, help="Which date to use.")
    group._addoption(
        '--sprout-desc', dest='sprout_desc', default=None, help="Set description of the pool.")


def pytest_addhooks(pluginmanager):
    import hooks
    pluginmanager.addhooks(hooks)


@pytest.mark.hookwrapper
def pytest_configure(config):
    # configures the parallel session, then fires pytest_parallel_configured
    yield
    if (config.option.appliances or (config.option.use_sprout and
            config.option.sprout_appliances > 1)):
        session = ParallelSession(config)
        config.pluginmanager.register(session, "parallel_session")
        store.parallelizer_role = 'master'
        config.hook.pytest_parallel_configured(parallel_session=session)
    else:
        config.hook.pytest_parallel_configured(parallel_session=None)


def dump_pool_info(printf, pool_data):
    printf("Fulfilled: {}".format(pool_data["fulfilled"]))
    printf("Progress: {}%".format(pool_data["progress"]))
    printf("Appliances:")
    for appliance in pool_data["appliances"]:
        name = appliance.pop("name")
        printf("\t{}:".format(name))
        for key in sorted(appliance.keys()):
            printf("\t\t{}: {}".format(key, appliance[key]))


def handle_end_session(signal, frame):
    # when signaled, end the current test session immediately
    if store.parallel_session:
        store.parallel_session.session_finished = True
signal.signal(signal.SIGQUIT, handle_end_session)


class SlaveDict(dict):
    """A normal dict, but with a special "add" method that autogenerated slaveids"""
    # intentionally in the class scope so all instances share the slave counter
    slaveid_generator = ('slave{:02d}'.format(i) for i in count())
    lock = RLock()
    _instances = []

    def __init__(self, *args, **kwargs):
        super(SlaveDict, self).__init__(*args, **kwargs)
        with self.lock:
            SlaveDict._instances.append(self)

    # autoincrement the slaveids when something is added
    def add(self, value):
        self[next(self.slaveid_generator)] = value

    # when removing a slave with this method, it is removed from all instances
    # use the normal `del` behavior to only remove from one instances
    def remove(self, key):
        with self.lock:
            for instance in self._instances:
                if key in instance:
                    del(instance[key])

    # helper to wrap dict method wrapper to generate methods protected by a lock
    # like a decorator, but takes a method name instead of wrapping
    def _lock_wrap(method_name):
        wrapped = getattr(dict, method_name)

        @wraps(wrapped)
        def wrapper(self, *args, **kwargs):
            with self.lock:
                return wrapped(self, *args, **kwargs)
        return wrapper
    # all mutating methods should be wrapped; if one is missing here that isn't intentional
    __setitem__ = _lock_wrap('__setitem__')
    __delitem__ = _lock_wrap('__delitem__')
    # destroy now-useless lock wrapper function
    del(_lock_wrap)


class ParallelSession(object):
    def __init__(self, config):
        self.config = config
        self.session = None
        self.session_finished = False
        self.countfailures = 0
        self.collection = OrderedDict()
        self.sent_tests = 0
        self.log = create_sublogger('master')
        self.maxfail = config.getvalue("maxfail")
        self._failed_collection_errors = {}
        self.terminal = store.terminalreporter
        self.trdist = None
        self.slaves = SlaveDict()
        self.slave_urls = SlaveDict()
        self.slave_tests = defaultdict(set)
        self.test_groups = self._test_item_generator()

        self._pool = []
        self.pool_lock = Lock()
        from utils.conf import cfme_data
        self.provs = sorted(set(cfme_data['management_systems'].keys()),
                            key=len, reverse=True)
        self.slave_allocation = collections.defaultdict(list)
        self.used_prov = set()

        self.failed_slave_test_groups = deque()
        self.slave_spawn_count = 0
        self.sprout_client = None
        self.sprout_timer = None
        self.sprout_pool = None
        if not self.config.op