Add subscriptions API to GVNFM
[vfc/gvnfm/vnflcm.git] / lcm / lcm / nf / tests / test_subscribe_notification.py
1 # Copyright (C) 2018 Verizon. All Rights Reserved
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 mock
16 from django.test import TestCase
17 from rest_framework.test import APIClient
18 import uuid
19
20
21 class TestSubscription(TestCase):
22     def setUp(self):
23         self.client = APIClient()
24
25     def tearDown(self):
26         pass
27
28     @mock.patch("requests.get")
29     @mock.patch.object(uuid, 'uuid4')
30     def test_subscribe_notification_simple(self, mock_uuid4, mock_requests):
31         temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
32         dummy_subscription = {
33             "callbackUri": "http://aurl.com"
34         }
35         mock_requests.return_value.status_code = 204
36         mock_requests.get.status_code = 204
37         mock_uuid4.return_value = temp_uuid
38         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
39         self.assertEqual(dummy_subscription["callbackUri"], response.data["callbackUri"])
40         self.assertEqual(temp_uuid, response.data["id"])
41
42     @mock.patch("requests.get")
43     @mock.patch.object(uuid, 'uuid4')
44     def test_subscribe_notification(self, mock_uuid4, mock_requests):
45         temp_uuid = "99442b18-a5c7-11e8-998c-bf1755941f13"
46         dummy_subscription = {
47             "callbackUri": "http://aurl.com",
48             "authentication": {
49                 "authType": ["BASIC"],
50                 "paramsBasic": {
51                     "username": "username",
52                     "password": "password"
53                 }
54             },
55             "filter": {
56                 "notificationTypes": ["VnfLcmOperationOccurrenceNotification"],
57                 "operationTypes": [
58                     "INSTANTIATE"
59                 ],
60                 "operationStates": [
61                     "STARTING"
62                 ],
63             }
64         }
65         mock_requests.return_value.status_code = 204
66         mock_requests.get.return_value.status_code = 204
67         mock_uuid4.return_value = temp_uuid
68         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
69         self.assertEqual(dummy_subscription["callbackUri"], response.data["callbackUri"])
70         self.assertEqual(temp_uuid, response.data["id"])
71
72     @mock.patch("requests.get")
73     def test_invalid_auth_subscription(self, mock_requests):
74         dummy_subscription = {
75             "callbackUri": "http://aurl.com",
76             "authentication": {
77                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
78                 "paramsBasic": {
79                     "username": "username",
80                     "password": "password"
81                 }
82             },
83             "filter": {
84                 "notificationTypes": ["VnfLcmOperationOccurrenceNotification"],
85                 "operationTypes": [
86                     "INSTANTIATE"
87                 ],
88                 "operationStates": [
89                     "STARTING"
90                 ],
91             }
92         }
93         mock_requests.return_value.status_code = 204
94         mock_requests.get.return_value.status_code = 204
95         expected_data = {
96             'error': 'Auth type should be BASIC'
97         }
98         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
99         self.assertEqual(expected_data, response.data)
100
101     @mock.patch("requests.get")
102     def test_invalid_notification_type(self, mock_requests):
103         dummy_subscription = {
104             "callbackUri": "http://aurl.com",
105             "filter": {
106                 "notificationTypes": ["VnfIdentifierDeletionNotification"],
107                 "operationTypes": [
108                     "INSTANTIATE"
109                 ],
110                 "operationStates": [
111                     "STARTING"
112                 ],
113             }
114         }
115         mock_requests.return_value.status_code = 204
116         mock_requests.get.return_value.status_code = 204
117         expected_data = {
118             'error': 'If you are setting operationTypes,then ' +
119             'notificationTypes must be VnfLcmOperationOccurrenceNotification'
120         }
121         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
122         self.assertEqual(expected_data, response.data)
123
124     @mock.patch("requests.get")
125     @mock.patch.object(uuid, 'uuid4')
126     def test_duplicate_subscription(self, mock_uuid4, mock_requests):
127         temp_uuid = str(uuid.uuid4())
128         dummy_subscription = {
129             "callbackUri": "http://aurl.com",
130             "filter": {
131                 "notificationTypes": ["VnfLcmOperationOccurrenceNotification"],
132                 "operationTypes": [
133                     "INSTANTIATE"
134                 ],
135                 "operationStates": [
136                     "STARTING"
137                 ]
138             }
139         }
140         mock_requests.return_value.status_code = 204
141         mock_requests.get.return_value.status_code = 204
142         mock_uuid4.return_value = temp_uuid
143         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
144         self.assertEqual(dummy_subscription["callbackUri"], response.data["callbackUri"])
145         self.assertEqual(temp_uuid, response.data["id"])
146         response = self.client.post("/api/vnflcm/v1/subscriptions", data=dummy_subscription, format='json')
147         self.assertEqual(303, response.status_code)
148         expected_data = {
149             "error": "Already Subscription exists with the same callbackUri and filter"
150         }
151         self.assertEqual(expected_data, response.data)