13.0.2

CHANGELOG.md

@@ -1,20 +1,33 @@
-# Список изменений в версии 13.0.1
+# Список изменений в версии 13.0.2
 
 ## Изменено
-| Идентификатор<br>задачи | Описание |
-| --- | --- |
+
 
 ## Добавлено
+### Глобально
+| Идентификатор<br>задачи | Описание |
+| --- | --- |
+| BANS-1463 | Добавлен модуль `decort_vm_folder_list`, позволяющий получить список папок для группировки ВМ в ресурсной группе. |
+| BANS-1467 | Добавлен модуль `decort_vm_folder_access`, позволяющий управлять правами доступа к папке виртуальных машин. |
+| BANS-1464 | Добавлен модуль `decort_vm_folder`, позволяющий привести к целевому состоянию и получить информацию о папке для группировки ВМ в ресурсной группе. |
+
 ### Модуль decort_vm
 | Идентификатор<br>задачи | Описание |
 | --- | --- |
-| BANS-1413 | Добавлен параметр `disks.objects.read_only`. |
+| BANS-1466 | Добавлены параметр и возвращаемое значение `folder_id`. |
+### Модуль decort_vm_list
+| Идентификатор<br>задачи | Описание |
+| --- | --- |
+| BANS-1468 | Добавлено возвращаемое значение `folder_id`. |
+| BANS-1476 | Добавлен параметр `include_descendants`. |
+| BANS-1477 | Добавлен параметр `filter.folder_id` для фильтрации виртуальных машин по идентификатору папки для группировки ВМ в ресурсной группе. |
+
+### Модуль decort_audit_list
+| Идентификатор<br>задачи | Описание |
+| --- | --- |
+| BANS-1840 | Добавлено возвращаемое значение `vmfolder_id`. |
 
 ## Удалено
-| Идентификатор<br>задачи | Описание |
-| --- | --- |
+
 
 ## Исправлено
-| Идентификатор<br>задачи | Описание |
-| --- | --- |
-| BANS-1423 | Модуль `decort_vm` завершал работу ошибкой запроса к API при подключении к ВМ интерфейса типа `EMPTY`. |

library/decort_vm.py

@@ -620,6 +620,12 @@ class decort_vm(DecortController):
                 custom_fields=custom_fields_fields,
             )
 
+        if self.aparams['folder_id'] is not None:
+            self.sdk_checkmode(self.api.cloudapi.vmfolder.computes_move)(
+                compute_ids=[self.comp_info['id']],
+                folder_id=self.aparams['folder_id'],
+            )
+
         # read in Compute facts once more after all initial setup is complete
         _, self.comp_info, _ = self.compute_find(
             comp_id=self.comp_id,
@@ -679,6 +685,16 @@ class decort_vm(DecortController):
                 **self.compute_update_args,
             )
 
+        folder_id = self.aparams['folder_id']
+        if (
+            folder_id is not None
+            and folder_id != (self.comp_info.get('folder_id') or 0)
+        ):
+            self.sdk_checkmode(self.api.cloudapi.vmfolder.computes_move)(
+                compute_ids=[self.comp_info['id']],
+                folder_id=folder_id,
+            )
+
         if self.amodule.params['rollback_to'] is not None:
             self.sdk_checkmode(self.api.cloudapi.compute.snapshot_rollback)(
                 vm_id=self.comp_info['id'],
@@ -979,6 +995,7 @@ class decort_vm(DecortController):
                     tech_status="",
                     account_id=0,
                     rg_id=0,
+                    folder_id=0,
                     username="",
                     password="",
                     public_ips=[],  # direct IPs; this list can be empty
@@ -1012,6 +1029,7 @@ class decort_vm(DecortController):
         ret_dict['tech_status'] = self.comp_info['techStatus']
         ret_dict['account_id'] = self.comp_info['accountId']
         ret_dict['rg_id'] = self.comp_info['rgId']
+        ret_dict['folder_id'] = self.comp_info.get('folder_id') or 0
         if self.comp_info['tags']:
             ret_dict['tags'] = self.comp_info['tags']
         # if the VM is an imported VM, then the 'accounts' list may be empty,
@@ -1399,6 +1417,9 @@ class decort_vm(DecortController):
                         ),
                     ),
                 ),
+                folder_id=dict(
+                    type='int',
+                ),
                 id=dict(
                     type='int',
                     default=0,

library/decort_vm_folder.py

@@ -0,0 +1,235 @@
+#!/usr/bin/python
+
+DOCUMENTATION = r'''
+---
+module: decort_vm_folder
+
+description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501
+'''
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.decort_utils import (
+    AnsibleModuleParamSet,
+    DecortController,
+)
+
+from dynamix_sdk import exceptions as sdk_exceptions
+from dynamix_sdk import types as sdk_types
+
+
+class VmFolderState(Enum):
+    absent = 'absent'
+
+
+@dataclass(frozen=True)
+class ParamSet(AnsibleModuleParamSet):
+    id: int | None
+    name: str | None
+    parent_id: int | None
+    rg_id: int | None
+    state: VmFolderState | None
+
+
+class DecortVmFolder(DecortController):
+    param_set: ParamSet
+    facts: dict[str, Any]
+    _id: int | None = None
+    _vm_folder_info: sdk_types.CloudapiVmfolderGetResultModel | None = None
+
+    def __init__(self) -> None:
+        super().__init__(AnsibleModule(**self.amodule_init_args))
+        self.param_set = ParamSet.from_dict(self.amodule.params)
+        self.facts = {}
+
+    @property
+    def amodule_init_args(self) -> dict[str, Any]:
+        return self.pack_amodule_init_args(
+            argument_spec=ParamSet.get_arg_spec(),
+            supports_check_mode=True,
+        )
+
+    @property
+    def id(self) -> int:
+        if self._id is None:
+            raise TypeError
+        return self._id
+
+    @property
+    def vm_folder_info(self) -> sdk_types.CloudapiVmfolderGetResultModel:
+        if self._vm_folder_info is None:
+            try:
+                vm_folder_info = self.api.cloudapi.vmfolder.get(
+                    folder_id=self.id,
+                )
+            except sdk_exceptions.RequestException as e:
+                if (
+                    e.orig_exception.response is not None
+                    and e.orig_exception.response.status_code == 404
+                ):
+                    self.message(
+                        self.MESSAGES.obj_not_found(
+                            obj='vm_folder',
+                            id=self.id,
+                        )
+                    )
+                    self.exit(fail=True)
+                raise e
+            self._vm_folder_info = vm_folder_info
+        return self._vm_folder_info
+
+    @DecortController.handle_sdk_exceptions
+    def run(self) -> None:
+        if self.param_set.id is not None:
+            self._id = self.param_set.id
+        elif self.param_set.name is not None:
+            self._id = self.find_id_by_name()
+
+        if self._id is not None:
+            self.get_info()
+            self.change()
+        else:
+            self.check_amodule_args_for_create()
+            self.create()
+
+        if self.result['changed']:
+            self.get_info()
+        self.exit()
+
+    def find_id_by_name(self) -> int | None:
+        rg_id = self.param_set.rg_id
+        name = self.param_set.name
+        if rg_id is None or name is None:
+            return None
+
+        matches: list[int] = []
+        page = 1
+        page_size = 100
+        while True:
+            folders = self.api.cloudapi.vmfolder.list(
+                rg_id=rg_id,
+                page=page,
+                size=page_size,
+            )
+            for folder in folders.data:
+                if folder.name != name:
+                    continue
+                if (
+                    self.param_set.parent_id is not None
+                    and folder.parent_id != self.param_set.parent_id
+                ):
+                    continue
+                matches.append(folder.id)
+
+            if not folders.data or len(folders.data) < page_size:
+                break
+            if (
+                folders.entry_count
+                and page * page_size >= folders.entry_count
+            ):
+                break
+            page += 1
+
+        if len(matches) > 1:
+            self.message(
+                msg='Check for parameter "name" failed: '
+                    'multiple vm folders with this name were found; '
+                    'specify parent_id or id'
+            )
+            self.exit(fail=True)
+        if matches:
+            return matches[0]
+        return None
+
+    def get_info(self) -> None:
+        if self._vm_folder_info is not None and not self.amodule.check_mode:
+            self._vm_folder_info = None
+        self.facts = self.vm_folder_info.model_dump()
+
+    def check_amodule_args_for_create(self) -> None:
+        check_errors = False
+
+        if self.param_set.rg_id is None:
+            check_errors = True
+            self.message(
+                msg='Check for parameter "rg_id" failed: '
+                    'rg_id must be specified when creating '
+                    'a new vm_folder'
+            )
+
+        if self.param_set.name is None:
+            check_errors = True
+            self.message(
+                msg='Check for parameter "name" failed: '
+                    'name must be specified when creating '
+                    'a new vm_folder'
+            )
+
+        if self.param_set.state == VmFolderState.absent:
+            check_errors = True
+            self.message(
+                msg='Check for parameter "state" failed: '
+                    'state can not be "absent" when creating '
+                    'a new vm_folder'
+            )
+
+        if check_errors:
+            self.exit(fail=True)
+
+    def create(self) -> None:
+        rg_id = self.param_set.rg_id
+        name = self.param_set.name
+        if rg_id is None or name is None:
+            raise TypeError
+        created_id = self.sdk_checkmode(
+            self.api.cloudapi.vmfolder.create
+        )(
+            rg_id=rg_id,
+            name=name,
+            parent_id=self.param_set.parent_id,
+        )
+        if created_id:
+            self._id = created_id
+
+    def change(self) -> None:
+        self.change_state()
+        self.change_params()
+
+    def change_state(self) -> None:
+        if self.param_set.state == VmFolderState.absent:
+            self.delete()
+
+    def change_params(self) -> None:
+        if (
+            self.param_set.name is not None
+            and self.param_set.name != self.vm_folder_info.name
+        ):
+            self.sdk_checkmode(self.api.cloudapi.vmfolder.rename)(
+                folder_id=self.id,
+                name=self.param_set.name,
+            )
+        if (
+            self.param_set.parent_id is not None
+            and self.param_set.parent_id != self.vm_folder_info.parent_id
+        ):
+            self.sdk_checkmode(self.api.cloudapi.vmfolder.move)(
+                folder_id=self.id,
+                new_parent_id=self.param_set.parent_id,
+            )
+
+    def delete(self) -> None:
+        self.sdk_checkmode(self.api.cloudapi.vmfolder.delete)(
+            folder_id=self.id,
+        )
+        self.exit()
+
+
+def main() -> None:
+    DecortVmFolder().run()
+
+
+if __name__ == '__main__':
+    main()

library/decort_vm_folder_access.py

@@ -0,0 +1,141 @@
+#!/usr/bin/python
+
+DOCUMENTATION = r'''
+---
+module: decort_vm_folder_access
+
+description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501
+'''
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.decort_utils import (
+    AnsibleModuleParamSet,
+    DecortController,
+)
+
+from dynamix_sdk import exceptions as sdk_exceptions
+from dynamix_sdk import types as sdk_types
+
+
+class VMFolderAccessState(Enum):
+    absent = 'absent'
+
+
+@dataclass(frozen=True)
+class ParamSet(AnsibleModuleParamSet):
+    folder_id: int
+    user: str
+    right: sdk_types.AccessTypeForSet | None
+    state: VMFolderAccessState | None
+
+
+class DecortVMFolderAccess(DecortController):
+    param_set: ParamSet
+    folder_info: sdk_types.CloudapiVmfolderGetResultModel
+    acl: sdk_types.ACLForVMFolderGetAPIResultNM | None = None
+
+    def __init__(self) -> None:
+        super().__init__(AnsibleModule(**self.amodule_init_args))
+        self.param_set = ParamSet.from_dict(self.amodule.params)
+
+    @property
+    def amodule_init_args(self) -> dict[str, Any]:
+        return self.pack_amodule_init_args(
+            argument_spec=ParamSet.get_arg_spec(),
+            supports_check_mode=True,
+        )
+
+    @DecortController.handle_sdk_exceptions
+    def run(self) -> None:
+        self.get_folder()
+        self.check_amodule_args()
+        self.change()
+        if self.result['changed'] and not self.amodule.check_mode:
+            self.get_folder()
+        self.package_facts()
+        self.exit()
+
+    def get_folder(self) -> None:
+        try:
+            self.folder_info = self.api.cloudapi.vmfolder.get(
+                folder_id=self.param_set.folder_id,
+            )
+        except sdk_exceptions.RequestException as e:
+            if (
+                e.orig_exception.response is not None
+                and e.orig_exception.response.status_code == 404
+            ):
+                self.message(
+                    self.MESSAGES.obj_not_found(
+                        obj='VM folder',
+                        id=self.param_set.folder_id,
+                    )
+                )
+                self.exit(fail=True)
+            raise e
+        self.acl = self.find_acl()
+
+    def find_acl(self) -> sdk_types.ACLForVMFolderGetAPIResultNM | None:
+        for acl in self.folder_info.acl:
+            if acl.explicit and acl.user_group_id == self.param_set.user:
+                return acl
+        return None
+
+    def check_amodule_args(self) -> None:
+        if (
+            self.param_set.state is None
+            and self.param_set.right is None
+            and self.acl is None
+        ):
+            self.message(
+                self.MESSAGES.obj_not_found(
+                    obj='VM folder ACL',
+                    id=self.param_set.user,
+                )
+            )
+            self.exit(fail=True)
+
+    def change(self) -> None:
+        if self.param_set.state == VMFolderAccessState.absent:
+            self.delete()
+            return
+
+        right = self.param_set.right
+        if right is None or (
+            self.acl is not None and self.acl.right == right
+        ):
+            return
+
+        self.sdk_checkmode(self.api.cloudapi.vmfolder.add_acl_access)(
+            folder_id=self.param_set.folder_id,
+            user=self.param_set.user,
+            right=right,
+        )
+
+    def delete(self) -> None:
+        if self.acl is None:
+            return
+        self.sdk_checkmode(self.api.cloudapi.vmfolder.del_acl_access)(
+            folder_id=self.param_set.folder_id,
+            user=self.param_set.user,
+        )
+        if not self.amodule.check_mode:
+            self.acl = None
+
+    def package_facts(self) -> None:
+        if self.acl is not None:
+            self.facts = self.acl.model_dump()
+        else:
+            self.facts = {}
+
+
+def main() -> None:
+    DecortVMFolderAccess().run()
+
+
+if __name__ == '__main__':
+    main()

library/decort_vm_folder_list.py

@@ -0,0 +1,71 @@
+#!/usr/bin/python
+
+DOCUMENTATION = r'''
+---
+module: decort_vm_folder_list
+
+description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501
+'''
+
+from dataclasses import dataclass
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.decort_utils import (
+    AnsibleModuleParamSet,
+    DecortController,
+)
+
+
+@dataclass(frozen=True)
+class FilterParam:
+    rg_id: int
+
+
+@dataclass(frozen=True)
+class PaginationParam:
+    page: int | None
+    size: int | None
+
+
+@dataclass(frozen=True)
+class ParamSet(AnsibleModuleParamSet):
+    filter: FilterParam
+    pagination: PaginationParam
+
+
+class DecortVMFolderList(DecortController):
+    param_set: ParamSet
+
+    def __init__(self):
+        super().__init__(AnsibleModule(**self.amodule_init_args))
+        self.param_set = ParamSet.from_dict(self.amodule.params)
+
+    @property
+    def amodule_init_args(self) -> dict:
+        return self.pack_amodule_init_args(
+            argument_spec=ParamSet.get_arg_spec(),
+            supports_check_mode=True,
+        )
+
+    @DecortController.handle_sdk_exceptions
+    def run(self):
+        self.get_info()
+        self.exit()
+
+    def get_info(self):
+        aparam_filter = self.param_set.filter
+        aparam_pagination = self.param_set.pagination
+
+        self.facts = self.api.cloudapi.vmfolder.list(
+            rg_id=aparam_filter.rg_id,
+            page=aparam_pagination.page,
+            size=aparam_pagination.size,
+        ).model_dump()['data']
+
+
+def main():
+    DecortVMFolderList().run()
+
+
+if __name__ == '__main__':
+    main()

library/decort_vm_list.py

@@ -36,12 +36,18 @@ class DecortVMList(DecortController):
                         ext_net_name=dict(
                             type='str',
                         ),
+                        folder_id=dict(
+                            type='int',
+                        ),
                         id=dict(
                             type='int',
                         ),
                         include_deleted=dict(
                             type='bool',
                         ),
+                        include_descendants=dict(
+                            type='bool',
+                        ),
                         ip_addr=dict(
                             type='str',
                         ),
@@ -129,8 +135,10 @@ class DecortVMList(DecortController):
             account_id=aparam_filter['account_id'],
             ext_net_id=aparam_filter['ext_net_id'],
             ext_net_name=aparam_filter['ext_net_name'],
+            folder_id=aparam_filter['folder_id'],
             id=aparam_filter['id'],
             include_deleted=aparam_filter['include_deleted'] or False,
+            include_descendants=aparam_filter['include_descendants'] or False,
             ip_addr=aparam_filter['ip_addr'],
             name=aparam_filter['name'],
             rg_id=aparam_filter['rg_id'],

module_utils/decort_utils.py

@@ -1,4 +1,5 @@
 from copy import deepcopy
+from dataclasses import Field, MISSING, dataclass, fields, is_dataclass
 from datetime import datetime
 from enum import Enum
 from functools import wraps
@@ -6,19 +7,24 @@ from importlib.metadata import version as get_package_version
 import inspect
 import json
 import re
+from types import UnionType
 from typing import (
     Any,
     Callable,
+    ClassVar,
     Iterable,
     Literal,
     Optional,
     ParamSpec,
     TypeVar,
     cast,
+    get_args,
+    get_origin,
 )
 import time
 
 from ansible.module_utils.basic import AnsibleModule, env_fallback
+import dacite
 from dynamix_sdk import BVSAuth, DECS3OAuth, Dynamix
 from dynamix_sdk import __name__ as SDK_PACKAGE_NAME
 from dynamix_sdk import exceptions as sdk_exceptions
@@ -33,6 +39,152 @@ P = ParamSpec('P')
 R = TypeVar('R')
 
 
+@dataclass(frozen=True)
+class AnsibleModuleParamSet:
+    """Dataclass that can generate AnsibleModule argument_spec."""
+
+    _ANSIBLE_TYPE_BY_PYTHON: ClassVar[dict[type, str]] = {
+        bool: 'bool',
+        int: 'int',
+        str: 'str',
+    }
+    _ALLOWED_FIELD_METADATA_KEYS: ClassVar[frozenset[str]] = frozenset({'choices'})  # noqa: E501
+
+    @classmethod
+    def get_arg_spec(cls) -> dict[str, Any]:
+        return cls._dataclass_to_arg_spec(cls)
+
+    @classmethod
+    def from_dict[T: AnsibleModuleParamSet](
+        cls: type[T], data: dict[str, Any]
+    ) -> T:
+        return dacite.from_dict(
+            data_class=cls,
+            data=data,
+            config=dacite.Config(cast=[Enum]),
+        )
+
+    @classmethod
+    def _dataclass_to_arg_spec(cls, model_cls: type) -> dict[str, Any]:
+        spec: dict[str, Any] = {}
+        for field in fields(model_cls):
+            spec[field.name] = cls._field_to_arg_spec(
+                field=field,
+                model_cls=model_cls,
+            )
+        return spec
+
+    @classmethod
+    def _field_to_arg_spec(
+        cls,
+        field: Field,
+        model_cls: type,
+    ) -> dict[str, Any]:
+        if field.default is None:
+            raise ValueError(
+                f'{model_cls.__name__}.{field.name}: '
+                'default=None is not allowed'
+            )
+
+        if get_origin(field.type) is UnionType:
+            args = get_args(field.type)
+            if len(args) != 2 or type(None) not in args:
+                raise ValueError(
+                    f'{model_cls.__name__}.{field.name}: '
+                    f'unsupported annotation {field.type!r}: '
+                    'expected T | None'
+                )
+            param_type = args[0] if args[1] is type(None) else args[1]
+            allows_none = True
+        else:
+            param_type = field.type
+            allows_none = False
+        has_default = (
+            field.default is not MISSING
+            or field.default_factory is not MISSING
+        )
+
+        unknown_metadata_keys = (
+            set(field.metadata) - cls._ALLOWED_FIELD_METADATA_KEYS
+        )
+        if unknown_metadata_keys:
+            raise ValueError(
+                f'{model_cls.__name__}.{field.name}: '
+                f'unsupported metadata keys {sorted(unknown_metadata_keys)!r}'
+            )
+
+        try:
+            entry = cls._type_to_arg_spec(param_type)
+        except ValueError as exc:
+            raise ValueError(
+                f'{model_cls.__name__}.{field.name}: {exc}'
+            )
+
+        if entry['type'] == 'dict' and 'options' in entry:
+            if not allows_none:
+                entry['apply_defaults'] = True
+            return entry
+
+        metadata_choices = field.metadata.get('choices')
+        if metadata_choices is not None:
+            entry['choices'] = list(metadata_choices)
+
+        if not allows_none and not has_default:
+            entry['required'] = True
+        elif has_default:
+            entry['default'] = cls._resolve_default(field)
+
+        return entry
+
+    @classmethod
+    def _type_to_arg_spec(cls, type_: Any) -> dict[str, Any]:
+        if get_origin(type_) is list:
+            list_args = get_args(type_)
+            if len(list_args) != 1:
+                raise ValueError(f'unsupported annotation {type_!r}')
+            elem_spec = cls._type_to_arg_spec(list_args[0])
+            entry: dict[str, Any] = {
+                'type': 'list',
+                'elements': elem_spec['type'],
+            }
+            if 'options' in elem_spec:
+                entry['options'] = elem_spec['options']
+            if 'choices' in elem_spec:
+                entry['choices'] = elem_spec['choices']
+            return entry
+
+        if isinstance(type_, type) and is_dataclass(type_):
+            return {
+                'type': 'dict',
+                'options': cls._dataclass_to_arg_spec(type_),
+            }
+        if isinstance(type_, type) and issubclass(type_, Enum):
+            return {
+                'type': 'str',
+                'choices': list(type_._member_names_),
+            }
+        if get_origin(type_) is Literal:
+            return {
+                'type': 'str',
+                'choices': list(get_args(type_)),
+            }
+        if type_ in cls._ANSIBLE_TYPE_BY_PYTHON:
+            return {'type': cls._ANSIBLE_TYPE_BY_PYTHON[type_]}
+        raise ValueError(f'unsupported annotation {type_!r}')
+
+    @staticmethod
+    def _resolve_default(field: Field) -> Any:
+        if field.default is not MISSING:
+            default = field.default
+        elif field.default_factory is not MISSING:
+            default = field.default_factory()
+        else:
+            raise ValueError(f'{field.name}: no default to resolve')
+        if isinstance(default, Enum):
+            return default.name
+        return default
+
+
 class DecortController(object):
     """DecortController is a utility class that holds target controller context and handles API requests formatting
     based on the requested authentication type.
@@ -53,7 +205,7 @@ class DecortController(object):
     _api: sdk_types.API | None = None
     _usermanager_whoami_result: None | dict = None
 
-    ANSIBLE_MODULES_VERSION = '13.0.1'
+    ANSIBLE_MODULES_VERSION = '13.0.2'
     COMPATIBLE_SDK_MINOR_VERSION = '2.0'
 
     VM_RESIZE_NOT = 0

requirements.txt

@@ -1,3 +1,4 @@
 ansible==11.6.0
+dacite==1.9.2
 requests==2.32.3
 git+https://repository.basistech.ru/BASIS/dynamix-python-sdk.git@2.0.latest