mirror of
https://github.com/dkmstr/openuds.git
synced 2025-01-03 01:17:56 +03:00
Fixing typing also on tests
This commit is contained in:
parent
3bf7af2e3d
commit
f9e2d88529
@ -167,12 +167,9 @@ class Dispatcher(View): # type: ignore
|
|||||||
content_type="text/plain",
|
content_type="text/plain",
|
||||||
)
|
)
|
||||||
except AttributeError:
|
except AttributeError:
|
||||||
allowedMethods = []
|
allowed_methods: list[str] = [n for n in ['get', 'post', 'put', 'delete'] if hasattr(handler, n)]
|
||||||
for n in ['get', 'post', 'put', 'delete']:
|
|
||||||
if hasattr(handler, n):
|
|
||||||
allowedMethods.append(n)
|
|
||||||
log.log_operation(handler, 405, log.LogLevel.ERROR)
|
log.log_operation(handler, 405, log.LogLevel.ERROR)
|
||||||
return http.HttpResponseNotAllowed(allowedMethods, content_type="text/plain")
|
return http.HttpResponseNotAllowed(allowed_methods, content_type="text/plain")
|
||||||
except exceptions.rest.AccessDenied:
|
except exceptions.rest.AccessDenied:
|
||||||
log.log_operation(handler, 403, log.LogLevel.ERROR)
|
log.log_operation(handler, 403, log.LogLevel.ERROR)
|
||||||
return http.HttpResponseForbidden('access denied', content_type="text/plain")
|
return http.HttpResponseForbidden('access denied', content_type="text/plain")
|
||||||
|
@ -62,7 +62,7 @@ class ServersTokens(ModelHandler):
|
|||||||
path = 'servers'
|
path = 'servers'
|
||||||
name = 'tokens'
|
name = 'tokens'
|
||||||
|
|
||||||
table_title = typing.cast('str', _('Registered Servers'))
|
table_title = _('Registered Servers')
|
||||||
table_fields = [
|
table_fields = [
|
||||||
{'hostname': {'title': _('Hostname')}},
|
{'hostname': {'title': _('Hostname')}},
|
||||||
{'ip': {'title': _('IP')}},
|
{'ip': {'title': _('IP')}},
|
||||||
@ -72,8 +72,8 @@ class ServersTokens(ModelHandler):
|
|||||||
{'stamp': {'title': _('Date'), 'type': 'datetime'}},
|
{'stamp': {'title': _('Date'), 'type': 'datetime'}},
|
||||||
]
|
]
|
||||||
|
|
||||||
def item_as_dict(self, item_: 'Model') -> dict[str, typing.Any]:
|
def item_as_dict(self, item: 'Model') -> dict[str, typing.Any]:
|
||||||
item = typing.cast('models.Server', item_) # We will receive for sure
|
item = typing.cast('models.Server', item) # We will receive for sure
|
||||||
return {
|
return {
|
||||||
'id': item.uuid,
|
'id': item.uuid,
|
||||||
'name': str(_('Token isued by {} from {}')).format(item.username, item.ip),
|
'name': str(_('Token isued by {} from {}')).format(item.username, item.ip),
|
||||||
@ -111,8 +111,8 @@ class ServersTokens(ModelHandler):
|
|||||||
class ServersServers(DetailHandler):
|
class ServersServers(DetailHandler):
|
||||||
custom_methods = ['maintenance']
|
custom_methods = ['maintenance']
|
||||||
|
|
||||||
def get_items(self, parent_: 'Model', item: typing.Optional[str]) -> types.rest.ManyItemsDictType:
|
def get_items(self, parent: 'Model', item: typing.Optional[str]) -> types.rest.ManyItemsDictType:
|
||||||
parent = typing.cast('models.ServerGroup', parent_) # We will receive for sure
|
parent = typing.cast('models.ServerGroup', parent) # We will receive for sure
|
||||||
try:
|
try:
|
||||||
multi = False
|
multi = False
|
||||||
if item is None:
|
if item is None:
|
||||||
@ -120,7 +120,7 @@ class ServersServers(DetailHandler):
|
|||||||
q = parent.servers.all()
|
q = parent.servers.all()
|
||||||
else:
|
else:
|
||||||
q = parent.servers.filter(uuid=process_uuid(item))
|
q = parent.servers.filter(uuid=process_uuid(item))
|
||||||
res = []
|
res: types.rest.ManyItemsDictType = []
|
||||||
i = None
|
i = None
|
||||||
for i in q:
|
for i in q:
|
||||||
val = {
|
val = {
|
||||||
@ -144,7 +144,7 @@ class ServersServers(DetailHandler):
|
|||||||
def get_title(self, parent: 'Model') -> str:
|
def get_title(self, parent: 'Model') -> str:
|
||||||
parent = ensure.is_instance(parent, models.ServerGroup)
|
parent = ensure.is_instance(parent, models.ServerGroup)
|
||||||
try:
|
try:
|
||||||
return _('Servers of {0}').format(parent.name)
|
return (_('Servers of {0}')).format(parent.name)
|
||||||
except Exception:
|
except Exception:
|
||||||
return str(_('Servers'))
|
return str(_('Servers'))
|
||||||
|
|
||||||
@ -316,7 +316,7 @@ class ServersGroups(ModelHandler):
|
|||||||
name = 'groups'
|
name = 'groups'
|
||||||
|
|
||||||
save_fields = ['name', 'comments', 'type', 'tags'] # Subtype is appended on pre_save
|
save_fields = ['name', 'comments', 'type', 'tags'] # Subtype is appended on pre_save
|
||||||
table_title = typing.cast(str, _('Servers Groups'))
|
table_title = _('Servers Groups')
|
||||||
table_fields = [
|
table_fields = [
|
||||||
{'name': {'title': _('Name')}},
|
{'name': {'title': _('Name')}},
|
||||||
{'comments': {'title': _('Comments')}},
|
{'comments': {'title': _('Comments')}},
|
||||||
@ -338,9 +338,9 @@ class ServersGroups(ModelHandler):
|
|||||||
type_ += '@default'
|
type_ += '@default'
|
||||||
kind, subkind = type_.split('@')[:2]
|
kind, subkind = type_.split('@')[:2]
|
||||||
if kind == types.servers.ServerType.SERVER.name:
|
if kind == types.servers.ServerType.SERVER.name:
|
||||||
kind = typing.cast(str, _('Standard'))
|
kind = _('Standard')
|
||||||
elif kind == types.servers.ServerType.UNMANAGED.name:
|
elif kind == types.servers.ServerType.UNMANAGED.name:
|
||||||
kind = typing.cast(str, _('Unmanaged'))
|
kind = _('Unmanaged')
|
||||||
title = _('of type') + f' {subkind.upper()} {kind}'
|
title = _('of type') + f' {subkind.upper()} {kind}'
|
||||||
return self.add_field(
|
return self.add_field(
|
||||||
self.add_default_fields(
|
self.add_default_fields(
|
||||||
|
@ -88,14 +88,14 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
|
|||||||
}
|
}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def service_to_dict(item: models.Service, perm: int, full: bool = False) -> dict[str, typing.Any]:
|
def service_to_dict(item: models.Service, perm: int, full: bool = False) -> types.rest.ItemDictType:
|
||||||
"""
|
"""
|
||||||
Convert a service db item to a dict for a rest response
|
Convert a service db item to a dict for a rest response
|
||||||
:param item: Service item (db)
|
:param item: Service item (db)
|
||||||
:param full: If full is requested, add "extra" fields to complete information
|
:param full: If full is requested, add "extra" fields to complete information
|
||||||
"""
|
"""
|
||||||
itemType = item.get_type()
|
itemType = item.get_type()
|
||||||
ret_value = {
|
ret_value: dict[str, typing.Any] = {
|
||||||
'id': item.uuid,
|
'id': item.uuid,
|
||||||
'name': item.name,
|
'name': item.name,
|
||||||
'tags': [tag.tag for tag in item.tags.all()],
|
'tags': [tag.tag for tag in item.tags.all()],
|
||||||
@ -166,7 +166,7 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
|
|||||||
service = parent.services.get(uuid=process_uuid(item))
|
service = parent.services.get(uuid=process_uuid(item))
|
||||||
service.__dict__.update(fields)
|
service.__dict__.update(fields)
|
||||||
|
|
||||||
if service is None:
|
if not service:
|
||||||
raise Exception('Cannot create service!')
|
raise Exception('Cannot create service!')
|
||||||
|
|
||||||
service.tags.set([models.Tag.objects.get_or_create(tag=val)[0] for val in tags])
|
service.tags.set([models.Tag.objects.get_or_create(tag=val)[0] for val in tags])
|
||||||
@ -322,11 +322,11 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
|
|||||||
except Exception:
|
except Exception:
|
||||||
raise self.invalid_item_response() from None
|
raise self.invalid_item_response() from None
|
||||||
|
|
||||||
def servicepools(self, parent: 'Model', item: str) -> typing.Any:
|
def servicepools(self, parent: 'Model', item: str) -> types.rest.ManyItemsDictType:
|
||||||
parent = ensure.is_instance(parent, models.Provider)
|
parent = ensure.is_instance(parent, models.Provider)
|
||||||
service = parent.services.get(uuid=process_uuid(item))
|
service = parent.services.get(uuid=process_uuid(item))
|
||||||
logger.debug('Got parameters for servicepools: %s, %s', parent, item)
|
logger.debug('Got parameters for servicepools: %s, %s', parent, item)
|
||||||
res = []
|
res: types.rest.ManyItemsDictType = []
|
||||||
for i in service.deployedServices.all():
|
for i in service.deployedServices.all():
|
||||||
try:
|
try:
|
||||||
self.ensure_has_access(
|
self.ensure_has_access(
|
||||||
|
@ -396,7 +396,7 @@ class Groups(DetailHandler):
|
|||||||
|
|
||||||
def get_types(self, parent: 'Model', for_type: typing.Optional[str]) -> collections.abc.Iterable[types.rest.TypeInfoDict]:
|
def get_types(self, parent: 'Model', for_type: typing.Optional[str]) -> collections.abc.Iterable[types.rest.TypeInfoDict]:
|
||||||
parent = ensure.is_instance(parent, Authenticator)
|
parent = ensure.is_instance(parent, Authenticator)
|
||||||
tDct = {
|
types_dict: dict[str, dict[str, str]] = {
|
||||||
'group': {'name': _('Group'), 'description': _('UDS Group')},
|
'group': {'name': _('Group'), 'description': _('UDS Group')},
|
||||||
'meta': {'name': _('Meta group'), 'description': _('UDS Meta Group')},
|
'meta': {'name': _('Meta group'), 'description': _('UDS Meta Group')},
|
||||||
}
|
}
|
||||||
@ -407,7 +407,7 @@ class Groups(DetailHandler):
|
|||||||
'description': v['description'],
|
'description': v['description'],
|
||||||
'icon': '',
|
'icon': '',
|
||||||
}
|
}
|
||||||
for k, v in tDct.items()
|
for k, v in types_dict.items()
|
||||||
]
|
]
|
||||||
|
|
||||||
if for_type is None:
|
if for_type is None:
|
||||||
|
@ -139,7 +139,7 @@ class BaseModelHandler(Handler):
|
|||||||
guiDesc['order'] = field.get('order', 0)
|
guiDesc['order'] = field.get('order', 0)
|
||||||
guiDesc['label'] = field.get('label', field['name'])
|
guiDesc['label'] = field.get('label', field['name'])
|
||||||
|
|
||||||
v = {
|
v: dict[str, typing.Any] = {
|
||||||
'name': field.get('name', ''),
|
'name': field.get('name', ''),
|
||||||
'value': field.get('value', ''),
|
'value': field.get('value', ''),
|
||||||
'gui': guiDesc,
|
'gui': guiDesc,
|
||||||
@ -289,7 +289,7 @@ class BaseModelHandler(Handler):
|
|||||||
icon=type_.icon64().replace('\n', ''),
|
icon=type_.icon64().replace('\n', ''),
|
||||||
).as_dict(**self.type_info(type_))
|
).as_dict(**self.type_info(type_))
|
||||||
if hasattr(type_, 'group'):
|
if hasattr(type_, 'group'):
|
||||||
res['group'] = _(type_.group) # Add group info is it is contained
|
res['group'] = getattr(type_, 'group') # Add group info is it is contained
|
||||||
return res
|
return res
|
||||||
|
|
||||||
def process_table_fields(
|
def process_table_fields(
|
||||||
@ -830,10 +830,10 @@ class ModelHandler(BaseModelHandler):
|
|||||||
return data
|
return data
|
||||||
|
|
||||||
# Filtering a non iterable (list or tuple)
|
# Filtering a non iterable (list or tuple)
|
||||||
if not isinstance(data, (list, tuple, GeneratorType)):
|
if not isinstance(data, collections.abc.Iterable):
|
||||||
return data
|
return data
|
||||||
|
|
||||||
logger.debug('data: %s, fltr: %s', data, self.fltr)
|
logger.debug('data: %s, fltr: %s', typing.cast(typing.Any, data), self.fltr)
|
||||||
try:
|
try:
|
||||||
fld, pattern = self.fltr.split('=')
|
fld, pattern = self.fltr.split('=')
|
||||||
s, e = '', ''
|
s, e = '', ''
|
||||||
@ -846,15 +846,19 @@ class ModelHandler(BaseModelHandler):
|
|||||||
|
|
||||||
r = re.compile(s + fnmatch.translate(pattern) + e, re.RegexFlag.IGNORECASE)
|
r = re.compile(s + fnmatch.translate(pattern) + e, re.RegexFlag.IGNORECASE)
|
||||||
|
|
||||||
def fltr_function(item: collections.abc.MutableMapping[str, typing.Any]) -> bool:
|
def fltr_function(item: typing.Any) -> bool:
|
||||||
|
if not isinstance(item, dict):
|
||||||
|
return False
|
||||||
try:
|
try:
|
||||||
if fld not in item or r.match(item[fld]) is None:
|
if fld not in item or r.match(typing.cast(dict[str, typing.Any], item)[fld]) is None:
|
||||||
return False
|
return False
|
||||||
except Exception:
|
except Exception:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
res = list(filter(fltr_function, data))
|
res: list[dict[str, typing.Any]] = list(
|
||||||
|
filter(fltr_function, typing.cast(collections.abc.Iterable[dict[str, typing.Any]], data))
|
||||||
|
)
|
||||||
|
|
||||||
logger.debug('After filtering: %s', res)
|
logger.debug('After filtering: %s', res)
|
||||||
return res
|
return res
|
||||||
@ -1048,9 +1052,7 @@ class ModelHandler(BaseModelHandler):
|
|||||||
raise self.invalid_request_response()
|
raise self.invalid_request_response()
|
||||||
try:
|
try:
|
||||||
# DB maybe case sensitive??, anyway, uuids are stored in lowercase
|
# DB maybe case sensitive??, anyway, uuids are stored in lowercase
|
||||||
item = self.model.objects.get(
|
item = self.model.objects.get(uuid=self._args[0].lower())
|
||||||
uuid=self._args[0].lower()
|
|
||||||
)
|
|
||||||
return self.get_logs(item)
|
return self.get_logs(item)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise self.invalid_item_response() from e
|
raise self.invalid_item_response() from e
|
||||||
|
@ -33,16 +33,8 @@ import typing
|
|||||||
import dataclasses
|
import dataclasses
|
||||||
import collections.abc
|
import collections.abc
|
||||||
|
|
||||||
from click import group
|
|
||||||
|
|
||||||
class TypeInfoDict(typing.TypedDict):
|
|
||||||
name: str
|
|
||||||
type: str
|
|
||||||
description: str
|
|
||||||
icon: str
|
|
||||||
|
|
||||||
group: typing.NotRequired[str]
|
|
||||||
|
|
||||||
|
TypeInfoDict = dict[str, typing.Any] # Alias for type info dict
|
||||||
|
|
||||||
@dataclasses.dataclass
|
@dataclasses.dataclass
|
||||||
class TypeInfo:
|
class TypeInfo:
|
||||||
@ -51,25 +43,27 @@ class TypeInfo:
|
|||||||
description: str
|
description: str
|
||||||
icon: str
|
icon: str
|
||||||
|
|
||||||
def as_dict(self, group: typing.Optional[str] = None) -> TypeInfoDict:
|
def as_dict(self, **kwargs: typing.Any) -> TypeInfoDict:
|
||||||
res: TypeInfoDict = {
|
res: dict[str, typing.Any] = {
|
||||||
'name': self.name,
|
'name': self.name,
|
||||||
'type': self.type,
|
'type': self.type,
|
||||||
'description': self.description,
|
'description': self.description,
|
||||||
'icon': self.icon,
|
'icon': self.icon,
|
||||||
}
|
}
|
||||||
if group is not None:
|
res.update(kwargs)
|
||||||
res['group'] = group
|
|
||||||
return res
|
return res
|
||||||
|
|
||||||
|
|
||||||
# This is a named tuple for convenience, and must be
|
# This is a named tuple for convenience, and must be
|
||||||
# compatible with tuple[str, bool] (name, needs_parent)
|
# compatible with tuple[str, bool] (name, needs_parent)
|
||||||
class ModelCustomMethod(typing.NamedTuple):
|
class ModelCustomMethod(typing.NamedTuple):
|
||||||
name: str
|
name: str
|
||||||
needs_parent: bool = True
|
needs_parent: bool = True
|
||||||
|
|
||||||
|
|
||||||
# Alias for item type
|
# Alias for item type
|
||||||
ItemDictType = dict[str, typing.Any]
|
ItemDictType = dict[str, typing.Any]
|
||||||
|
|
||||||
# Alias for get_items return type
|
# Alias for get_items return type
|
||||||
ManyItemsDictType = typing.Union[list[ItemDictType], ItemDictType]
|
ManyItemsDictType = typing.Union[list[ItemDictType], ItemDictType]
|
||||||
|
@ -99,7 +99,7 @@ class Tag(UUIDModel):
|
|||||||
|
|
||||||
|
|
||||||
class TaggingMixin(models.Model):
|
class TaggingMixin(models.Model):
|
||||||
tags: 'models.ManyToManyField[models.Model, TaggingMixin]' = models.ManyToManyField(Tag)
|
tags: 'models.ManyToManyField[Tag, TaggingMixin]' = models.ManyToManyField(Tag)
|
||||||
|
|
||||||
class Meta: # pylint: disable=too-few-public-methods
|
class Meta: # pylint: disable=too-few-public-methods
|
||||||
abstract = True
|
abstract = True
|
||||||
|
Loading…
Reference in New Issue
Block a user