1 # Copyright (C) 2021 Wind River Systems, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from unittest.mock import MagicMock
18 from o2ims.domain import configuration_obj
19 from o2ims.views import provision_view
20 from o2common.config import config
23 def test_new_smo_endpoint():
24 configuration_id1 = str(uuid.uuid4())
25 configuration1 = configuration_obj.Configuration(
26 configuration_id1, "https://callback/uri/write/here",
28 assert configuration_id1 is not None and\
29 configuration1.configurationId == configuration_id1
32 def test_view_smo_endpoint(mock_uow):
33 session, uow = mock_uow
35 configuration_id1 = str(uuid.uuid4())
37 conf1.serialize_smo.return_value = {
38 "id": configuration_id1,
40 session.return_value.query.return_value = [conf1]
42 configuration_list = provision_view.configurations(uow)
43 assert str(configuration_list[0].get(
44 "id")) == configuration_id1
47 def test_view_smo_endpoint_one(mock_uow):
48 session, uow = mock_uow
50 configuration_id1 = str(uuid.uuid4())
51 session.return_value.query.return_value.filter_by.return_value.first.\
52 return_value.serialize_smo.return_value = None
55 configuration_res = provision_view.configuration_one(
56 configuration_id1, uow)
57 assert configuration_res is None
59 session.return_value.query.return_value.filter_by.return_value.first.\
60 return_value.serialize_smo.return_value = {
61 "id": configuration_id1,
64 configuration_res = provision_view.configuration_one(
65 configuration_id1, uow)
66 assert str(configuration_res.get(
67 "id")) == configuration_id1
70 def test_flask_get_list(mock_flask_uow):
71 session, app = mock_flask_uow
72 session.query.return_value = []
73 apibase = config.get_provision_api_base()
75 with app.test_client() as client:
76 # Get list and return empty list
77 ##########################
78 resp = client.get(apibase+"/smo-endpoint")
79 assert resp.get_data() == b'[]\n'
82 def test_flask_get_one(mock_flask_uow):
83 session, app = mock_flask_uow
85 session.return_value.query.return_value.filter_by.return_value.\
86 first.return_value = None
87 apibase = config.get_provision_api_base()
89 with app.test_client() as client:
90 # Get one and return 404
91 ###########################
92 configuration_id1 = str(uuid.uuid4())
93 resp = client.get(apibase+"/smo-endpoint/"+configuration_id1)
94 assert resp.status_code == 404
97 def test_flask_post(mock_flask_uow):
98 session, app = mock_flask_uow
99 apibase = config.get_provision_api_base()
101 with app.test_client() as client:
102 session.return_value.execute.return_value = []
104 conf_callback = 'http://registration/callback/url'
105 resp = client.post(apibase+'/smo-endpoint', json={
106 'endpoint': conf_callback
108 assert resp.status_code == 201
109 assert 'id' in resp.get_json()
112 def test_flask_delete(mock_flask_uow):
113 session, app = mock_flask_uow
114 apibase = config.get_provision_api_base()
116 with app.test_client() as client:
117 session.return_value.execute.return_value.first.return_value = {}
119 configuration_id1 = str(uuid.uuid4())
120 resp = client.delete(apibase+"/smo-endpoint/"+configuration_id1)
121 assert resp.status_code == 204
124 def test_flask_not_allowed(mock_flask_uow):
125 _, app = mock_flask_uow
126 apibase = config.get_provision_api_base()
128 with app.test_client() as client:
130 # Testing SMO endpoint not support method
131 ##########################
132 uri = apibase + "/smo-endpoint"
133 resp = client.put(uri)
134 assert resp.status == '405 METHOD NOT ALLOWED'
135 resp = client.patch(uri)
136 assert resp.status == '405 METHOD NOT ALLOWED'
137 resp = client.delete(uri)
138 assert resp.status == '405 METHOD NOT ALLOWED'
140 configuration_id1 = str(uuid.uuid4())
141 uri = apibase + "/smo-endpoint/" + configuration_id1
142 resp = client.post(uri)
143 assert resp.status == '405 METHOD NOT ALLOWED'
144 resp = client.put(uri)
145 assert resp.status == '405 METHOD NOT ALLOWED'
146 resp = client.patch(uri)
147 assert resp.status == '405 METHOD NOT ALLOWED'