move heal test json
[vfc/nfvo/lcm.git] / lcm / ns / tests / test_query_subscriptions.py
1 # Copyright (c) 2019, CMCC Technologies Co., Ltd.
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 json
16 from django.test import TestCase
17 from rest_framework import status
18 from rest_framework.test import APIClient
19 from lcm.pub.database.models import SubscriptionModel
20 from lcm.ns.tests import SUBSCRIPTION_DICT
21
22
23 class TestQuerySubscriptions(TestCase):
24     def setUp(self):
25         self.client = APIClient()
26         self.test_single_subscription = SUBSCRIPTION_DICT.copy()
27         self.subscription_id = "99442b18-a5c7-11e8-998c-bf1755941f16"
28         self.ns_instance_id = "cd552c9c-ab6f-11e8-b354-236c32aa91a1"
29         SubscriptionModel.objects.all().delete()
30
31     def tearDown(self):
32         pass
33
34     def test_get_subscriptions(self):
35         ns_instance_filter = {
36             "nsdIds": [],
37             "nsInstanceIds": [self.ns_instance_id],
38             "nsInstanceNames": []
39         }
40         links = {
41             "self": {
42                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
43             }
44         }
45         SubscriptionModel(
46             subscription_id=self.subscription_id,
47             callback_uri="http://aurl.com",
48             auth_info="{}",
49             notification_types="['NsLcmOperationOccurrenceNotification']",
50             operation_types="['INSTANTIATE']",
51             operation_states="['STARTING']",
52             links=json.dumps(links),
53             ns_instance_filter=json.dumps(ns_instance_filter)).save()
54         response = self.client.get("/api/nslcm/v1/subscriptions", format='json')
55         self.assertEqual(response.status_code, status.HTTP_200_OK)
56         self.assertEqual([self.test_single_subscription], response.data)
57
58     def test_get_subscriptions_with_ns_instance_id(self):
59         ns_instance_filter = {
60             "nsdIds": [],
61             "nsInstanceIds": [self.ns_instance_id],
62             "nsInstanceNames": []
63         }
64         links = {
65             "self": {
66                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
67             }
68         }
69         SubscriptionModel(
70             subscription_id=self.subscription_id,
71             callback_uri="http://aurl.com",
72             auth_info="{}",
73             notification_types="['NsLcmOperationOccurrenceNotification']",
74             operation_types="['INSTANTIATE']",
75             operation_states="['STARTING']",
76             links=json.dumps(links),
77             ns_instance_filter=json.dumps(ns_instance_filter)).save()
78         dummy_ns_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
79         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
80         ns_instance_filter["nsInstanceIds"].append(dummy_ns_id)
81         SubscriptionModel(
82             subscription_id=dummy_subscription_id,
83             callback_uri="http://aurl.com",
84             auth_info="{}",
85             notification_types="['NsLcmOperationOccurrenceNotification']",
86             operation_types="['INSTANTIATE']",
87             operation_states="['STARTING']",
88             links=json.dumps(links),
89             ns_instance_filter=json.dumps(ns_instance_filter)).save()
90         response = self.client.get("/api/nslcm/v1/subscriptions?nsInstanceId=" + dummy_ns_id, format='json')
91         expected_response = self.test_single_subscription.copy()
92         expected_response["id"] = dummy_subscription_id
93         expected_response["filter"]["nsInstanceSubscriptionFilter"]["nsInstanceIds"] = ns_instance_filter["nsInstanceIds"]
94         self.assertEqual(response.status_code, status.HTTP_200_OK)
95         self.assertEqual([expected_response], response.data)
96
97     def test_get_subscriptions_with_unknown_ns_instance_id(self):
98         ns_instance_filter = {
99             "nsdIds": [],
100             "nsInstanceIds": [self.ns_instance_id],
101             "nsInstanceNames": []
102         }
103         links = {
104             "self": {
105                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
106             }
107         }
108         SubscriptionModel(
109             subscription_id=self.subscription_id,
110             callback_uri="http://aurl.com",
111             auth_info="{}",
112             notification_types="['NsLcmOperationOccurrenceNotification']",
113             operation_types="['INSTANTIATE']",
114             operation_states="['STARTING']",
115             links=json.dumps(links),
116             ns_instance_filter=json.dumps(ns_instance_filter)).save()
117         response = self.client.get("/api/nslcm/v1/subscriptions?nsInstanceId=dummy", format='json')
118         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
119
120     def test_get_subscriptions_with_invalid_filter(self):
121         ns_instance_filter = {
122             "nsdIds": [],
123             "nsInstanceIds": [self.ns_instance_id],
124             "nsInstanceNames": []
125         }
126         links = {
127             "self": {
128                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
129             }
130         }
131         SubscriptionModel(
132             subscription_id=self.subscription_id,
133             callback_uri="http://aurl.com",
134             auth_info="{}",
135             notification_types="['NsLcmOperationOccurrenceNotification']",
136             operation_types="['INSTANTIATE']",
137             operation_states="['STARTING']",
138             links=json.dumps(links),
139             ns_instance_filter=json.dumps(ns_instance_filter)).save()
140         response = self.client.get("/api/nslcm/v1/subscriptions?dummy=dummy", format='json')
141         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
142
143     def test_get_subscriptions_with_operation_type_filter(self):
144         ns_instance_filter = {
145             "nsdIds": [],
146             "nsInstanceIds": [self.ns_instance_id],
147             "nsInstanceNames": []
148         }
149         links = {
150             "self": {
151                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
152             }
153         }
154         SubscriptionModel(
155             subscription_id=self.subscription_id,
156             callback_uri="http://aurl.com",
157             auth_info="{}",
158             notification_types="['NsLcmOperationOccurrenceNotification']",
159             operation_types="['INSTANTIATE']",
160             operation_states="['STARTING']",
161             links=json.dumps(links),
162             ns_instance_filter=json.dumps(ns_instance_filter)).save()
163         dummy_ns_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
164         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
165         ns_instance_filter["nsInstanceIds"].append(dummy_ns_id)
166         SubscriptionModel(
167             subscription_id=dummy_subscription_id,
168             callback_uri="http://aurl.com",
169             auth_info="{}",
170             notification_types="['NsLcmOperationOccurrenceNotification']",
171             operation_types="['SCALE']",
172             operation_states="['STARTING']",
173             links=json.dumps(links),
174             ns_instance_filter=json.dumps(ns_instance_filter)).save()
175         response = self.client.get("/api/nslcm/v1/subscriptions?operationTypes=SCALE", format='json')
176         expected_response = self.test_single_subscription.copy()
177         expected_response["id"] = dummy_subscription_id
178         expected_response["filter"]["nsInstanceSubscriptionFilter"]["nsInstanceIds"] = ns_instance_filter["nsInstanceIds"]
179         expected_response["filter"]["operationTypes"] = ["SCALE"]
180         self.assertEqual(response.status_code, status.HTTP_200_OK)
181         self.assertEqual([expected_response], response.data)