# -*- test-case-name: twisted.trial.test.test_runner -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
A miscellany of code used to run Trial tests.

Maintainer: Jonathan Lange
"""

from __future__ import absolute_import, division

__all__ = [
    'TestSuite',

    'DestructiveTestSuite', 'ErrorHolder', 'LoggedSuite',
    'TestHolder', 'TestLoader', 'TrialRunner', 'TrialSuite',

    'filenameToModule', 'isPackage', 'isPackageDirectory', 'isTestCase',
    'name', 'samefile', 'NOT_IN_TEST',
    ]

import doctest
import imp
import inspect
import os
import sys
import time
import types
import warnings

from twisted.python import reflect, log, failure, modules, filepath
from twisted.python.compat import _PY3

from twisted.internet import defer
from twisted.trial import util, unittest
from twisted.trial.itrial import ITestCase
from twisted.trial.reporter import _ExitWrapper, UncleanWarningsReporterWrapper
from twisted.trial._asyncrunner import _ForceGarbageCollectionDecorator, _iterateTests
from twisted.trial._synctest import _logObserver

# These are imported so that they remain in the public API for t.trial.runner
from twisted.trial.unittest import TestSuite

from zope.interface import implementer

pyunit = __import__('unittest')



def isPackage(module):
    """Given an object return True if the object looks like a package"""
    if not isinstance(module, types.ModuleType):
        return False
    basename = os.path.splitext(os.path.basename(module.__file__))[0]
    return basename == '__init__'


def isPackageDirectory(dirname):
    """Is the directory at path 'dirname' a Python package directory?
    Returns the name of the __init__ file (it may have a weird extension)
    if dirname is a package directory.  Otherwise, returns False"""
    for ext in list(zip(*imp.get_suffixes()))[0]:
        initFile = '__init__' + ext
        if os.path.exists(os.path.join(dirname, initFile)):
            return initFile
    return False


def samefile(filename1, filename2):
    """
    A hacky implementation of C{os.path.samefile}. Used by L{filenameToModule}
    when the platform doesn't provide C{os.path.samefile}. Do not use this.
    """
    return os.path.abspath(filename1) == os.path.abspath(filename2)


def filenameToModule(fn):
    """
    Given a filename, do whatever possible to return a module object matching
    that file.

    If the file in question is a module in Python path, properly import and
    return that module. Otherwise, load the source manually.

    @param fn: A filename.
    @return: A module object.
    @raise ValueError: If C{fn} does not exist.
    """
    if not os.path.exists(fn):
        raise ValueError("%r doesn't exist" % (fn,))
    try:
        ret = reflect.namedAny(reflect.filenameToModuleName(fn))
    except (ValueError, AttributeError):
        # Couldn't find module.  The file 'fn' is not in PYTHONPATH
        return _importFromFile(fn)

    if not hasattr(ret, "__file__"):
        # This isn't a Python module in a package, so import it from a file
        return _importFromFile(fn)

    # ensure that the loaded module matches the file
    retFile = os.path.splitext(ret.__file__)[0] + '.py'
    # not all platforms (e.g. win32) have os.path.samefile
    same = getattr(os.path, 'samefile', samefile)
    if os.path.isfile(fn) and not same(fn, retFile):
        del sys.modules[ret.__name__]
        ret = _importFromFile(fn)
    return ret


def _importFromFile(fn, moduleName=None):
    fn = _resolveDirectory(fn)
    if not moduleName:
        moduleName = os.path.splitext(os.path.split(fn)[-1])[0]
    if moduleName in sys.modules:
        return sys.modules[moduleName]
    with open(fn, 'r') as fd:
        module = imp.load_source(moduleName, fn, fd)
    return module


def _resolveDirectory(fn):
    if os.path.isdir(fn):
        initFile = isPackageDirectory(fn)
        if initFile:
            fn = os.path.join(fn, initFile)
        else:
            raise ValueError('%r is not a package directory' % (fn,))
    return fn


def _getMethodNameInClass(method):
    """
    Find the attribute name on the method's class which refers to the method.

    For some methods, notably decorators which have not had __name__ set correctly:

    getattr(method.im_class, method.__name__) != method
    """
    if getattr(method.im_class, method.__name__, object()) != method:
        for alias in dir(method.im_class):
            if getattr(method.im_class, alias, object()) == method:
                return alias
    return method.__name__


class DestructiveTestSuite(TestSuite):
    """
    A test suite which remove the tests once run, to minimize memory usage.
    """

    def run(self, result):
        """
        Almost the same as L{TestSuite.run}, but with C{self._tests} being
        empty at the end.
        """
        while self._tests:
            if result.shouldStop:
                break
            test = self._tests.pop(0)
            test(result)
        return result



# When an error occurs outside of any test, the user will see this string
# in place of a test's name.
NOT_IN_TEST = "<not in test>"



class LoggedSuite(TestSuite):
    """
    Any errors logged in this suite will be reported to the L{TestResult}
    object.
    """

    def run(self, result):
        """
        Run the suite, storing all errors in C{result}. If an error is logged
        while no tests are running, then it will be added as an error to
        C{result}.

        @param result: A L{TestResult} object.
        """
        observer = _logObserver
        observer._add()
        super(LoggedSuite, self).run(result)
        observer._remove()
        for error in observer.getErrors():
            result.addError(TestHolder(NOT_IN_TEST), error)
        observer.flushErrors()



class TrialSuite(TestSuite):
    """
    Suite to wrap around every single test in a C{trial} run. Used internally
    by Trial to set up things necessary for Trial tests to work, regardless of
    what context they are run in.
    """

    def __init__(self, tests=(), forceGarbageCollection=False):
        if forceGarbageCollection:
            newTests = []
            for test in tests:
                test = unittest.decorate(
                    test, _ForceGarbageCollectionDecorator)
                newTests.append(test)
            tests = newTests
        suite = LoggedSuite(tests)
        super(TrialSuite, self).__init__([suite])


    def _bail(self):
        from twisted.internet import reactor
        d = defer.Deferred()
        reactor.addSystemEventTrigger('after', 'shutdown',
                                      lambda: d.callback(None))
        reactor.fireSystemEvent('shutdown') # radix's suggestion
        # As long as TestCase does crap stuff with the reactor we need to
        # manually shutdown the reactor here, and that requires util.wait
        # :(
        # so that the shutdown event completes
        unittest.TestCase('mktemp')._wait(d)

    def run(self, result):
        try:
            TestSuite.run(self, result)
        finally:
            self._bail()


def name(thing):
    """
    @param thing: an object from modules (instance of PythonModule,
        PythonAttribute), a TestCase subclass, or an instance of a TestCase.
    """
    if isTestCase(thing):
        # TestCase subclass
        theName = reflect.qual(thing)
    else:
        # thing from trial, or thing from modules.
        # this monstrosity exists so that modules' objects do not have to
        # implement id(). -jml
        try:
            theName = thing.id()
        except AttributeError:
            theName = thing.name
    return theName


def isTestCase(obj):
    """
    @return: C{True} if C{obj} is a class that contains test cases, C{False}
        otherwise. Used to find all the tests in a module.
    """
    try:
        return issubclass(obj, pyunit.TestCase)
    except TypeError:
        return False



@implementer(ITestCase)
class TestHolder(object):
    """
    Placeholder for a L{TestCase} inside a reporter. As far as a L{TestResult}
    is concerned, this looks exactly like a unit test.
    """

    failureException = None

    def __init__(self, description):
        """
        @param description: A string to be displayed L{TestResult}.
        """
        self.description = description


    def __call__(self, result):
        return self.run(result)


    def id(self):
        return self.description


    def countTestCases(self):
        return 0


    def run(self, result):
        """
        This test is just a placeholder. Run the test successfully.

        @param result: The C{TestResult} to store the results in.
        @type result: L{twisted.trial.itrial.IReporter}.
        """
        result.startTest(self)
        result.addSuccess(self)
        result.stopTest(self)


    def shortDescription(self):
        return self.description



class ErrorHolder(TestHolder):
    """
    Used to insert arbitrary errors into a test suite run. Provides enough
    methods to look like a C{TestCase}, however, when it is run, it simply adds
    an error to the C{TestResult}. The most common use-case is for when a
    module fails to import.
    """

    def __init__(self, description, error):
        """
        @param description: A string used by C{TestResult}s to identify this
        error. Generally, this is the name of a module that failed to import.

        @param error: The error to be added to the result. Can be an `exc_info`
        tuple or a L{twisted.python.failure.Failure}.
        """
        super(ErrorHolder, self).__init__(description)
        self.error = util.excInfoOrFailureToExcInfo(error)


    def __repr__(self):
        return "<ErrorHolder description=%r error=%r>" % (
            self.description, self.error[1])


    def run(self, result):
        """
        Run the test, reporting the error.

        @param result: The C{TestResult} to store the results in.
        @type result: L{twisted.trial.itrial.IReporter}.
        """
        result.startTest(self)
        result.addError(self, self.error)
        result.stopTest(self)



class TestLoader(object):
    """
    I find tests inside function, modules, files -- whatever -- then return
    them wrapped inside a Test (either a L{TestSuite} or a L{TestCase}).

    @ivar methodPrefix: A string prefix. C{TestLoader} will assume that all the
    methods in a class that begin with C{methodPrefix} are test cases.

    @ivar modulePrefix: A string prefix. Every module in a package that begins
    with C{modulePrefix} is considered a module full of tests.

    @ivar forceGarbageCollection: A flag applied to each C{TestCase} loaded.
    See L{unittest.TestCase} for more information.

    @ivar sorter: A key function used to sort C{TestCase}s, test classes,
    modules and packages.

    @ivar suiteFactory: A callable which is passed a list of tests (which
    themselves may be suites of tests). Must return a test suite.
    """

    methodPrefix = 'test'
    modulePrefix = 'test_'

    def __init__(self):
        self.suiteFactory = TestSuite
        self.sorter = name
        self._importErrors = []

    def sort(self, xs):
        """
        Sort the given things using L{sorter}.

        @param xs: A list of test cases, class or modules.
        """
        return sorted(xs, key=self.sorter)

    def findTestClasses(self, module):
        """Given a module, return all Trial test classes"""
        classes = []
        for name, val in inspect.getmembers(module):