# Copyright 2014 TrilioData Inc.
# All Rights Reserved.

"""Base classes for our unit tests.

Allows overriding of CONF for use of fakes, and some black magic for
inline callbacks.

"""


import contextlib
import functools
import os
import shutil
import tempfile
import uuid
import fixtures
import mox
import stubout
import testtools
import sqlalchemy as sa

from oslo_log import log as logging
from contego.tests.common import timeutils

try:
    from oslo.config import cfg
except ImportError:
    from oslo_config import cfg


test_opts = [
    cfg.StrOpt(
        "sqlite_clean_db", default="clean.sqlite", help="File name of clean sqlite db",
    ),
    cfg.BoolOpt(
        "fake_tests", default=True, help="should we use everything for testing"
    ),
]

CONF = cfg.CONF
CONF.register_opts(test_opts)

backend_opts = [
    cfg.StrOpt(
        "override_block",
        default="cinder",
        help="by default block disk_type is mapped to cinder"
        " It can be override to lvm by this option",
    ),
    cfg.StrOpt(
        "contego_staging_dir",
        default=None,
        help="by default uses CONF.instances_path"
        " It can be override depending on user configuration",
    ),
]

LOG = logging.getLogger(__name__)
CONF = cfg.CONF
CONF.register_opts(backend_opts, "backends")

LOG = logging.getLogger(__name__)

_DB_CACHE = None


class TestingException(Exception):
    pass


class TestCase(testtools.TestCase):
    """Test case base class for all unit tests."""

    def setUp(self):
        """Run before each test method to initialize test environment."""
        super(TestCase, self).setUp()

        test_timeout = os.environ.get("OS_TEST_TIMEOUT", 0)
        try:
            test_timeout = int(test_timeout)
        except ValueError:
            # If timeout value is invalid do not set a timeout.
            test_timeout = 0
        if test_timeout > 0:
            self.useFixture(fixtures.Timeout(test_timeout, gentle=True))
        self.useFixture(fixtures.NestedTempfile())
        self.useFixture(fixtures.TempHomeDir())

        if (
            os.environ.get("OS_STDOUT_CAPTURE") == "True"
            or os.environ.get("OS_STDOUT_CAPTURE") == "1"
        ):
            stdout = self.useFixture(fixtures.StringStream("stdout")).stream
            self.useFixture(fixtures.MonkeyPatch("sys.stdout", stdout))
        if (
            os.environ.get("OS_STDERR_CAPTURE") == "True"
            or os.environ.get("OS_STDERR_CAPTURE") == "1"
        ):
            stderr = self.useFixture(fixtures.StringStream("stderr")).stream
            self.useFixture(fixtures.MonkeyPatch("sys.stderr", stderr))

        self.log_fixture = self.useFixture(fixtures.FakeLogger())

        CONF([], default_config_files=[])

        self.start = timeutils.utcnow()
        self.log_fixture = self.useFixture(fixtures.FakeLogger())

        # emulate some of the mox stuff, we can't use the metaclass
        # because it screws with our generators
        self.mox = mox.Mox()
        self.stubs = stubout.StubOutForTesting()
        self.addCleanup(CONF.reset)
        self.addCleanup(self.mox.UnsetStubs)
        self.addCleanup(self.stubs.UnsetAll)
        self.addCleanup(self.stubs.SmartUnsetAll)
        self.addCleanup(self.mox.VerifyAll)
        self.injected = []
        self._services = []

    def tearDown(self):
        """Runs after each test method to tear down test environment."""

        # Stop any timers
        for x in self.injected:
            try:
                x.stop()
            except AssertionError:
                pass

        # Kill any services
        for x in self._services:
            try:
                x.kill()
            except Exception as ex:
                LOG.exception(ex)
                pass

        # Delete attributes that don't start with _ so they don't pin
        # memory around unnecessarily for the duration of the test
        # suite
        for key in [k for k in list(self.__dict__.keys()) if k[0] != "_"]:
            del self.__dict__[key]
        super(TestCase, self).tearDown()

    def flags(self, **kw):
        """Override CONF variables for a test."""
        for k, v in kw.items():
            CONF.set_override(k, v)

    def start_service(self, name, host=None, **kwargs):
        host = host and host or uuid.uuid4().hex
        kwargs.setdefault("host", host)
        kwargs.setdefault("binary", "contego-%s" % name)
        svc = service.Service.create(**kwargs)
        svc.start()
        self._services.append(svc)
        return svc

    # Useful assertions
    def assertDictMatch(self, d1, d2, approx_equal=False, tolerance=0.001):
        """Assert two dicts are equivalent.

        This is a 'deep' match in the sense that it handles nested
        dictionaries appropriately.

        NOTE:

            If you don't care (or don't know) a given value, you can specify
            the string DONTCARE as the value. This will cause that dict-item
            to be skipped.

        """

        def raise_assertion(msg):
            d1str = str(d1)
            d2str = str(d2)
            base_msg = (
                "Dictionaries do not match. %(msg)s d1: %(d1str)s "
                "d2: %(d2str)s" % {"msg": msg, "d1str": d1str, "d2str": d2str}
            )
            raise AssertionError(base_msg)

        d1keys = set(d1.keys())
        d2keys = set(d2.keys())
        if d1keys != d2keys:
            d1only = d1keys - d2keys
            d2only = d2keys - d1keys
            raise_assertion(
                "Keys in d1 and not d2: %(d1only)s. "
                "Keys in d2 and not d1: %(d2only)s"
                % {"d1only": d1only, "d2only": d2only}
            )

        for key in d1keys:
            d1value = d1[key]
            d2value = d2[key]
            try:
                error = abs(float(d1value) - float(d2value))
                within_tolerance = error <= tolerance
            except (ValueError, TypeError):
                # If both values aren't convertable to float, just ignore
                # ValueError if arg is a str, TypeError if it's something else
                # (like None)
                within_tolerance = False

            if hasattr(d1value, "keys") and hasattr(d2value, "keys"):
                self.assertDictMatch(d1value, d2value)
            elif "DONTCARE" in (d1value, d2value):
                continue
            elif approx_equal and within_tolerance:
                continue
            elif d1value != d2value:
                raise_assertion(
                    "d1['%(key)s']=%(d1value)s != "
                    "d2['%(key)s']=%(d2value)s"
                    % {"key": key, "d1value": d1value, "d2value": d2value}
                )