diff options
Diffstat (limited to 'tests/conftest.py')
| -rw-r--r-- | tests/conftest.py | 299 |
1 files changed, 299 insertions, 0 deletions
diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..0dcb2bb --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,299 @@ +# ============LICENSE_START======================================================= +# Copyright (c) 2018 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========================================================= +# +# ECOMP is a trademark and service mark of AT&T Intellectual Property. +""" +startdard pytest file that contains the shared fixtures +https://docs.pytest.org/en/latest/fixture.html +""" +import base64 +import copy +import json + +import pytest + +from policyhandler.config import Config +from policyhandler.deploy_handler import DeployHandler +from policyhandler.discovery import DiscoveryClient +from policyhandler.onap.audit import Audit +from policyhandler.policy_consts import CATCH_UP, POLICY_NAME, TARGET_ENTITY +from policyhandler.policy_receiver import PolicyReceiver +from policyhandler.policy_rest import PolicyRest + +from .mock_deploy_handler import MockDeploymentHandler +from .mock_policy_engine import MockPolicyEngine +from .mock_settings import Settings +from .mock_tracker import MockHttpResponse, Tracker +from .mock_websocket import MockWebSocket + + +@pytest.fixture(autouse=True) +def _auto_test_cycle(request): + """log all the test starts and ends""" + if request.cls: + test_name = "%s::%s::%s" % (request.module.__name__, + request.cls.__name__, + request.function.__name__) + else: + test_name = "%s::%s" % (request.module.__name__, request.function.__name__) + + Tracker.reset(test_name) + if Settings.logger: + Settings.logger.info(">>>>>>> start %s", test_name) + yield _auto_test_cycle + if Settings.logger: + Settings.logger.info(">>>>>>> tracked messages: %s", Tracker.to_string()) + Settings.logger.info(">>>>>>> ended %s", test_name) + + +@pytest.fixture(scope="session", autouse=True) +def _auto_setup_policy_engine(): + """initialize the mock-policy-engine per the whole test session""" + Settings.init() + + Settings.logger.info("create _auto_setup_policy_engine") + MockPolicyEngine.init() + yield _auto_setup_policy_engine + Settings.logger.info("teardown _auto_setup_policy_engine") + + +@pytest.fixture() +def fix_pdp_post(monkeypatch): + """monkeyed request /getConfig to PDP""" + def monkeyed_policy_rest_post(uri, json=None, **kwargs): + """monkeypatch for the POST to policy-engine""" + res_json = MockPolicyEngine.get_config(json.get(POLICY_NAME)) + return MockHttpResponse("post", uri, res_json, json=json, **kwargs) + + Settings.logger.info("setup fix_pdp_post") + PolicyRest._lazy_init() + monkeypatch.setattr('policyhandler.policy_rest.PolicyRest._requests_session.post', + monkeyed_policy_rest_post) + yield fix_pdp_post + Settings.logger.info("teardown fix_pdp_post") + +@pytest.fixture() +def fix_deploy_handler(monkeypatch): + """monkeyed requests to deployment-handler""" + def monkeyed_deploy_handler_put(uri, **kwargs): + """monkeypatch for policy-update request.put to deploy_handler""" + return MockHttpResponse("put", uri, MockDeploymentHandler.default_response(), + **kwargs) + + def monkeyed_deploy_handler_get(uri, **kwargs): + """monkeypatch policy-update request.get to deploy_handler""" + return MockHttpResponse("get", uri, MockDeploymentHandler.get_deployed_policies(), + **kwargs) + + Settings.logger.info("setup fix_deploy_handler") + audit = None + if DeployHandler._lazy_inited is False: + audit = Audit(req_message="fix_deploy_handler") + DeployHandler._lazy_init(audit) + + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.put', + monkeyed_deploy_handler_put) + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.get', + monkeyed_deploy_handler_get) + + yield fix_deploy_handler + if audit: + audit.audit_done("teardown") + Settings.logger.info("teardown fix_deploy_handler") + + +@pytest.fixture() +def fix_cherrypy_engine_exit(monkeypatch): + """monkeyed cherrypy.engine.exit()""" + Settings.logger.info("setup fix_cherrypy_engine_exit") + + def monkeyed_cherrypy_engine_exit(): + """monkeypatch for deploy_handler""" + Settings.logger.info("cherrypy_engine_exit()") + + monkeypatch.setattr('policyhandler.web_server.cherrypy.engine.exit', + monkeyed_cherrypy_engine_exit) + yield fix_cherrypy_engine_exit + Settings.logger.info("teardown fix_cherrypy_engine_exit") + + +@pytest.fixture() +def fix_pdp_post_big(monkeypatch): + """monkeyed request /getConfig to PDP""" + def monkeyed_policy_rest_post(uri, **kwargs): + """monkeypatch for the POST to policy-engine""" + res_json = MockPolicyEngine.get_configs_all() + return MockHttpResponse("post", uri, res_json, **kwargs) + + Settings.logger.info("setup fix_pdp_post_big") + PolicyRest._lazy_init() + monkeypatch.setattr('policyhandler.policy_rest.PolicyRest._requests_session.post', + monkeyed_policy_rest_post) + yield fix_pdp_post_big + Settings.logger.info("teardown fix_pdp_post_big") + + +class MockException(Exception): + """mock exception""" + pass + + +@pytest.fixture() +def fix_pdp_post_boom(monkeypatch): + """monkeyed request /getConfig to PDP - exception""" + def monkeyed_policy_rest_post_boom(uri, **_): + """monkeypatch for the POST to policy-engine""" + raise MockException("fix_pdp_post_boom {}".format(uri)) + + Settings.logger.info("setup fix_pdp_post_boom") + PolicyRest._lazy_init() + monkeypatch.setattr('policyhandler.policy_rest.PolicyRest._requests_session.post', + monkeyed_policy_rest_post_boom) + yield fix_pdp_post_boom + Settings.logger.info("teardown fix_pdp_post_boom") + + +@pytest.fixture() +def fix_select_latest_policies_boom(monkeypatch): + """monkeyed exception""" + def monkeyed_boom(*args, **kwargs): + """monkeypatch for the select_latest_policies""" + raise MockException("monkeyed_boom") + + Settings.logger.info("setup fix_select_latest_policies_boom") + monkeypatch.setattr('policyhandler.policy_utils.PolicyUtils.select_latest_policies', + monkeyed_boom) + monkeypatch.setattr('policyhandler.policy_utils.PolicyUtils.select_latest_policy', + monkeyed_boom) + monkeypatch.setattr('policyhandler.policy_utils.PolicyUtils.extract_policy_id', + monkeyed_boom) + + yield fix_select_latest_policies_boom + Settings.logger.info("teardown fix_select_latest_policies_boom") + +@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(Settings.mock_config).encode()).decode("utf-8")}] + return MockHttpResponse("get", uri, res_json) + + Settings.logger.info("setup fix_discovery") + monkeypatch.setattr('policyhandler.discovery.requests.get', monkeyed_discovery) + yield fix_discovery + Settings.logger.info("teardown fix_discovery") + + +@pytest.fixture(scope="module") +def fix_auto_catch_up(): + """increase the frequency of auto catch_up""" + + Settings.logger.info("setup fix_auto_catch_up %s", json.dumps(Settings.mock_config)) + prev_config = copy.deepcopy(Settings.mock_config) + Settings.mock_config.get(Config.SERVICE_NAME_POLICY_HANDLER, {}) \ + .get(CATCH_UP, {})[Config.TIMER_INTERVAL] = 5 + Settings.logger.info("fix_auto_catch_up %s", json.dumps(Settings.mock_config)) + Settings.rediscover_config() + + yield fix_auto_catch_up + Settings.rediscover_config(prev_config) + Settings.logger.info("teardown fix_auto_catch_up") + + +@pytest.fixture() +def fix_deploy_handler_413(monkeypatch): + """monkeyed failed discovery request.get""" + def monkeyed_deploy_handler_put(uri, **kwargs): + """monkeypatch for deploy_handler""" + return MockHttpResponse( + "put", uri, + {"server_instance_uuid": Settings.deploy_handler_instance_uuid}, + status_code=413, **kwargs + ) + + def monkeyed_deploy_handler_get(uri, **kwargs): + """monkeypatch policy-update request.get to deploy_handler""" + return MockHttpResponse("get", uri, MockDeploymentHandler.get_deployed_policies(), + **kwargs) + + Settings.logger.info("setup fix_deploy_handler_413") + audit = None + if DeployHandler._lazy_inited is False: + audit = Audit(req_message="fix_deploy_handler_413") + DeployHandler._lazy_init(audit) + + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.put', + monkeyed_deploy_handler_put) + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.get', + monkeyed_deploy_handler_get) + + yield fix_deploy_handler_413 + if audit: + audit.audit_done("teardown") + Settings.logger.info("teardown fix_deploy_handler_413") + + +@pytest.fixture() +def fix_deploy_handler_404(monkeypatch): + """monkeyed failed discovery request.get""" + def monkeyed_deploy_handler_put(uri, **kwargs): + """monkeypatch for deploy_handler""" + return MockHttpResponse("put", uri, MockDeploymentHandler.default_response(), + **kwargs) + + def monkeyed_deploy_handler_get(uri, **kwargs): + """monkeypatch policy-update request.get to deploy_handler""" + return MockHttpResponse("get", uri, MockDeploymentHandler.default_response(), + **kwargs) + + Settings.logger.info("setup fix_deploy_handler_404") + audit = None + if DeployHandler._lazy_inited is False: + audit = Audit(req_message="fix_deploy_handler_404") + DeployHandler._lazy_init(audit) + + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.put', + monkeyed_deploy_handler_put) + monkeypatch.setattr('policyhandler.deploy_handler.DeployHandler._requests_session.get', + monkeyed_deploy_handler_get) + + yield fix_deploy_handler_404 + if audit: + audit.audit_done("teardown") + Settings.logger.info("teardown fix_deploy_handler_404") + +@pytest.fixture() +def fix_policy_receiver_websocket(monkeypatch): + """monkeyed websocket for policy_receiver""" + Settings.logger.info("setup fix_policy_receiver_websocket") + monkeypatch.setattr('policyhandler.policy_receiver.websocket', MockWebSocket) + yield fix_policy_receiver_websocket + Settings.logger.info("teardown fix_policy_receiver_websocket") |