Move integration tests to k8s
[ric-plt/a1.git] / 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 from rmr.rmr_mocks import rmr_mocks
20 from a1 import app
21 from a1 import exceptions
22 from rmr import rmr
23 import testing_helpers
24 import pytest
25
26 # http://flask.pocoo.org/docs/1.0/testing/
27 @pytest.fixture
28 def client():
29     db_fd, app.app.config["DATABASE"] = tempfile.mkstemp()
30     app.app.config["TESTING"] = True
31     cl = app.app.test_client()
32
33     yield cl
34
35     os.close(db_fd)
36     os.unlink(app.app.config["DATABASE"])
37
38
39 def _fake_dequeue(
40     monkeypatch,
41     msg_payload={"status": "SUCCESS", "foo": "bar"},
42     msg_type=20001,
43     msg_state=0,
44     jsonb=True,
45     unexpected_first=True,
46 ):
47     """
48     generates a mock rmr message response (returns a function that does; uses closures to set params)
49     """
50     new_messages = []
51     # stick a message we don't want at the front of the queue, then stick the message we want
52     if unexpected_first:
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)
58
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)
64
65     def f():
66         return new_messages
67
68     return f
69
70
71 # Actual Tests
72
73
74 def _test_put_patch(monkeypatch):
75     testing_helpers.patch_all(monkeypatch)
76     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(0))  # good sends for this whole batch
77
78     # we need to repatch alloc (already patched in patch_rmr) to fix the transactionid, alloc is called in send and recieve
79     def fake_alloc(_unused, _alsounused):
80         sbuf = rmr_mocks.Rmr_mbuf_t()
81         sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
82         return sbuf
83
84     # we also need to repatch set, since in the send function, we alloc, then set a new transid
85     def fake_set_transactionid(sbuf):
86         sbuf.contents.xaction = b"d49b53e478b711e9a1130242ac110002"
87
88     # Note, we could have just patched summary, but this patches at a "lower level" so is a better test
89     monkeypatch.setattr("rmr.rmr.rmr_alloc_msg", fake_alloc)
90     monkeypatch.setattr("rmr.rmr.generate_and_set_transaction_id", fake_set_transactionid)
91
92
93 def test_xapp_put_bad(client, monkeypatch):
94     """Test policy put fails"""
95     _test_put_patch(monkeypatch)
96     # return from policy handler has a status indicating FAIL
97     monkeypatch.setattr(
98         "a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"status": "FAIL", "foo": "bar"})
99     )
100     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
101     assert res.status_code == 502
102     assert res.json["reason"] == "BAD STATUS"
103     assert res.json["return_payload"] == {"status": "FAIL", "foo": "bar"}
104
105     # return from policy handler has no status field
106     monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload={"foo": "bar"}))
107     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
108     assert res.status_code == 502
109     assert res.json["reason"] == "NO STATUS"
110     assert res.json["return_payload"] == {"foo": "bar"}
111
112     # return from policy handler not a json
113     monkeypatch.setattr(
114         "a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_payload="booger", jsonb=False)
115     )
116     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
117     assert res.status_code == 502
118     assert res.json["reason"] == "NOT JSON"
119     assert res.json["return_payload"] == "booger"
120
121     # bad type
122     monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_type=666))
123     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
124     assert res.status_code == 504
125     assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
126
127     # bad state
128     monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch, msg_state=666))
129     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
130     assert res.status_code == 504
131     assert res.data == b"\"A1 was expecting an ACK back but it didn't receive one or didn't recieve the expected ACK\"\n"
132
133
134 def test_xapp_put_good(client, monkeypatch):
135     """ test policy put good"""
136     _test_put_patch(monkeypatch)
137
138     # do a good one
139     monkeypatch.setattr("a1.a1rmr._dequeue_all_waiting_messages", _fake_dequeue(monkeypatch))
140     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
141     assert res.status_code == 200
142     assert res.json == {"status": "SUCCESS", "foo": "bar"}
143
144
145 def test_xapp_put_bad_send(client, monkeypatch):
146     """
147     Test bad send failures
148     """
149     testing_helpers.patch_all(monkeypatch)
150
151     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(10))
152     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
153     assert res.status_code == 504
154     assert res.data == b'"A1 was unable to send a needed message to a downstream subscriber"\n'
155
156     monkeypatch.setattr("rmr.rmr.rmr_send_msg", rmr_mocks.send_mock_generator(5))
157     res = client.put("/ric/policies/control_admission_time", json=testing_helpers.good_payload())
158     assert res.status_code == 504
159     assert res.data == b'"A1 was unable to send a needed message to a downstream subscriber"\n'
160
161
162 def test_bad_requests(client, monkeypatch):
163     """Test bad requests"""
164     testing_helpers.patch_all(monkeypatch)
165
166     # test a 404
167     res = client.put("/ric/policies/noexist", json=testing_helpers.good_payload())
168     assert res.status_code == 404
169
170     # bad media type
171     res = client.put("/ric/policies/control_admission_time", data="notajson")
172     assert res.status_code == 415
173
174     # test a PUT body against a poliucy not expecting one
175     res = client.put("/ric/policies/test_policy", json=testing_helpers.good_payload())
176     assert res.status_code == 400
177     assert res.data == b'"BODY SUPPLIED BUT POLICY HAS NO EXPECTED BODY"\n'
178
179
180 def test_missing_manifest(client, monkeypatch):
181     """
182     test that we get a 500 with an approrpiate message on a missing manifest
183     """
184
185     def f():
186         raise exceptions.MissingManifest()
187
188     monkeypatch.setattr("a1.utils.get_ric_manifest", f)
189
190     res = client.put(
191         "/ric/policies/control_admission_time", json=testing_helpers.good_payload(), headers={"Content-Type": "text/plain"}
192     )
193     assert res.status_code == 500
194     assert res.data == b'"A1 was unable to find the required RIC manifest. report this!"\n'
195
196
197 def test_missing_rmr(client, monkeypatch):
198     """
199     test that we get a 500 with an approrpiate message on a missing rmr rmr_string
200     """
201     testing_helpers.patch_all(monkeypatch, nonexisting_rmr=True)
202     res = client.put(
203         "/ric/policies/control_admission_time", json=testing_helpers.good_payload(), headers={"Content-Type": "text/plain"}
204     )
205     assert res.status_code == 500
206     assert res.data == b'"A1 does not have a mapping for the desired rmr string. report this!"\n'