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 # ==================================================================================
19 from rmr.rmr_mocks import rmr_mocks
21 from a1 import exceptions
23 import testing_helpers
26 # http://flask.pocoo.org/docs/1.0/testing/
29 db_fd, app.app.config["DATABASE"] = tempfile.mkstemp()
30 app.app.config["TESTING"] = True
31 cl = app.app.test_client()
36 os.unlink(app.app.config["DATABASE"])
41 msg_payload={"status": "SUCCESS", "foo": "bar"},
45 unexpected_first=True,
48 generates a mock rmr message response (returns a function that does; uses closures to set params)
51 # stick a message we don't want at the front of the queue, then stick the message we want
53 monkeypatch.setattr("rmr.rmr.rmr_torcv_msg", rmr_mocks.rcv_mock_generator(msg_payload, -1, msg_state, jsonb))
54 sbuf = rmr.rmr_alloc_msg(None, None)
55 sbuf = rmr.rmr_torcv_msg(None, sbuf, None)
56 summary = rmr.message_summary(sbuf)
57 new_messages.append(summary)
59 monkeypatch.setattr("rmr.rmr.rmr_torcv_msg", rmr_mocks.rcv_mock_generator(msg_payload, msg_type, msg_state, jsonb))
60 sbuf = rmr.rmr_alloc_msg(None, None)
61 sbuf = rmr.rmr_torcv_msg(None, sbuf, None)
62 summary = rmr.message_summary(sbuf)
63 new_messages.append(summary)
71 def _test_put_patch(monkeypatch):
72 testing_helpers.patch_all(monkeypatch)
73 monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(0)) # good sends for this whole batch
75 # we need to repatch alloc (already patched in patch_rmr) to fix the transactionid, alloc is called in send and recieve
76 def fake_alloc(_unused, _alsounused):
77 sbuf = rmr_mocks.Rmr_mbuf_t()
78 sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
81 # we also need to repatch set, since in the send function, we alloc, then set a new transid
82 def fake_set_transactionid(sbuf):
83 sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
85 # Note, we could have just patched summary, but this patches at a "lower level" so is a better test
86 monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
87 monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_set_transactionid)
93 def test_policy_get(client, monkeypatch):
97 _test_put_patch(monkeypatch)
99 "a1.a1rmr._dequeue_all_waiting_messages",
100 _fake_dequeue(monkeypatch, msg_payload={"GET ack": "pretend policy is here"}, msg_type=20003),
102 res = client.get("/a1-p/policies/admission_control_policy")
103 assert res.status_code == 200
104 assert res.json == {"GET ack": "pretend policy is here"}
107 def test_policy_get_unsupported(client, monkeypatch):
111 testing_helpers.patch_all(monkeypatch, nofetch=True)
112 res = client.get("/a1-p/policies/admission_control_policy")
113 assert res.status_code == 400
114 assert res.data == b'"POLICY DOES NOT SUPPORT FETCHING"\n'
117 def test_xapp_put_good(client, monkeypatch):
118 """ test policy put good"""
119 _test_put_patch(monkeypatch)
120 monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch))
121 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
122 assert res.status_code == 200
123 assert res.json == {"status": "SUCCESS", "foo": "bar"}
126 def test_xapp_put_bad(client, monkeypatch):
127 """Test policy put fails"""
128 _test_put_patch(monkeypatch)
129 # return from policy handler has a status indicating FAIL
131 "a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"status": "FAIL", "foo": "bar"})
133 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
134 assert res.status_code == 502
135 assert res.json["reason"] == "BAD STATUS"
136 assert res.json["return_payload"] == {"status": "FAIL", "foo": "bar"}
138 # return from policy handler has no status field
139 monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"foo": "bar"}))
140 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
141 assert res.status_code == 502
142 assert res.json["reason"] == "NO STATUS"
143 assert res.json["return_payload"] == {"foo": "bar"}
145 # return from policy handler not a json
147 "a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload="booger", jsonb=False)
149 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
150 assert res.status_code == 502
151 assert res.json["reason"] == "NOT JSON"
152 assert res.json["return_payload"] == "booger"
155 monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_type=666))
156 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
157 assert res.status_code == 504
158 assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
161 monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_state=666))
162 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
163 assert res.status_code == 504
164 assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
167 def test_xapp_put_bad_send(client, monkeypatch):
169 Test bad send failures
171 testing_helpers.patch_all(monkeypatch)
173 monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch))
174 res = client.put("/a1-p/policies/admission_control_policy", json={"not": "expected"})
175 assert res.status_code == 400
177 monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(10))
178 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
179 assert res.status_code == 504
180 assert res.data == b'"A1 was unable to send a needed message to a downstream subscriber"\n'
182 monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(5))
183 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
184 assert res.status_code == 504
185 assert res.data == b'"A1 was unable to send a needed message to a downstream subscriber"\n'
188 def test_bad_requests(client, monkeypatch):
189 """Test bad requests"""
190 testing_helpers.patch_all(monkeypatch)
193 res = client.put("/a1-p/policies/noexist", json=testing_helpers.good_payload())
194 assert res.status_code == 404
197 res = client.put("/a1-p/policies/admission_control_policy", data="notajson")
198 assert res.status_code == 415
200 # test a PUT body against a poliucy not expecting one
201 res = client.put("/a1-p/policies/test_policy", json=testing_helpers.good_payload())
202 assert res.status_code == 400
203 assert res.data == b'"BODY SUPPLIED BUT POLICY HAS NO EXPECTED BODY"\n'
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'
221 def test_missing_rmr(client, monkeypatch):
223 test that we get a 500 with an approrpiate message on a missing rmr rmr_string
225 testing_helpers.patch_all(monkeypatch, nonexisting_rmr=True)
226 res = client.put("/a1-p/policies/admission_control_policy", json=testing_helpers.good_payload())
227 assert res.status_code == 500
228 assert res.data == b'"A1 does not have a mapping for the desired rmr string. report this!"\n'
231 def test_healthcheck(client):
235 res = client.get("/a1-p/healthcheck")
236 assert res.status_code == 200