tests.test_core.test_auth.TestAuth.test_06_delete_user_request() - Code Metrics - Inspection of "Create Authentication module" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#966)

by Gleyberson

created 2019-12-03 14:34 UTC

TestAuth.test_06_delete_user_request() A

↳ Parent: tests.test_core.test_auth

Complexity

Conditions

Size

Total Lines	11
Code Lines	8

Duplication

Lines	11
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	8
nop	2
dl	11
loc	11
rs	10
c	0
b	0
f	0

"""Test kytos.core.auth module."""
import asyncio
import base64
import hashlib
from unittest import TestCase
from unittest.mock import Mock, patch

from kytos.core import Controller
from kytos.core.auth import Auth
from kytos.core.config import KytosConfig

KYTOS_CORE_API = "http://127.0.0.1:8181/api/kytos/"
API_URI = KYTOS_CORE_API+"core"
STOREHOUSE_API_URI = KYTOS_CORE_API+"storehouse/v1/kytos.core.auth.users"


# pylint: disable=unused-argument
class TestAuth(TestCase):
    """Auth tests."""

    def setUp(self):
        """Instantiate a controller and an Auth."""
        self.patched_events = []  # {'event_name': box_object}
        self.server_name_url = 'http://localhost:8181/api/kytos'
        self.controller = self._get_controller_mock()
        self.auth = Auth(self.controller)
        self.username, self.password = self._create_super_user()
        self.token = self._get_token()
        self.user_data = {
            "username": "authtempuser",
            "email": "[email protected]",
            "password": "password",
        }

    def _patch_event_trigger(self, event):
        """Patch event callback trigger."""
        for patched_event in self.patched_events:
            box = patched_event.get(event.content.get('callback').__name__)
            event.content.get('callback')(None, box, None)

    def _get_controller_mock(self):
        """Return a controller mock."""
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)
        options = KytosConfig().options['daemon']
        options.jwt_secret = 'jwt_secret'

        controller = Controller(options, loop=loop)
        controller.log = Mock()

        # Patch event callback trigger.
        controller.buffers.app.put = self._patch_event_trigger

        return controller

    @staticmethod
    def get_auth_test_client(auth):
        """Return a flask api test client."""
        auth.controller.api_server.register_napp_endpoints(auth)
        return auth.controller.api_server.app.test_client()

    @patch('kytos.core.auth.Auth._create_superuser')
    def _create_super_user(self, mock_username=None):
        """Create a superuser to integration test."""
        username = "test"
        password = "password"
        email = "[email protected]"

        mock_username.return_value.get_username.return_value = username
        mock_username.return_value.get_email.return_value = email
        self.auth._create_superuser()  # pylint: disable=protected-access

        return username, password

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def _get_token(self, mock_jwt_secret=None):
        """Make a request to get a token to be used in tests."""
        box = Mock()
        box.data = {
            # "password" digested
            'password': 'b109f3bbbc244eb82441917ed06d618b9008dd09b3befd1b5e073'
                        '94c706a8bb980b1d7785e5976ec049b46df5f1326af5a2ea6d103'
                        'fd07c95385ffab0cacbc86'
        }
        header = {
            "Authorization": "Basic "
            + base64.b64encode(
                bytes(self.username + ":" + self.password, "ascii")
            ).decode("ascii")
        }
        # Patch _find_user_callback event callback.
        self.patched_events.append({'_find_user_callback': box})
        url = "%s/auth/login/" % API_URI
        api = self.get_auth_test_client(self.auth)
        success_response = api.open(url, method='GET', headers=header)

        json_response = success_response.json
        return json_response["token"]

    def _validate_schema(self, my_dict, check_against):
        """Check if a dict respects a given schema."""
        for key, value in check_against.items():
            if isinstance(value, dict):
                return self._validate_schema(my_dict[key], value)
            if not isinstance(my_dict[key], value):
                return False
        return True

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_01_login_request(self, mock_jwt_secret):
        """Test auth login endpoint."""
        valid_header = {
            "Authorization": "Basic "
            + base64.b64encode(
                bytes(self.username + ":" + self.password, "ascii")
            ).decode("ascii")
        }
        invalid_header = {
            "Authorization": "Basic "
            + base64.b64encode(
                bytes("nonexistent" + ":" + "nonexistent", "ascii")
            ).decode("ascii")
        }
        box = Mock()
        box.data = {
            # "password" digested
            'password': 'b109f3bbbc244eb82441917ed06d618b9008dd09b3befd1b5e073'
                        '94c706a8bb980b1d7785e5976ec049b46df5f1326af5a2ea6d103'
                        'fd07c95385ffab0cacbc86'
        }
        # Patch _find_user_callback event callback.
        self.patched_events.append({'_find_user_callback': box})
        url = "%s/auth/login/" % API_URI
        api = self.get_auth_test_client(self.auth)
        success_response = api.open(url, method='GET', headers=valid_header)
        error_response = api.open(url, method='GET', headers=invalid_header)

        self.assertEqual(success_response.status_code, 200)
        self.assertEqual(error_response.status_code, 401)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_02_list_users_request(self, mock_jwt_secret):
        """Test auth list users endpoint."""
        valid_header = {"Authorization": "Bearer %s" % self.token}
        invalid_header = {"Authorization": "Bearer invalidtoken"}
        schema = {"users": list}
        password = "password".encode()
        # Patch _list_users_callback event callback.
        event_boxes = [self.user_data,
                       {"username": "authtempuser2",
                        "email": "[email protected]",
                        "password": hashlib.sha512(password).hexdigest()}]
        self.patched_events.append({'_list_users_callback': event_boxes})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/" % API_URI
        success_response = api.open(url, method='GET', headers=valid_header)
        error_response = api.open(url, method='GET', headers=invalid_header)
        is_valid = self._validate_schema(success_response.json, schema)

        self.assertEqual(success_response.status_code, 200)
        self.assertEqual(error_response.status_code, 401)
        self.assertTrue(is_valid)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")

    def test_03_create_user_request(self, mock_jwt_secret):
        """Test auth create user endpoint."""
        header = {"Authorization": "Bearer %s" % self.token}
        # Patch _create_user_callback event callback.
        self.patched_events.append({'_create_user_callback': self.user_data})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/" % API_URI
        success_response = api.open(url, method='POST', json=self.user_data,
                                    headers=header)

        self.assertEqual(success_response.status_code, 200)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_03_create_user_request_error(self, mock_jwt_secret):
        """Test auth create user endpoint."""
        header = {"Authorization": "Bearer %s" % self.token}
        # Patch _create_user_callback event callback.
        self.patched_events.append({'_create_user_callback': None})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/" % API_URI
        error_response = api.open(url, method='POST', json=self.user_data,
                                  headers=header)

        self.assertEqual(error_response.status_code, 409)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_04_list_user_request(self, mock_jwt_secret):
        """Test auth list user endpoint."""
        valid_header = {"Authorization": "Bearer %s" % self.token}
        schema = {"data": {"email": str, "username": str}}
        box = Mock()
        box.data = self.user_data
        self.patched_events.append({'_find_user_callback': box})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
        success_response = api.open(url, method='GET', headers=valid_header)
        is_valid = self._validate_schema(success_response.json, schema)

        self.assertEqual(success_response.status_code, 200)
        self.assertTrue(is_valid)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")

    def test_04_list_user_request_error(self, mock_jwt_secret):
        """Test auth list user endpoint."""
        valid_header = {"Authorization": "Bearer %s" % self.token}
        self.patched_events.append({'_find_user_callback': None})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, 'user3')
        error_response = api.open(url, method='GET', headers=valid_header)

        self.assertEqual(error_response.status_code, 404)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_05_update_user_request(self, mock_jwt_secret):
        """Test auth update user endpoint."""
        valid_header = {"Authorization": "Bearer %s" % self.token}
        data = {"email": "[email protected]"}
        self.patched_events.append({'_update_user_callback': data})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
        success_response = api.open(url, method='PATCH', json=data,
                                    headers=valid_header)

        self.assertEqual(success_response.status_code, 200)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")

    def test_05_update_user_request_error(self, mock_jwt_secret):
        """Test auth update user endpoint."""
        valid_header = {"Authorization": "Bearer %s" % self.token}
        self.patched_events.append({'_update_user_callback': None})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, 'user5')
        error_response = api.open(url, method='PATCH', json={},
                                  headers=valid_header)

        self.assertEqual(error_response.status_code, 404)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")

    def test_06_delete_user_request(self, mock_jwt_secret):
        """Test auth delete user endpoint."""
        header = {"Authorization": "Bearer %s" % self.token}
        # Patch _delete_user_callback event callback.
        self.patched_events.append({'_delete_user_callback': self.user_data})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
        success_response = api.open(url, method='DELETE', headers=header)

        self.assertEqual(success_response.status_code, 200)

    @patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
    def test_06_delete_user_request_error(self, mock_jwt_secret):
        """Test auth delete user endpoint."""
        header = {"Authorization": "Bearer %s" % self.token}
        # Patch _delete_user_callback event callback.
        self.patched_events.append({'_delete_user_callback': None})
        api = self.get_auth_test_client(self.auth)
        url = "%s/auth/users/%s" % (API_URI, "nonexistent")
        success_response = api.open(url, method='DELETE', headers=header)

        self.assertEqual(success_response.status_code, 404)


1		"""Test kytos.core.auth module."""
2		import asyncio
3		import base64
4		import hashlib
5		from unittest import TestCase
6		from unittest.mock import Mock, patch
7
8		from kytos.core import Controller
9		from kytos.core.auth import Auth
10		from kytos.core.config import KytosConfig
11
12		KYTOS_CORE_API = "http://127.0.0.1:8181/api/kytos/"
13		API_URI = KYTOS_CORE_API+"core"
14		STOREHOUSE_API_URI = KYTOS_CORE_API+"storehouse/v1/kytos.core.auth.users"
15
16
17		# pylint: disable=unused-argument
18		class TestAuth(TestCase):
19		"""Auth tests."""
20
21		def setUp(self):
22		"""Instantiate a controller and an Auth."""
23		self.patched_events = [] # {'event_name': box_object}
24		self.server_name_url = 'http://localhost:8181/api/kytos'
25		self.controller = self._get_controller_mock()
26		self.auth = Auth(self.controller)
27		self.username, self.password = self._create_super_user()
28		self.token = self._get_token()
29		self.user_data = {
30		"username": "authtempuser",
31		"email": "[email protected]",
32		"password": "password",
33		}
34
35		def _patch_event_trigger(self, event):
36		"""Patch event callback trigger."""
37		for patched_event in self.patched_events:
38		box = patched_event.get(event.content.get('callback').__name__)
39		event.content.get('callback')(None, box, None)
40
41		def _get_controller_mock(self):
42		"""Return a controller mock."""
43		loop = asyncio.new_event_loop()
44		asyncio.set_event_loop(None)
45		options = KytosConfig().options['daemon']
46		options.jwt_secret = 'jwt_secret'
47
48		controller = Controller(options, loop=loop)
49		controller.log = Mock()
50
51		# Patch event callback trigger.
52		controller.buffers.app.put = self._patch_event_trigger
53
54		return controller
55
56		@staticmethod
57		def get_auth_test_client(auth):
58		"""Return a flask api test client."""
59		auth.controller.api_server.register_napp_endpoints(auth)
60		return auth.controller.api_server.app.test_client()
61
62		@patch('kytos.core.auth.Auth._create_superuser')
63		def _create_super_user(self, mock_username=None):
64		"""Create a superuser to integration test."""
65		username = "test"
66		password = "password"
67		email = "[email protected]"
68
69		mock_username.return_value.get_username.return_value = username
70		mock_username.return_value.get_email.return_value = email
71		self.auth._create_superuser() # pylint: disable=protected-access
72
73		return username, password
74
75		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
76		def _get_token(self, mock_jwt_secret=None):
77		"""Make a request to get a token to be used in tests."""
78		box = Mock()
79		box.data = {
80		# "password" digested
81		'password': 'b109f3bbbc244eb82441917ed06d618b9008dd09b3befd1b5e073'
82		'94c706a8bb980b1d7785e5976ec049b46df5f1326af5a2ea6d103'
83		'fd07c95385ffab0cacbc86'
84		}
85		header = {
86		"Authorization": "Basic "
87		+ base64.b64encode(
88		bytes(self.username + ":" + self.password, "ascii")
89		).decode("ascii")
90		}
91		# Patch _find_user_callback event callback.
92		self.patched_events.append({'_find_user_callback': box})
93		url = "%s/auth/login/" % API_URI
94		api = self.get_auth_test_client(self.auth)
95		success_response = api.open(url, method='GET', headers=header)
96
97		json_response = success_response.json
98		return json_response["token"]
99
100		def _validate_schema(self, my_dict, check_against):
101		"""Check if a dict respects a given schema."""
102		for key, value in check_against.items():
103		if isinstance(value, dict):
104		return self._validate_schema(my_dict[key], value)
105		if not isinstance(my_dict[key], value):
106		return False
107		return True
108
109		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
110		def test_01_login_request(self, mock_jwt_secret):
111		"""Test auth login endpoint."""
112		valid_header = {
113		"Authorization": "Basic "
114		+ base64.b64encode(
115		bytes(self.username + ":" + self.password, "ascii")
116		).decode("ascii")
117		}
118		invalid_header = {
119		"Authorization": "Basic "
120		+ base64.b64encode(
121		bytes("nonexistent" + ":" + "nonexistent", "ascii")
122		).decode("ascii")
123		}
124		box = Mock()
125		box.data = {
126		# "password" digested
127		'password': 'b109f3bbbc244eb82441917ed06d618b9008dd09b3befd1b5e073'
128		'94c706a8bb980b1d7785e5976ec049b46df5f1326af5a2ea6d103'
129		'fd07c95385ffab0cacbc86'
130		}
131		# Patch _find_user_callback event callback.
132		self.patched_events.append({'_find_user_callback': box})
133		url = "%s/auth/login/" % API_URI
134		api = self.get_auth_test_client(self.auth)
135		success_response = api.open(url, method='GET', headers=valid_header)
136		error_response = api.open(url, method='GET', headers=invalid_header)
137
138		self.assertEqual(success_response.status_code, 200)
139		self.assertEqual(error_response.status_code, 401)
140
141		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
142		def test_02_list_users_request(self, mock_jwt_secret):
143		"""Test auth list users endpoint."""
144		valid_header = {"Authorization": "Bearer %s" % self.token}
145		invalid_header = {"Authorization": "Bearer invalidtoken"}
146		schema = {"users": list}
147		password = "password".encode()
148		# Patch _list_users_callback event callback.
149		event_boxes = [self.user_data,
150		{"username": "authtempuser2",
151		"email": "[email protected]",
152		"password": hashlib.sha512(password).hexdigest()}]
153		self.patched_events.append({'_list_users_callback': event_boxes})
154		api = self.get_auth_test_client(self.auth)
155		url = "%s/auth/users/" % API_URI
156		success_response = api.open(url, method='GET', headers=valid_header)
157		error_response = api.open(url, method='GET', headers=invalid_header)
158		is_valid = self._validate_schema(success_response.json, schema)
159
160		self.assertEqual(success_response.status_code, 200)
161		self.assertEqual(error_response.status_code, 401)
162		self.assertTrue(is_valid)
163
164	View Code Duplication	@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
		0 ignored issues – show Duplication introduced 2019-11-28 12:40 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
165		def test_03_create_user_request(self, mock_jwt_secret):
166		"""Test auth create user endpoint."""
167		header = {"Authorization": "Bearer %s" % self.token}
168		# Patch _create_user_callback event callback.
169		self.patched_events.append({'_create_user_callback': self.user_data})
170		api = self.get_auth_test_client(self.auth)
171		url = "%s/auth/users/" % API_URI
172		success_response = api.open(url, method='POST', json=self.user_data,
173		headers=header)
174
175		self.assertEqual(success_response.status_code, 200)
176
177		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
178		def test_03_create_user_request_error(self, mock_jwt_secret):
179		"""Test auth create user endpoint."""
180		header = {"Authorization": "Bearer %s" % self.token}
181		# Patch _create_user_callback event callback.
182		self.patched_events.append({'_create_user_callback': None})
183		api = self.get_auth_test_client(self.auth)
184		url = "%s/auth/users/" % API_URI
185		error_response = api.open(url, method='POST', json=self.user_data,
186		headers=header)
187
188		self.assertEqual(error_response.status_code, 409)
189
190		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
191		def test_04_list_user_request(self, mock_jwt_secret):
192		"""Test auth list user endpoint."""
193		valid_header = {"Authorization": "Bearer %s" % self.token}
194		schema = {"data": {"email": str, "username": str}}
195		box = Mock()
196		box.data = self.user_data
197		self.patched_events.append({'_find_user_callback': box})
198		api = self.get_auth_test_client(self.auth)
199		url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
200		success_response = api.open(url, method='GET', headers=valid_header)
201		is_valid = self._validate_schema(success_response.json, schema)
202
203		self.assertEqual(success_response.status_code, 200)
204		self.assertTrue(is_valid)
205
206	View Code Duplication	@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
		0 ignored issues – show Duplication introduced 2019-11-28 01:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
207		def test_04_list_user_request_error(self, mock_jwt_secret):
208		"""Test auth list user endpoint."""
209		valid_header = {"Authorization": "Bearer %s" % self.token}
210		self.patched_events.append({'_find_user_callback': None})
211		api = self.get_auth_test_client(self.auth)
212		url = "%s/auth/users/%s" % (API_URI, 'user3')
213		error_response = api.open(url, method='GET', headers=valid_header)
214
215		self.assertEqual(error_response.status_code, 404)
216
217		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
218		def test_05_update_user_request(self, mock_jwt_secret):
219		"""Test auth update user endpoint."""
220		valid_header = {"Authorization": "Bearer %s" % self.token}
221		data = {"email": "[email protected]"}
222		self.patched_events.append({'_update_user_callback': data})
223		api = self.get_auth_test_client(self.auth)
224		url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
225		success_response = api.open(url, method='PATCH', json=data,
226		headers=valid_header)
227
228		self.assertEqual(success_response.status_code, 200)
229
230	View Code Duplication	@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
		0 ignored issues – show Duplication introduced 2019-11-28 21:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
231		def test_05_update_user_request_error(self, mock_jwt_secret):
232		"""Test auth update user endpoint."""
233		valid_header = {"Authorization": "Bearer %s" % self.token}
234		self.patched_events.append({'_update_user_callback': None})
235		api = self.get_auth_test_client(self.auth)
236		url = "%s/auth/users/%s" % (API_URI, 'user5')
237		error_response = api.open(url, method='PATCH', json={},
238		headers=valid_header)
239
240		self.assertEqual(error_response.status_code, 404)
241
242	View Code Duplication	@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
		0 ignored issues – show Duplication introduced 2019-11-28 21:16 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
243		def test_06_delete_user_request(self, mock_jwt_secret):
244		"""Test auth delete user endpoint."""
245		header = {"Authorization": "Bearer %s" % self.token}
246		# Patch _delete_user_callback event callback.
247		self.patched_events.append({'_delete_user_callback': self.user_data})
248		api = self.get_auth_test_client(self.auth)
249		url = "%s/auth/users/%s" % (API_URI, self.user_data.get("username"))
250		success_response = api.open(url, method='DELETE', headers=header)
251
252		self.assertEqual(success_response.status_code, 200)
253
254		@patch('kytos.core.auth.Auth.get_jwt_secret', return_value="abc")
255		def test_06_delete_user_request_error(self, mock_jwt_secret):
256		"""Test auth delete user endpoint."""
257		header = {"Authorization": "Bearer %s" % self.token}
258		# Patch _delete_user_callback event callback.
259		self.patched_events.append({'_delete_user_callback': None})
260		api = self.get_auth_test_client(self.auth)
261		url = "%s/auth/users/%s" % (API_URI, "nonexistent")
262		success_response = api.open(url, method='DELETE', headers=header)
263
264		self.assertEqual(success_response.status_code, 404)
265

kytos / kytos

Pull Request — master (#966)

TestAuth.test_06_delete_user_request() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like