tests/conftest.py

# ============LICENSE_START=======================================================
# Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#
"""
standard pytest file that contains the shared fixtures
https://docs.pytest.org/en/latest/fixture.html
"""
import base64
import copy
import json
import os

import pytest

from policyhandler.config import Config
from policyhandler.discovery import DiscoveryClient
from policyhandler.policy_consts import CATCH_UP, TARGET_ENTITY
from policyhandler.utils import Utils

from .mock_settings import MockSettings
from .mock_tracker import MockHttpResponse, Tracker

_LOGGER = Utils.get_logger(__file__)

MockSettings.init_mock_config()

@pytest.fixture(scope="session", autouse=True)
def _auto_setup__global():
    """initialize the _auto_setup__global per the whole test session"""
    _LOGGER.info("_auto_setup__global")

    yield _auto_setup__global
    Tracker.log_all_tests()
    _LOGGER.info("teardown _auto_setup__global")


@pytest.fixture(autouse=True, scope="module")
def _auto_module_cycle(request):
    """log all the test starts and ends"""
    module_name = request.module.__name__.replace(".", "/")

    _LOGGER.info("start_module: %s %s", module_name, "->"*25)
    yield _auto_module_cycle
    _LOGGER.info("end_module: %s %s", module_name, "<-"*25)

@pytest.fixture(autouse=True)
def _auto_test_cycle(request):
    """log all the test starts and ends"""
    _LOGGER.info("-"*75)
    module_name = request.module.__name__.replace(".", "/")
    if request.cls:
        test_name = "%s.py::%s::%s" % (module_name, request.cls.__name__,
                                       request.function.__name__)
    else:
        test_name = "%s.py::%s" % (module_name, request.function.__name__)

    Tracker.reset(test_name)
    _LOGGER.info(">>>>>>> start [%s]: %s", len(Tracker.test_names), test_name)
    yield _auto_test_cycle
    _LOGGER.info(">>>>>>> tracked messages: %s", Tracker.to_string())
    _LOGGER.info(">>>>>>> %s[%s]: %s", Tracker.get_status(test_name),
                 len(Tracker.test_names), test_name)


@pytest.fixture()
def fix_cherrypy_engine_exit(monkeypatch):
    """monkeyed cherrypy.engine.exit()"""
    _LOGGER.info("setup fix_cherrypy_engine_exit")

    def monkeyed_cherrypy_engine_exit():
        """monkeypatch for deploy_handler"""
        _LOGGER.info("cherrypy_engine_exit()")

    monkeypatch.setattr('policyhandler.web_server.cherrypy.engine.exit',
                        monkeyed_cherrypy_engine_exit)
    yield fix_cherrypy_engine_exit
    _LOGGER.info("teardown fix_cherrypy_engine_exit")



@pytest.fixture()
def fix_discovery(monkeypatch):
    """monkeyed discovery request.get"""
    def monkeyed_discovery(uri):
        """monkeypatch for get from consul"""
        res_json = {}
        dh_service = None
        if Config.discovered_config:
            _, dh_config = Config.discovered_config.get_by_key(Config.DEPLOY_HANDLER)
            dh_config = dh_config and dh_config.get(TARGET_ENTITY)
        if dh_service and uri == DiscoveryClient.CONSUL_SERVICE_MASK.format(
                Config.consul_url, dh_service):
            res_json = [{
                "ServiceAddress": "1.1.1.1",
                "ServicePort": "123"
            }]
        elif uri == DiscoveryClient.CONSUL_KV_MASK.format(
                Config.consul_url, Config.system_name):
            res_json = [{"Value": base64.b64encode(
                json.dumps(MockSettings.mock_config).encode()).decode("utf-8")}]
        return MockHttpResponse("get", uri, res_json=res_json)

    _LOGGER.info("setup fix_discovery")
    monkeypatch.setattr('policyhandler.discovery.requests.get', monkeyed_discovery)
    yield fix_discovery
    _LOGGER.info("teardown fix_discovery")

@pytest.fixture(scope="module")
def fix_auto_catch_up():
    """increase the frequency of auto catch_up"""

    _LOGGER.info("setup fix_auto_catch_up %s", json.dumps(MockSettings.mock_config))
    prev_config = copy.deepcopy(MockSettings.mock_config)
    MockSettings.mock_config.get(Config.SERVICE_NAME_POLICY_HANDLER, {}) \
        .get(CATCH_UP, {})[Config.TIMER_INTERVAL] = 5
    _LOGGER.info("fix_auto_catch_up %s", json.dumps(MockSettings.mock_config))
    MockSettings.rediscover_config()

    yield fix_auto_catch_up
    MockSettings.rediscover_config(prev_config)
    _LOGGER.info("teardown fix_auto_catch_up")

@pytest.fixture(scope="module")
def fix_pdp_authorization():
    """set env vars that overwrite the headers.Authorization on pdp and dmaap_mr clients"""
    _LOGGER.info("setup fix_pdp_authorization %s", json.dumps(MockSettings.mock_config))
    prev_config = copy.deepcopy(MockSettings.mock_config)

    os.environ.update({
        Config.PDP_USER: "alex-PDP_USER",
        Config.PDP_PWD: "alex-PDP_PWD",
        Config.DMAAP_MR_USER: "alex-DMAAP_MR_USER",
        Config.DMAAP_MR_PWD: "alex-DMAAP_MR_PWD"
    })
    Config._local_config._config = None
    Config._pdp_authorization = None
    Config._dmaap_mr_authorization = None
    MockSettings.reinit_mock_config()
    MockSettings.rediscover_config()

    _LOGGER.info("fix_pdp_authorization %s, %s: %s:%s %s:%s",
                 json.dumps(Config._pdp_authorization), json.dumps(Config._dmaap_mr_authorization),
                 os.environ.get(Config.PDP_USER), os.environ.get(Config.PDP_PWD),
                 os.environ.get(Config.DMAAP_MR_USER), os.environ.get(Config.DMAAP_MR_PWD))

    yield fix_pdp_authorization

    del os.environ[Config.PDP_USER]
    del os.environ[Config.PDP_PWD]
    del os.environ[Config.DMAAP_MR_USER]
    del os.environ[Config.DMAAP_MR_PWD]

    Config._local_config._config = None
    Config._pdp_authorization = None
    Config._dmaap_mr_authorization = None
    MockSettings.reinit_mock_config()
    MockSettings.rediscover_config(prev_config)
    _LOGGER.info("teardown fix_pdp_authorization %s, %s: %s:%s %s:%s",
                 json.dumps(Config._pdp_authorization), json.dumps(Config._dmaap_mr_authorization),
                 os.environ.get(Config.PDP_USER), os.environ.get(Config.PDP_PWD),
                 os.environ.get(Config.DMAAP_MR_USER), os.environ.get(Config.DMAAP_MR_PWD))