From fa3e30c518885afbf1030796b36a20095bc97fbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adolfo=20G=C3=B3mez=20Garc=C3=ADa?= Date: Sun, 21 Aug 2022 19:37:32 +0200 Subject: [PATCH] fixing tests --- server/src/tests/REST/test_initialize.py | 45 +++++++++++++++++------- server/src/tests/REST/test_test.py | 2 +- server/src/tests/fixtures/services.py | 23 ++++++------ server/src/tests/utils/rest/test.py | 22 +++++++++--- server/src/uds/REST/methods/actor_v3.py | 13 ++++--- 5 files changed, 71 insertions(+), 34 deletions(-) diff --git a/server/src/tests/REST/test_initialize.py b/server/src/tests/REST/test_initialize.py index 55d04b2d..cc2a5aeb 100644 --- a/server/src/tests/REST/test_initialize.py +++ b/server/src/tests/REST/test_initialize.py @@ -29,6 +29,7 @@ @author: Adolfo Gómez, dkmaster at dkmon dot com """ import typing +import functools import logging from django.conf import settings @@ -48,11 +49,11 @@ class ActorInitializeV3(rest.test.RESTTestCase): Test actor functionality """ - def invoke_success(self, token: str, mac: str) -> typing.Dict[str, typing.Any]: + def invoke_success(self, type_: typing.Union[typing.Literal['managed'], typing.Literal['unmanaged']], token: str, mac: str) -> typing.Dict[str, typing.Any]: response = self.client.post( '/uds/rest/actor/v3/initialize', data={ - 'type': 'managed', + 'type': type_, 'version': VERSION, 'token': token, 'id': [{'mac': mac, 'ip': '1.2.3.4'}] @@ -65,11 +66,11 @@ class ActorInitializeV3(rest.test.RESTTestCase): return data['result'] - def invoke_failure(self, token: str, mac: str, expectForbbiden: bool) -> typing.Dict[str, typing.Any]: + def invoke_failure(self, type_: typing.Union[typing.Literal['managed'], typing.Literal['unmanaged']], token: str, mac: str, expectForbbiden: bool) -> typing.Dict[str, typing.Any]: response = self.client.post( '/uds/rest/actor/v3/initialize', data={ - 'type': 'managed', + 'type': type_, 'version': VERSION, 'token': token, 'id': [{'mac': mac, 'ip': '4.3.2.1'}] @@ -86,18 +87,20 @@ class ActorInitializeV3(rest.test.RESTTestCase): def test_initialize_managed(self) -> None: """ - Test actor initialize v3 + Test actor initialize v3 for managed actor """ _, actor_token = self.login_and_register() # Get the user service unique_id - unique_id = self.user_service.getUniqueId() + unique_id = self.user_service_managed.getUniqueId() + success = functools.partial(self.invoke_success, 'managed', actor_token) + failure = functools.partial(self.invoke_failure, 'managed') - result = self.invoke_success(actor_token, unique_id) + result = success(unique_id) # Ensure own token is assigned - self.assertEqual(result['own_token'], self.user_service.uuid) + self.assertEqual(result['own_token'], self.user_service_managed.uuid) # Ensure no alias token is provided self.assertIsNone(result['alias_token']) @@ -112,16 +115,34 @@ class ActorInitializeV3(rest.test.RESTTestCase): # Ensure requested action is rename self.assertEqual(os['action'], 'rename') # And name is userservice name - self.assertEqual(os['name'], self.user_service.friendly_name) + self.assertEqual(os['name'], self.user_service_managed.friendly_name) # Now invoke failure - self.invoke_failure('invalid token', unique_id, True) + failure('invalid token', unique_id, True) # Now invoke failure with valid token but invalid mac - result = self.invoke_failure(actor_token, 'invalid mac', False) + result = failure(actor_token, 'invalid mac', False) self.assertIsNone(result['own_token'], None) self.assertIsNone(result['alias_token']) self.assertIsNone(result['os']) - self.assertIsNone(result['unique_id']) \ No newline at end of file + self.assertIsNone(result['unique_id']) + + + def test_initialize_unmanaged(self): + """ + Test actor initialize v3 for unmanaged actor + """ + actor_token = self.user_service_managed.deployed_service.service.token + + unique_id = self.user_service_managed.getUniqueId() + + success = functools.partial(self.invoke_success, 'unmanaged', actor_token) + failure = functools.partial(self.invoke_failure, 'unmanaged') + + # This will succeed + result = success(unique_id) + + logger.info('Result is %s', result) + diff --git a/server/src/tests/REST/test_test.py b/server/src/tests/REST/test_test.py index 9a3bfa16..47d3fdcd 100644 --- a/server/src/tests/REST/test_test.py +++ b/server/src/tests/REST/test_test.py @@ -107,7 +107,7 @@ class TestActorV3(rest.test.RESTTestCase): def test_test_unmanaged(self) -> None: # try for a first few services - service = self.user_service.deployed_service.service + service = self.user_service_managed.deployed_service.service rest_token, actor_token = self.login_and_register() # Get service token self.do_test(UNMANAGED, service.token) diff --git a/server/src/tests/fixtures/services.py b/server/src/tests/fixtures/services.py index 45b1ac0f..bfb066c2 100644 --- a/server/src/tests/fixtures/services.py +++ b/server/src/tests/fixtures/services.py @@ -47,7 +47,8 @@ glob = { def createSingleTestingUserServiceStructure( - user: 'models.User', groups: typing.List['models.Group'] + user: 'models.User', groups: typing.List['models.Group'], + type_: typing.Union[typing.Literal['managed'], typing.Literal['unmanaged']], ) -> 'models.UserService': from uds.services.Test.provider import TestProvider @@ -80,15 +81,17 @@ def createSingleTestingUserServiceStructure( 'onLogout': 'remove', 'idle': 300, } - osmanager: 'models.OSManager' = models.OSManager.objects.create( - name='OS Manager %d' % (glob['osmanager_id']), - comments='Comment for OS Manager %d' % (glob['osmanager_id']), - data_type=TestOSManager.typeType, - data=TestOSManager( - environment.Environment(str(glob['osmanager_id'])), values - ).serialize(), - ) - glob['osmanager_id'] += 1 + osmanager: typing.Optional['models.OSManager'] = None + if type_ == 'managed': + osmanager = models.OSManager.objects.create( + name='OS Manager %d' % (glob['osmanager_id']), + comments='Comment for OS Manager %d' % (glob['osmanager_id']), + data_type=TestOSManager.typeType, + data=TestOSManager( + environment.Environment(str(glob['osmanager_id'])), values + ).serialize(), + ) + glob['osmanager_id'] += 1 values = { 'testURL': 'http://www.udsenterprise.com', diff --git a/server/src/tests/utils/rest/test.py b/server/src/tests/utils/rest/test.py index 61026d66..a74022c5 100644 --- a/server/src/tests/utils/rest/test.py +++ b/server/src/tests/utils/rest/test.py @@ -50,7 +50,8 @@ class RESTTestCase(test.UDSTransactionTestCasse): staffs: typing.List[models.User] plain_users: typing.List[models.User] - user_service: models.UserService + user_service_managed: models.UserService + user_service_unamanaged: models.UserService def setUp(self) -> None: # Set up data for REST Test cases @@ -76,8 +77,19 @@ class RESTTestCase(test.UDSTransactionTestCasse): self.auth, number_of_users=NUMBER_OF_ITEMS_TO_CREATE, groups=self.groups ) - self.user_service = fixtures.services.createSingleTestingUserServiceStructure( - self.admins[0], self.groups + self.user_service_managed = ( + fixtures.services.createSingleTestingUserServiceStructure( + self.admins[0], + self.groups, + 'managed', + ) + ) + self.user_service_unamanaged = ( + fixtures.services.createSingleTestingUserServiceStructure( + self.admins[0], + self.groups, + 'unmanaged', + ) ) @staticmethod @@ -121,7 +133,9 @@ class RESTTestCase(test.UDSTransactionTestCasse): # - The login auth token # - The actor token def login_and_register(self, as_admin: bool = True) -> typing.Tuple[str, str]: - token = self.login(as_admin=as_admin) # Token not used, alreade inserted on login + token = self.login( + as_admin=as_admin + ) # Token not used, alreade inserted on login response = self.client.post( '/uds/rest/actor/v3/register', data=self.register_data(constants.STRING_CHARS), diff --git a/server/src/uds/REST/methods/actor_v3.py b/server/src/uds/REST/methods/actor_v3.py index 004b9773..9e058ce6 100644 --- a/server/src/uds/REST/methods/actor_v3.py +++ b/server/src/uds/REST/methods/actor_v3.py @@ -74,7 +74,7 @@ class BlockAccess(Exception): # Helpers def fixIdsList(idsList: typing.List[str]) -> typing.List[str]: - return [i.upper() for i in idsList] + [i.lower() for i in idsList] + return list(set([i.upper() for i in idsList] + [i.lower() for i in idsList])) def checkBlockedIp(ip: str) -> None: @@ -304,14 +304,13 @@ class Initialize(ActorV3Action): if self._params['type'] == UNMANAGED: alias_token = token # Store token as possible alias # First, try to locate on alias table - if ServiceTokenAlias.objects.get(alias=token).exists(): + if ServiceTokenAlias.objects.filter(alias=token).exists(): # Retrieve real service from token alias service = ServiceTokenAlias.objects.get(alias=token).service - # If not found, try to locate on service table - if ( - service is None - ): # Not on alias token, try to locate on Service table - service = Service.objects.get(token=token) + # If not found an alias, try to locate on service table + # Not on alias token, try to locate on Service table + if not service: + service = typing.cast('Service', Service.objects.get(token=token)) # And create a new alias for it, and save alias_token = ( cryptoManager().randomString()