1 # ==================================================================================
2 # Copyright (c) 2019 Nokia
3 # Copyright (c) 2018-2019 AT&T Intellectual Property.
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
20 from rmr.rmr_mocks import rmr_mocks
22 import testing_helpers
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"
31 # http://flask.pocoo.org/docs/1.0/testing/
34 db_fd, app.app.config["DATABASE"] = tempfile.mkstemp()
35 app.app.config["TESTING"] = True
36 cl = app.app.test_client()
41 os.unlink(app.app.config["DATABASE"])
44 def _fake_dequeue(_filter_type):
46 for monkeypatching a1rmnr.dequeue_all_messages
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]
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
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"
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"
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)
77 # def test_policy_get(client, monkeypatch):
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),
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"}
91 # def test_policy_get_unsupported(client, monkeypatch):
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'
101 def test_xapp_put_good(client, monkeypatch):
102 """ test policy put good"""
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
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
116 res = client.get(ADM_CTRL_INSTANCE)
117 assert res.status_code == 200
118 assert res.json == testing_helpers.good_payload()
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"}]
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
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
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"})
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"}
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"}
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)
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"
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"
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"
179 def test_bad_requests(client, monkeypatch):
181 Test bad send failures
183 testing_helpers.patch_all(monkeypatch)
184 res = client.put(ADM_CTRL_INSTANCE, json={"not": "expected"})
185 assert res.status_code == 400
188 res = client.put(ADM_CTRL_INSTANCE, data="notajson")
189 assert res.status_code == 415
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'
197 # def test_bad_requests(client, monkeypatch):
198 # """Test bad requests"""
199 # testing_helpers.patch_all(monkeypatch)
202 # res = client.put("/a1-p/policies/noexist", json=testing_helpers.good_payload())
203 # assert res.status_code == 404
206 # def test_missing_manifest(client, monkeypatch):
208 # test that we get a 500 with an approrpiate message on a missing manifest
212 # raise exceptions.MissingManifest()
214 # monkeypatch.setattr("a1.utils.get_ric_manifest", f)
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'
223 def test_healthcheck(client):
227 res = client.get("/a1-p/healthcheck")
228 assert res.status_code == 200