burmatovan/openuds
1
0
Fork 0
forked from shaba/openuds
Code Issues Pull Requests Projects Releases Wiki Activity

fixing tests

Browse Source
This commit is contained in:
Adolfo Gómez García 2022-08-21 19:37:32 +02:00
parent c6be2562ee
commit fa3e30c518
5 changed files with 71 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
server/src/tests/REST/test_initialize.py
View File

@ -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'])
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)

2
server/src/tests/REST/test_test.py
View File

@ -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)

23
server/src/tests/fixtures/services.py vendored
View File

@ -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',

22
server/src/tests/utils/rest/test.py
View File

@ -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),

13
server/src/uds/REST/methods/actor_v3.py
View File

@ -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()