Update opMulti operations of the filter
[pti/o2.git] / tests / integration / test_ocloud_route.py
1 # Copyright (C) 2021 Wind River Systems, Inc.
2 #
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
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
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.
14
15 import uuid
16 import pytest
17
18 from o2common.config import config
19 from o2ims.domain import ocloud
20 from o2ims.domain import resource_type as rt
21
22
23 pytestmark = pytest.mark.usefixtures("mappers")
24
25
26 def setup_ocloud():
27     ocloudid1 = str(uuid.uuid4())
28     ocloud1 = ocloud.Ocloud(
29         ocloudid1, "ocloud1", config.get_api_url(),
30         "ocloud 1 for integration test", 1)
31     return ocloud1
32
33
34 def test_route_olcouds(sqlite_flask_uow):
35     uow, app = sqlite_flask_uow
36
37     with uow:
38         ocloud1 = setup_ocloud()
39         ocloud1_UUID = ocloud1.oCloudId
40         uow.oclouds.add(ocloud1)
41         uow.commit()
42
43     with app.test_client() as c:
44         apibase = config.get_o2ims_api_base()
45         resp = c.get(apibase+"/")
46         assert resp.status_code == 200
47         assert ocloud1_UUID.encode() in resp.data
48
49
50 def test_route_resource_types(sqlite_flask_uow):
51     uow, app = sqlite_flask_uow
52
53     with uow:
54         ocloud1_id = str(uuid.uuid4())
55         resource_type_id1 = str(uuid.uuid4())
56         resource_type1 = ocloud.ResourceType(
57             resource_type_id1, "resourcetype1", rt.ResourceTypeEnum.PSERVER,
58             ocloud1_id)
59         uow.resource_types.add(resource_type1)
60         uow.commit()
61
62     with app.test_client() as c:
63         apibase = config.get_o2ims_api_base()
64         resp = c.get(apibase+"/resourceTypes")
65         assert resp.status_code == 200
66         assert resource_type_id1.encode() in resp.data
67
68         resp = c.get(apibase+'/resourceTypes/'+resource_type_id1)
69         assert resp.status_code == 200
70         json_data = resp.get_json()
71         assert resource_type_id1 in json_data['resourceTypeId']
72
73
74 def test_route_resource_pools(sqlite_flask_uow):
75     uow, app = sqlite_flask_uow
76
77     with uow:
78         ocloud1_id = str(uuid.uuid4())
79         resource_pool_id1 = str(uuid.uuid4())
80         resource_pool1 = ocloud.ResourcePool(
81             resource_pool_id1, "resourcepool1", config.get_api_url(),
82             ocloud1_id)
83         uow.resource_pools.add(resource_pool1)
84         uow.commit()
85
86     with app.test_client() as c:
87         apibase = config.get_o2ims_api_base()
88         resp = c.get(apibase+"/resourcePools")
89         assert resp.status_code == 200
90         assert resource_pool_id1.encode() in resp.data
91
92         resp = c.get(apibase+'/resourcePools/'+resource_pool_id1)
93         assert resp.status_code == 200
94         json_data = resp.get_json()
95         assert resource_pool_id1 in json_data['resourcePoolId']
96
97
98 def test_route_resources(sqlite_flask_uow):
99     uow, app = sqlite_flask_uow
100
101     with uow:
102         resource_id1 = str(uuid.uuid4())
103         resource_type_id1 = str(uuid.uuid4())
104         resource_pool_id1 = str(uuid.uuid4())
105         resource1 = ocloud.Resource(
106             resource_id1, resource_type_id1, resource_pool_id1, 'resource1')
107         uow.resources.add(resource1)
108         uow.commit()
109
110     with app.test_client() as c:
111         apibase = config.get_o2ims_api_base()
112         resp = c.get(apibase+"/resourcePools/"+resource_pool_id1+"/resources")
113         assert resp.status_code == 200
114         assert resource_id1.encode() in resp.data
115
116         resp = c.get(apibase+"/resourcePools/"+resource_pool_id1 +
117                      "/resources/" + resource_id1)
118         assert resp.status_code == 200
119         json_data = resp.get_json()
120         assert resource_pool_id1 in json_data['resourcePoolId']
121         assert resource_type_id1 in json_data['resourceTypeId']
122         assert resource_id1 in json_data['resourceId']
123
124
125 def test_route_deployment_managers(sqlite_flask_uow):
126     uow, app = sqlite_flask_uow
127
128     with uow:
129         ocloud_id1 = str(uuid.uuid4())
130         deployment_manager_id1 = str(uuid.uuid4())
131         deployment_manager1 = ocloud.DeploymentManager(
132             deployment_manager_id1, "k8s1", ocloud_id1,
133             config.get_api_url()+"/k8s1")
134         uow.deployment_managers.add(deployment_manager1)
135         uow.commit()
136
137     with app.test_client() as c:
138         apibase = config.get_o2ims_api_base()
139         resp = c.get(apibase+"/deploymentManagers")
140         assert resp.status_code == 200
141         assert deployment_manager_id1.encode() in resp.data
142
143         resp = c.get(apibase+'/deploymentManagers/'+deployment_manager_id1)
144         assert resp.status_code == 200
145         json_data = resp.get_json()
146         assert deployment_manager_id1 in json_data['deploymentManagerId']
147
148
149 def test_route_subscriptions(sqlite_flask_uow):
150     _, app = sqlite_flask_uow
151
152     with app.test_client() as c:
153         apibase = config.get_o2ims_api_base()
154
155         sub_callback = 'http://subscription/callback/url'
156         resp = c.post(apibase+'/subscriptions', json={
157             'callback': sub_callback,
158             'consumerSubscriptionId': 'consumerSubId1',
159             'filter': 'empty'
160         })
161         assert resp.status_code == 201
162         json_data = resp.get_json()
163         assert 'subscriptionId' in json_data
164         subscriptionId1 = json_data['subscriptionId']
165
166         resp = c.get(apibase+'/subscriptions')
167         assert resp.status_code == 200
168         json_data = resp.get_json()
169         assert 1 == len(json_data)
170
171         resp = c.get(apibase+'/subscriptions/'+subscriptionId1)
172         assert resp.status_code == 200
173         json_data = resp.get_json()
174         assert sub_callback in json_data['callback']
175
176         resp = c.delete(apibase+'/subscriptions/'+subscriptionId1)
177         assert resp.status_code == 204
178
179         resp = c.get(apibase+'/subscriptions')
180         assert resp.status_code == 200
181         json_data = resp.get_json()
182         assert 0 == len(json_data)