tests/test_controller.py

   1 # ==================================================================================
   2 #       Copyright (c) 2019 Nokia
   3 #       Copyright (c) 2018-2019 AT&T Intellectual Property.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 import tempfile
  18 import os
  19
  20 from rmr.rmr_mocks import rmr_mocks
  21 from a1 import app
  22 import testing_helpers
  23 import pytest
  24
  25
  26 ADM_CTRL = "admission_control_policy"
  27 ADM_CTRL_INSTANCE = "/a1-p/policytypes/20000/policies/" + ADM_CTRL
  28 ADM_CTRL_INSTANCE_STATUS = ADM_CTRL_INSTANCE + "/status"
  29
  30
  31 # http://flask.pocoo.org/docs/1.0/testing/
  32 @pytest.fixture
  33 def client():
  34     db_fd, app.app.config["DATABASE"] = tempfile.mkstemp()
  35     app.app.config["TESTING"] = True
  36     cl = app.app.test_client()
  37
  38     yield cl
  39
  40     os.close(db_fd)
  41     os.unlink(app.app.config["DATABASE"])
  42
  43
  44 def _fake_dequeue(_filter_type):
  45     """
  46     for monkeypatching a1rmnr.dequeue_all_messages
  47     """
  48     fake_msg = {}
  49     pay = b'{"policy_type_id": 20000, "policy_instance_id": "admission_control_policy", "handler_id": "test_receiver", "status": "OK"}'
  50     fake_msg["payload"] = pay
  51     new_messages = [fake_msg]
  52     return new_messages
  53
  54
  55 def _test_put_patch(monkeypatch):
  56     testing_helpers.patch_all(monkeypatch)
  57     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(0))  # good sends for this whole batch
  58
  59     # we need to repatch alloc (already patched in patch_rmr) to fix the transactionid, alloc is called in send and recieve
  60     def fake_alloc(_unused, _alsounused):
  61         sbuf = rmr_mocks.Rmr_mbuf_t()
  62         sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
  63         return sbuf
  64
  65     # we also need to repatch set, since in the send function, we alloc, then set a new transid
  66     def fake_set_transactionid(sbuf):
  67         sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
  68
  69     # Note, we could have just patched summary, but this patches at a "lower level" so is a better test
  70     monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
  71     monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_set_transactionid)
  72
  73
  74 # Actual Tests
  75
  76
  77 # def test_policy_get(client, monkeypatch):
  78 #     """
  79 #     test policy GET
  80 #     """
  81 #     _test_put_patch(monkeypatch)
  82 #     monkeypatch.setattr(
  83 #         "a1.a1rmr.dequeue_all_waiting_messages",
  84 #         _fake_dequeue(monkeypatch, msg_payload={"GET ack": "pretend policy is here"}, msg_type=20003),
  85 #     )
  86 #     res = client.get("/a1-p/policies/admission_control_policy")
  87 #     assert res.status_code == 200
  88 #     assert res.json == {"GET ack": "pretend policy is here"}
  89 #
  90 #
  91 # def test_policy_get_unsupported(client, monkeypatch):
  92 #     """
  93 #     test policy GET
  94 #     """
  95 #     testing_helpers.patch_all(monkeypatch, nofetch=True)
  96 #     res = client.get("/a1-p/policies/admission_control_policy")
  97 #     assert res.status_code == 400
  98 #     assert res.data == b'"POLICY DOES NOT SUPPORT FETCHING"\n'
  99 #
 100 #
 101 def test_xapp_put_good(client, monkeypatch):
 102     """ test policy put good"""
 103
 104     # nothing there yet
 105     res = client.get(ADM_CTRL_INSTANCE)
 106     assert res.status_code == 404
 107     res = client.get(ADM_CTRL_INSTANCE_STATUS)
 108     assert res.status_code == 404
 109
 110     # create a good instance
 111     _test_put_patch(monkeypatch)
 112     res = client.put(ADM_CTRL_INSTANCE, json=testing_helpers.good_payload())
 113     assert res.status_code == 201
 114
 115     # get the instance
 116     res = client.get(ADM_CTRL_INSTANCE)
 117     assert res.status_code == 200
 118     assert res.json == testing_helpers.good_payload()
 119
 120     # get the instance status
 121     monkeypatch.setattr("a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue)
 122     res = client.get(ADM_CTRL_INSTANCE_STATUS)
 123     assert res.status_code == 200
 124     assert res.json == [{"handler_id": "test_receiver", "status": "OK"}]
 125
 126     # assert that rmr bad states don't cause problems
 127     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(10))
 128     res = client.put(ADM_CTRL_INSTANCE, json=testing_helpers.good_payload())
 129     assert res.status_code == 201
 130
 131     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(5))
 132     res = client.put(ADM_CTRL_INSTANCE, json=testing_helpers.good_payload())
 133     assert res.status_code == 201
 134
 135
 136 #
 137 #
 138 # def test_xapp_put_bad(client, monkeypatch):
 139 #     """Test policy put fails"""
 140 #     _test_put_patch(monkeypatch)
 141 #     # return from policy handler has a status indicating FAIL
 142 #     monkeypatch.setattr(
 143 #         "a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"status": "FAIL", "foo": "bar"})
 144 #     )
 145 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 146 #     assert res.status_code == 502
 147 #     assert res.json["reason"] == "BAD STATUS"
 148 #     assert res.json["return_payload"] == {"status": "FAIL", "foo": "bar"}
 149 #
 150 #     # return from policy handler has no status field
 151 #     monkeypatch.setattr("a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"foo": "bar"}))
 152 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 153 #     assert res.status_code == 502
 154 #     assert res.json["reason"] == "NO STATUS"
 155 #     assert res.json["return_payload"] == {"foo": "bar"}
 156 #
 157 #     # return from policy handler not a json
 158 #     monkeypatch.setattr(
 159 #         "a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload="booger", jsonb=False)
 160 #     )
 161 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 162 #     assert res.status_code == 502
 163 #     assert res.json["reason"] == "NOT JSON"
 164 #     assert res.json["return_payload"] == "booger"
 165 #
 166 #     # bad type
 167 #     monkeypatch.setattr("a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_type=666))
 168 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 169 #     assert res.status_code == 504
 170 #     assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
 171 #
 172 #     # bad state
 173 #     monkeypatch.setattr("a1.a1rmr.dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_state=666))
 174 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 175 #     assert res.status_code == 504
 176 #     assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
 177 #
 178 #
 179 def test_bad_requests(client, monkeypatch):
 180     """
 181     Test bad send failures
 182     """
 183     testing_helpers.patch_all(monkeypatch)
 184     res = client.put(ADM_CTRL_INSTANCE, json={"not": "expected"})
 185     assert res.status_code == 400
 186
 187     # bad media type
 188     res = client.put(ADM_CTRL_INSTANCE, data="notajson")
 189     assert res.status_code == 415
 190
 191     # test a PUT body against a poliucy not expecting one
 192     res = client.put("/a1-p/policytypes/20001/policies/test_policy", json=testing_helpers.good_payload())
 193     assert res.status_code == 400
 194     assert res.data == b'"BODY SUPPLIED BUT POLICY HAS NO EXPECTED BODY"\n'
 195
 196
 197 # def test_bad_requests(client, monkeypatch):
 198 #     """Test bad requests"""
 199 #     testing_helpers.patch_all(monkeypatch)
 200 #
 201 #     # test a 404
 202 #     res = client.put("/a1-p/policies/noexist", json=testing_helpers.good_payload())
 203 #     assert res.status_code == 404
 204
 205
 206 # def test_missing_manifest(client, monkeypatch):
 207 #     """
 208 #     test that we get a 500 with an approrpiate message on a missing manifest
 209 #     """
 210 #
 211 #     def f():
 212 #         raise exceptions.MissingManifest()
 213 #
 214 #     monkeypatch.setattr("a1.utils.get_ric_manifest", f)
 215 #
 216 #     res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
 217 #     assert res.status_code == 500
 218 #     assert res.data == b'"A1 was unable to find the required RIC manifest. report this!"\n'
 219 #
 220 #
 221
 222
 223 def test_healthcheck(client):
 224     """
 225     test healthcheck
 226     """
 227     res = client.get("/a1-p/healthcheck")
 228     assert res.status_code == 200