Prepare release 2.2.0

Prepare data for the v2.2.0 release with few new modules and bugfixes.

Change-Id: Id593b623b389cedb140fb05e8063f48ef7eacc36

CHANGELOG.rst

@@ -5,6 +5,27 @@ Ansible OpenStack Collection Release Notes
 .. contents:: Topics
 
 
+v2.2.0
+======
+
+Release Summary
+---------------
+
+New module for volume_type and bugfixes
+
+Minor Changes
+-------------
+
+- Add volume_encryption_type modules
+- Add volume_type modules
+
+Bugfixes
+--------
+
+- Fix image module filter
+- Fix port module idempotency
+- Fix router module idempotency
+
 v2.1.0
 ======
 

changelogs/changelog.yaml

@@ -515,3 +515,14 @@ releases:
       - Highlight our mode of operation more prominently
       release_summary: New module for Ironic and bugfixes
     release_date: '2023-04-19'
+  2.2.0:
+    changes:
+      bugfixes:
+      - Fix image module filter
+      - Fix port module idempotency
+      - Fix router module idempotency
+      minor_changes:
+      - Add volume_encryption_type modules
+      - Add volume_type modules
+      release_summary: New module for volume_type and bugfixes
+    release_date: '2023-12-01'

ci/roles/image/tasks/main.yml

@@ -24,6 +24,13 @@
         path: '{{ tmp_file.path }}'
         size: 1M
 
+    - name: Calculating file checksum
+      ansible.builtin.stat:
+        path: "{{ tmp_file.path }}"
+        checksum_algorithm: sha512
+        get_checksum: true
+      register: image_details
+
     - name: Ensure mock kernel and ramdisk images (defaults)
       openstack.cloud.image:
          cloud: "{{ cloud }}"
@@ -42,6 +49,7 @@
          name: ansible_image
          filename: "{{ tmp_file.path }}"
          is_protected: true
+         checksum: "{{ image_details.stat.checksum }}"
          disk_format: raw
          tags:
            - test

ci/roles/router/tasks/main.yml

@@ -384,7 +384,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.100
 
 - name: Gather routers info
@@ -412,7 +412,7 @@
      external_gateway_info:
         network: "{{ external_network_name }}"
         external_fixed_ips:
-           - subnet: shade_subnet5
+           - subnet_id: shade_subnet5
              ip: 10.6.6.100
            - subnet: shade_subnet5
              ip: 10.6.6.101
@@ -426,7 +426,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.100
         - subnet: shade_subnet5
           ip: 10.6.6.101
@@ -461,7 +461,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.101
 
 - name: Update router (remove external fixed ips) again
@@ -473,7 +473,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.101
   register: router
 
@@ -506,7 +506,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.101
 
 - name: Gather routers info
@@ -533,7 +533,7 @@
          - shade_subnet1
      network: "{{ external_network_name }}"
      external_fixed_ips:
-        - subnet: shade_subnet5
+        - subnet_id: shade_subnet5
           ip: 10.6.6.101
   register: router
 

ci/roles/security_group/tasks/main.yml

@@ -72,4 +72,25 @@
      name: ansible_security_group
      state: absent
 
+- name: Create stateless security group
+  openstack.cloud.security_group:
+     cloud: "{{ cloud }}"
+     name: ansible_security_group_stateless
+     stateful: false
+     state: present
+     description: 'Created from Ansible playbook'
+  register: security_group_stateless
+
+- name: Assert return values of security_group module
+  assert:
+    that:
+      - security_group_stateless.security_group.name == 'ansible_security_group_stateless'
+      - security_group_stateless.security_group.stateful == False
+
+- name: Delete stateless security group
+  openstack.cloud.security_group:
+     cloud: "{{ cloud }}"
+     name: ansible_security_group_stateless
+     state: absent
+
 - include_tasks: rules.yml

ci/roles/volume_type/defaults/main.yml

@@ -0,0 +1,10 @@
+---
+volume_backend_name: LVM_iSCSI
+volume_type_name: test_type
+volume_type_description: Test volume type
+
+enc_provider_name: nova.volume.encryptors.luks.LuksEncryptor
+enc_cipher: aes-xts-plain64
+enc_control_location: front-end
+enc_control_alt_location: back-end
+enc_key_size: 256

ci/roles/volume_type/tasks/main.yml

@@ -0,0 +1,85 @@
+---
+- name: Create volume type
+  openstack.cloud.volume_type:
+    name: "{{ volume_type_name }}"
+    cloud: "{{ cloud }}"
+    state: present
+    extra_specs:
+      volume_backend_name: "{{ volume_backend_name }}"
+    description: "{{ volume_type_description }}"
+    is_public: true
+  register: the_result
+- name: Check created volume type
+  vars:
+    the_volume: "{{ the_result.volume_type }}"
+  ansible.builtin.assert:
+    that:
+      - "'id' in the_result.volume_type"
+      - the_volume.description == volume_type_description
+      - the_volume.is_public == True
+      - the_volume.name == volume_type_name
+      - the_volume.extra_specs['volume_backend_name'] == volume_backend_name
+    success_msg: >-
+      Created volume: {{ the_result.volume_type.id }},
+      Name: {{ the_result.volume_type.name }},
+      Description: {{ the_result.volume_type.description }}
+
+- name: Test, check idempotency
+  openstack.cloud.volume_type:
+    name: "{{ volume_type_name }}"
+    cloud: "{{ cloud }}"
+    state: present
+    extra_specs:
+      volume_backend_name: "{{ volume_backend_name }}"
+    description: "{{ volume_type_description }}"
+    is_public: true
+  register: the_result
+- name: Check result.changed is false
+  ansible.builtin.assert:
+    that:
+      - the_result.changed == false
+    success_msg: "Request with the same details lead to no changes"
+
+- name: Add extra spec
+  openstack.cloud.volume_type:
+    cloud: "{{ cloud }}"
+    name: "{{ volume_type_name }}"
+    state: present
+    extra_specs:
+      volume_backend_name: "{{ volume_backend_name }}"
+      some_spec: fake_spec
+    description: "{{ volume_type_description }}"
+    is_public: true
+  register: the_result
+- name: Check volume type extra spec
+  ansible.builtin.assert:
+    that:
+      - "'some_spec' in the_result.volume_type.extra_specs"
+      - the_result.volume_type.extra_specs["some_spec"] == "fake_spec"
+    success_msg: >-
+      New extra specs: {{ the_result.volume_type.extra_specs }}
+
+# is_public update attempt using openstacksdk result in unexpected attribute
+# error... TODO: Find solution
+#
+# - name: Make volume type private
+#   openstack.cloud.volume_type:
+#     cloud: "{{ cloud }}"
+#     name: "{{ volume_type_alt_name }}"
+#     state: present
+#     extra_specs:
+#       volume_backend_name: "{{ volume_backend_name }}"
+#       # some_other_spec: test
+#     description: Changed 3rd time test volume type
+#     is_public: true
+#   register: the_result
+
+- name: Volume encryption tests
+  ansible.builtin.include_tasks: volume_encryption.yml
+
+- name: Delete volume type
+  openstack.cloud.volume_type:
+    cloud: "{{ cloud }}"
+    name: "{{ volume_type_name }}"
+    state: absent
+  register: the_result

ci/roles/volume_type/tasks/volume_encryption.yml

@@ -0,0 +1,67 @@
+---
+- name: Test, Volume type has no encryption
+  openstack.cloud.volume_type_info:
+    cloud: "{{ cloud }}"
+    name: "{{ volume_type_name }}"
+  register: the_result
+- name: Check volume type has no encryption
+  ansible.builtin.assert:
+    that:
+      - the_result.encryption.id == None
+    success_msg: >-
+      Success: Volume type has no encryption at the moment
+
+- name: Test, create volume type encryption
+  openstack.cloud.volume_type_encryption:
+    cloud: "{{ cloud }}"
+    volume_type: "{{ volume_type_name }}"
+    state: present
+    encryption_provider: "{{ enc_provider_name }}"
+    encryption_cipher: "{{ enc_cipher }}"
+    encryption_control_location: "{{ enc_control_location }}"
+    encryption_key_size: "{{ enc_key_size }}"
+  register: the_result
+- name: Check volume type encryption
+  ansible.builtin.assert:
+    that:
+      - the_result.encryption.cipher == enc_cipher
+      - the_result.encryption.control_location == enc_control_location
+      - the_result.encryption.key_size == enc_key_size
+      - the_result.encryption.provider == enc_provider_name
+    success_msg: >-
+      Success: {{ the_result.encryption.encryption_id }}
+
+- name: Test, update volume type encryption
+  openstack.cloud.volume_type_encryption:
+    cloud: "{{ cloud }}"
+    volume_type: "{{ volume_type_name }}"
+    state: present
+    encryption_provider: "{{ enc_provider_name }}"
+    encryption_cipher: "{{ enc_cipher }}"
+    encryption_control_location: "{{ enc_control_alt_location }}"
+    encryption_key_size: "{{ enc_key_size }}"
+  register: the_result
+- name: Check volume type encryption change
+  ansible.builtin.assert:
+    that:
+      - the_result.encryption.control_location == enc_control_alt_location
+    success_msg: >-
+      New location: {{ the_result.encryption.control_location }}
+
+- name: Test, delete volume type encryption
+  openstack.cloud.volume_type_encryption:
+    cloud: "{{ cloud }}"
+    volume_type: "{{ volume_type_name }}"
+    state: absent
+  register: the_result
+- name: Get volume type details
+  openstack.cloud.volume_type_info:
+    cloud: "{{ cloud }}"
+    name: "{{ volume_type_name }}"
+  register: the_result
+- name: Check volume type has no encryption
+  ansible.builtin.assert:
+    that:
+      - the_result.encryption.id == None
+    success_msg: >-
+      Success: Volume type has no encryption

ci/run-collection.yml

@@ -53,6 +53,7 @@
     - { role: subnet, tags: subnet }
     - { role: subnet_pool, tags: subnet_pool }
     - { role: volume, tags: volume }
+    - { role: volume_type, tags: volume_type }
     - { role: volume_backup, tags: volume_backup }
     - { role: volume_snapshot, tags: volume_snapshot }
     - { role: volume_type_access, tags: volume_type_access }

galaxy.yml

@@ -32,4 +32,4 @@ build_ignore:
   - .vscode
   - ansible_collections_openstack.egg-info
   - changelogs
-version: 2.1.0
+version: 2.2.0

plugins/inventory/openstack.py

@@ -271,9 +271,9 @@ class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
         if not attempt_to_read_cache or cache_needs_update:
             self.display.vvvv('Retrieving servers from Openstack clouds')
             clouds_yaml_path = self.get_option('clouds_yaml_path')
-            config_files = (
-                openstack.config.loader.CONFIG_FILES
-                + ([clouds_yaml_path] if clouds_yaml_path else []))
+            config_files = openstack.config.loader.CONFIG_FILES
+            if clouds_yaml_path:
+                config_files += clouds_yaml_path
 
             config = openstack.config.loader.OpenStackConfig(
                 config_files=config_files)

plugins/modules/image.py

@@ -485,7 +485,7 @@ class ImageModule(OpenStackModule):
         if image_name_or_id:
             image = self.conn.get_image(
                 image_name_or_id,
-                filters={(k, self.params[k])
+                filters={k: self.params[k]
                          for k in ['checksum'] if self.params[k] is not None})
 
         changed = False

plugins/modules/port.py

@@ -138,10 +138,11 @@ options:
             of I(no_security_groups): C(true)."
         type: bool
         default: 'false'
-    port_security_enabled:
+    is_port_security_enabled:
         description:
           - Whether to enable or disable the port security on the network.
         type: bool
+        aliases: ['port_security_enabled']
     security_groups:
         description:
           - Security group(s) ID(s) or name(s) associated with the port.
@@ -479,7 +480,7 @@ class PortModule(OpenStackModule):
         name=dict(required=True),
         network=dict(),
         no_security_groups=dict(default=False, type='bool'),
-        port_security_enabled=dict(type='bool'),
+        is_port_security_enabled=dict(type='bool', aliases=['port_security_enabled']),
         security_groups=dict(type='list', elements='str'),
         state=dict(default='present', choices=['absent', 'present']),
     )
@@ -655,7 +656,7 @@ class PortModule(OpenStackModule):
                   'extra_dhcp_opts',
                   'is_admin_state_up',
                   'mac_address',
-                  'port_security_enabled',
+                  'is_port_security_enabled',
                   'fixed_ips',
                   'name']:
             if self.params[k] is not None:

plugins/modules/router.py

@@ -366,38 +366,36 @@ class RouterModule(OpenStackModule):
             if 'ip_address' in p:
                 cur_fip_map[p['subnet_id']].add(p['ip_address'])
         req_fip_map = defaultdict(set)
-        for p in external_fixed_ips:
-            if 'ip_address' in p:
-                req_fip_map[p['subnet_id']].add(p['ip_address'])
+        if external_fixed_ips is not None:
+            # User passed expected external_fixed_ips configuration.
+            # Build map of requested ips/subnets.
+            for p in external_fixed_ips:
+                if 'ip_address' in p:
+                    req_fip_map[p['subnet_id']].add(p['ip_address'])
 
-        # Check if external ip addresses need to be added
-        for fip in external_fixed_ips:
-            subnet = fip['subnet_id']
-            ip = fip.get('ip_address', None)
-            if subnet in cur_fip_map:
-                if ip is not None and ip not in cur_fip_map[subnet]:
-                    # mismatching ip for subnet
+            # Check if external ip addresses need to be added
+            for fip in external_fixed_ips:
+                subnet = fip['subnet_id']
+                ip = fip.get('ip_address', None)
+                if subnet in cur_fip_map:
+                    if ip is not None and ip not in cur_fip_map[subnet]:
+                        # mismatching ip for subnet
+                        return True
+                else:
+                    # adding ext ip with subnet 'subnet'
                     return True
-            else:
-                # adding ext ip with subnet 'subnet'
-                return True
 
-        # Check if external ip addresses need to be removed
-        for fip in cur_ext_fips:
-            subnet = fip['subnet_id']
-            ip = fip['ip_address']
-            if subnet in req_fip_map:
-                if ip not in req_fip_map[subnet]:
-                    # removing ext ip with subnet (ip clash)
+            # Check if external ip addresses need to be removed.
+            for fip in cur_ext_fips:
+                subnet = fip['subnet_id']
+                ip = fip['ip_address']
+                if subnet in req_fip_map:
+                    if ip not in req_fip_map[subnet]:
+                        # removing ext ip with subnet (ip clash)
+                        return True
+                else:
+                    # removing ext ip with subnet
                     return True
-            else:
-                # removing ext ip with subnet
-                return True
-
-        if not external_fixed_ips and len(cur_ext_fips) > 1:
-            # No external fixed ips requested but
-            # router has several external fixed ips
-            return True
 
         # Check if internal interfaces need update
         if to_add or to_remove or missing_port_ids:
@@ -448,7 +446,8 @@ class RouterModule(OpenStackModule):
         return kwargs
 
     def _build_router_interface_config(self, filters):
-        external_fixed_ips = []
+        # Undefine external_fixed_ips to have possibility to unset them
+        external_fixed_ips = None
         internal_ports_missing = []
         internal_ifaces = []
 
@@ -459,9 +458,11 @@ class RouterModule(OpenStackModule):
                 .get('external_fixed_ips')
         ext_fixed_ips = ext_fixed_ips or self.params['external_fixed_ips']
         if ext_fixed_ips:
+            # User passed external_fixed_ips configuration. Initialize ips list
+            external_fixed_ips = []
             for iface in ext_fixed_ips:
                 subnet = self.conn.network.find_subnet(
-                    iface['subnet'], ignore_missing=False, **filters)
+                    iface['subnet_id'], ignore_missing=False, **filters)
                 fip = dict(subnet_id=subnet.id)
                 if 'ip_address' in iface:
                     fip['ip_address'] = iface['ip_address']

plugins/modules/security_group.py

@@ -113,6 +113,10 @@ options:
     choices: [present, absent]
     default: present
     type: str
+  stateful:
+    description:
+      - Should the resource be stateful or stateless.
+    type: bool
 extends_documentation_fragment:
   - openstack.cloud.openstack
 '''
@@ -201,6 +205,14 @@ EXAMPLES = r'''
     name: foo
     description: security group for foo servers
 
+- name: Create a stateless security group
+  openstack.cloud.security_group:
+    cloud: mordred
+    state: present
+    stateful: false
+    name: foo
+    description: stateless security group for foo servers
+
 - name: Update the existing 'foo' security group description
   openstack.cloud.security_group:
     cloud: mordred
@@ -260,6 +272,7 @@ class SecurityGroupModule(OpenStackModule):
             ),
         ),
         state=dict(default='present', choices=['absent', 'present']),
+        stateful=dict(type="bool"),
     )
 
     module_kwargs = dict(
@@ -405,7 +418,7 @@ class SecurityGroupModule(OpenStackModule):
 
     def _create(self):
         kwargs = dict((k, self.params[k])
-                      for k in ['description', 'name']
+                      for k in ['description', 'name', 'stateful']
                       if self.params[k] is not None)
 
         project_name_or_id = self.params['project']

plugins/modules/volume_type.py

@@ -0,0 +1,241 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2023 Cleura AB
+# GNU General Public License v3.0+
+# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r'''
+---
+module: volume_type
+short_description: Manage OpenStack volume type
+author: OpenStack Ansible SIG
+description:
+  - Add, remove or update volume types in OpenStack.
+options:
+  name:
+    description:
+      - Volume type name or id.
+    required: true
+    type: str
+  description:
+    description:
+      - Description of the volume type.
+    type: str
+  extra_specs:
+    description:
+      - List of volume type properties
+    type: dict
+  is_public:
+    description:
+      - Make volume type accessible to the public.
+      - Can be set only during creation
+    type: bool
+  state:
+    description:
+      - Indicate desired state of the resource.
+      - When I(state) is C(present), then I(is_public) is required.
+    choices: ['present', 'absent']
+    default: present
+    type: str
+extends_documentation_fragment:
+  - openstack.cloud.openstack
+'''
+
+EXAMPLES = r'''
+    - name: Delete volume type by name
+      openstack.cloud.volume_type:
+        name: test_type
+        state: absent
+
+    - name: Delete volume type by id
+      openstack.cloud.volume_type:
+        name: fbadfa6b-5f17-4c26-948e-73b94de57b42
+        state: absent
+
+    - name: Create volume type
+      openstack.cloud.volume_type:
+        name: unencrypted_volume_type
+        state: present
+        extra_specs:
+          volume_backend_name: LVM_iSCSI
+        description: Unencrypted volume type
+        is_public: True
+'''
+
+RETURN = '''
+volume_type:
+  description: Dictionary describing volume type
+  returned: On success when I(state) is 'present'
+  type: dict
+  contains:
+    name:
+      description: volume type name
+      returned: success
+      type: str
+      sample: test_type
+    extra_specs:
+      description: volume type extra parameters
+      returned: success
+      type: dict
+      sample: null
+    is_public:
+      description: whether the volume type is public
+      returned: success
+      type: bool
+      sample: True
+    description:
+      description: volume type description
+      returned: success
+      type: str
+      sample: Unencrypted volume type
+'''
+
+from ansible_collections.openstack.cloud.plugins.module_utils.openstack import OpenStackModule
+
+
+class VolumeTypeModule(OpenStackModule):
+    argument_spec = dict(
+        name=dict(type='str', required=True),
+        description=dict(type='str', required=False),
+        extra_specs=dict(type='dict', required=False),
+        is_public=dict(type='bool'),
+        state=dict(
+            type='str', default='present', choices=['absent', 'present']),
+    )
+    module_kwargs = dict(
+        required_if=[('state', 'present', ['is_public'])],
+        supports_check_mode=True,
+    )
+
+    @staticmethod
+    def _extract_result(details):
+        if details is not None:
+            return details.to_dict(computed=False)
+        return {}
+
+    def run(self):
+        state = self.params['state']
+        name_or_id = self.params['name']
+        volume_type = self.conn.block_storage.find_type(name_or_id)
+
+        if self.ansible.check_mode:
+            self.exit_json(
+                changed=self._will_change(state, volume_type))
+
+        if state == 'present' and not volume_type:
+            # Create type
+            create_result = self._create()
+            volume_type = self._extract_result(create_result)
+            self.exit_json(changed=True, volume_type=volume_type)
+
+        elif state == 'present' and volume_type:
+            # Update type
+            update = self._build_update(volume_type)
+            update_result = self._update(volume_type, update)
+            volume_type = self._extract_result(update_result)
+            self.exit_json(changed=bool(update), volume_type=volume_type)
+
+        elif state == 'absent' and volume_type:
+            # Delete type
+            self._delete(volume_type)
+            self.exit_json(changed=True)
+
+    def _build_update(self, volume_type):
+        return {
+            **self._build_update_extra_specs(volume_type),
+            **self._build_update_volume_type(volume_type)}
+
+    def _build_update_extra_specs(self, volume_type):
+        update = {}
+
+        old_extra_specs = volume_type['extra_specs']
+        new_extra_specs = self.params['extra_specs'] or {}
+
+        delete_extra_specs_keys = \
+            set(old_extra_specs.keys()) - set(new_extra_specs.keys())
+
+        if delete_extra_specs_keys:
+            update['delete_extra_specs_keys'] = delete_extra_specs_keys
+
+        stringified = {k: str(v) for k, v in new_extra_specs.items()}
+
+        if old_extra_specs != stringified:
+            update['create_extra_specs'] = new_extra_specs
+
+        return update
+
+    def _build_update_volume_type(self, volume_type):
+        update = {}
+        allowed_attributes = [
+            'is_public', 'description', 'name']
+        type_attributes = {
+            k: self.params[k]
+            for k in allowed_attributes
+            if k in self.params and self.params.get(k) is not None
+            and self.params.get(k) != volume_type.get(k)}
+
+        if type_attributes:
+            update['type_attributes'] = type_attributes
+
+        return update
+
+    def _create(self):
+        kwargs = {k: self.params[k]
+                  for k in ['name', 'is_public', 'description', 'extra_specs']
+                  if self.params.get(k) is not None}
+        volume_type = self.conn.block_storage.create_type(**kwargs)
+        return volume_type
+
+    def _delete(self, volume_type):
+        self.conn.block_storage.delete_type(volume_type.id)
+
+    def _update(self, volume_type, update):
+        if not update:
+            return volume_type
+        volume_type = self._update_volume_type(volume_type, update)
+        volume_type = self._update_extra_specs(volume_type, update)
+        return volume_type
+
+    def _update_extra_specs(self, volume_type, update):
+        delete_extra_specs_keys = update.get('delete_extra_specs_keys')
+        if delete_extra_specs_keys:
+            self.conn.block_storage.delete_type_extra_specs(
+                volume_type, delete_extra_specs_keys)
+            # refresh volume_type information
+            volume_type = self.conn.block_storage.find_type(volume_type.id)
+
+        create_extra_specs = update.get('create_extra_specs')
+        if create_extra_specs:
+            self.conn.block_storage.update_type_extra_specs(
+                volume_type, **create_extra_specs)
+            # refresh volume_type information
+            volume_type = self.conn.block_storage.find_type(volume_type.id)
+
+        return volume_type
+
+    def _update_volume_type(self, volume_type, update):
+        type_attributes = update.get('type_attributes')
+        if type_attributes:
+            updated_type = self.conn.block_storage.update_type(
+                volume_type, **type_attributes)
+            return updated_type
+        return volume_type
+
+    def _will_change(self, state, volume_type):
+        if state == 'present' and not volume_type:
+            return True
+        if state == 'present' and volume_type:
+            return bool(self._build_update(volume_type))
+        if state == 'absent' and volume_type:
+            return True
+        return False
+
+
+def main():
+    module = VolumeTypeModule()
+    module()
+
+
+if __name__ == '__main__':
+    main()

plugins/modules/volume_type_encryption.py

@@ -0,0 +1,233 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2023 Cleura AB
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r'''
+---
+module: volume_type_encryption
+short_description: Manage OpenStack volume type encryption
+author: OpenStack Ansible SIG
+description:
+  - Add, remove or update volume type encryption in OpenStack.
+options:
+  volume_type:
+    description:
+      - Volume type name or id.
+    required: true
+    type: str
+  state:
+    description:
+      - Indicate desired state of the resource.
+      - When I(state) is C(present), then I(encryption options) are required.
+    choices: ['present', 'absent']
+    default: present
+    type: str
+  encryption_provider:
+    description:
+      - class that provides encryption support for the volume type
+      - admin only
+    type: str
+  encryption_cipher:
+    description:
+      - encryption algorithm or mode
+      - admin only
+    type: str
+  encryption_control_location:
+    description:
+      - Set the notional service where the encryption is performed
+      - admin only
+    choices: ['front-end', 'back-end']
+    type: str
+  encryption_key_size:
+    description:
+      - Set the size of the encryption key of this volume type
+      - admin only
+    choices: [128, 256, 512]
+    type: int
+extends_documentation_fragment:
+  - openstack.cloud.openstack
+'''
+
+EXAMPLES = r'''
+    - name: Create volume type encryption
+      openstack.cloud.volume_type_encryption:
+        volume_type: test_type
+        state: present
+        encryption_provider: nova.volume.encryptors.luks.LuksEncryptor
+        encryption_cipher: aes-xts-plain64
+        encryption_control_location: front-end
+        encryption_key_size: 256
+
+    - name: Delete volume type encryption
+      openstack.cloud.volume_type_encryption:
+        volume_type: test_type
+        state: absent
+      register: the_result
+'''
+
+RETURN = '''
+encryption:
+  description: Dictionary describing volume type encryption
+  returned: On success when I(state) is 'present'
+  type: dict
+  contains:
+    cipher:
+      description: encryption cipher
+      returned: success
+      type: str
+      sample: aes-xts-plain64
+    control_location:
+      description: encryption location
+      returned: success
+      type: str
+      sample: front-end
+    created_at:
+      description: Resource creation date and time
+      returned: success
+      type: str
+      sample: "2023-08-04T10:23:03.000000"
+    deleted:
+      description: Boolean if the resource was deleted
+      returned: success
+      type: str
+      sample: false,
+    deleted_at:
+      description: Resource delete date and time
+      returned: success
+      type: str
+      sample: null,
+    encryption_id:
+      description: UUID of the volume type encryption
+      returned: success
+      type: str
+      sample: b75d8c5c-a6d8-4a5d-8c86-ef4f1298525d
+    id:
+      description: Alias to encryption_id
+      returned: success
+      type: str
+      sample: b75d8c5c-a6d8-4a5d-8c86-ef4f1298525d
+    key_size:
+      description: Size of the key
+      returned: success
+      type: str
+      sample: 256,
+    provider:
+      description: Encryption provider
+      returned: success
+      type: str
+      sample: "nova.volume.encryptors.luks.LuksEncryptor"
+    updated_at:
+      description: Resource last update date and time
+      returned: success
+      type: str
+      sample: null
+'''
+
+from ansible_collections.openstack.cloud.plugins.module_utils.openstack import OpenStackModule
+
+
+class VolumeTypeModule(OpenStackModule):
+    argument_spec = dict(
+        volume_type=dict(type='str', required=True),
+        state=dict(
+            type='str', default='present', choices=['absent', 'present']),
+        encryption_provider=dict(type='str', required=False),
+        encryption_cipher=dict(type='str', required=False),
+        encryption_control_location=dict(
+            type='str', choices=['front-end', 'back-end'], required=False),
+        encryption_key_size=dict(
+            type='int', choices=[128, 256, 512], required=False),
+    )
+    module_kwargs = dict(
+        required_if=[('state', 'present', [
+            'encryption_provider', 'encryption_cipher',
+            'encryption_control_location', 'encryption_key_size'])],
+        supports_check_mode=True,
+    )
+
+    @staticmethod
+    def _extract_result(details):
+        if details is not None:
+            return details.to_dict(computed=False)
+        return {}
+
+    def run(self):
+        state = self.params['state']
+        name = self.params['volume_type']
+        volume_type = self.conn.block_storage.find_type(name)
+
+        # TODO: Add get type_encryption by id
+        type_encryption = self.conn.block_storage.get_type_encryption(
+            volume_type.id)
+        encryption_id = type_encryption.get('encryption_id')
+
+        if self.ansible.check_mode:
+            self.exit_json(
+                changed=self._will_change(state, encryption_id))
+
+        if state == 'present':
+            update = self._build_update_type_encryption(type_encryption)
+            if not bool(update):
+                # No change is required
+                self.exit_json(changed=False)
+
+            if not encryption_id:  # Create new type encryption
+                result = self.conn.block_storage.create_type_encryption(
+                    volume_type, **update)
+            else:  # Update existing type encryption
+                result = self.conn.block_storage.update_type_encryption(
+                    encryption=type_encryption, **update)
+            encryption = self._extract_result(result)
+            self.exit_json(changed=bool(update), encryption=encryption)
+        elif encryption_id is not None:
+            # absent state requires type encryption delete
+            self.conn.block_storage.delete_type_encryption(type_encryption)
+            self.exit_json(changed=True)
+
+    def _build_update_type_encryption(self, type_encryption):
+        attributes_map = {
+            'encryption_provider': 'provider',
+            'encryption_cipher': 'cipher',
+            'encryption_key_size': 'key_size',
+            'encryption_control_location': 'control_location'}
+
+        encryption_attributes = {
+            attributes_map[k]: self.params[k]
+            for k in self.params
+            if k in attributes_map.keys() and self.params.get(k) is not None
+            and self.params.get(k) != type_encryption.get(attributes_map[k])}
+
+        if 'encryption_provider' in encryption_attributes.keys():
+            encryption_attributes['provider'] = \
+                encryption_attributes['encryption_provider']
+
+        return encryption_attributes
+
+    def _update_type_encryption(self, type_encryption, update):
+        if update:
+            updated_type = self.conn.block_storage.update_type_encryption(
+                encryption=type_encryption,
+                **update)
+            return updated_type
+        return {}
+
+    def _will_change(self, state, type_encryption):
+        encryption_id = type_encryption.get('encryption_id')
+        if state == 'present' and not encryption_id:
+            return True
+        if state == 'present' and encryption_id is not None:
+            return bool(self._build_update_type_encryption(type_encryption))
+        if state == 'absent' and encryption_id is not None:
+            return True
+        return False
+
+
+def main():
+    module = VolumeTypeModule()
+    module()
+
+
+if __name__ == '__main__':
+    main()

plugins/modules/volume_type_info.py

@@ -0,0 +1,175 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2023 Cleura AB
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r'''
+---
+module: volume_type_info
+short_description: Get OpenStack volume type details
+author: OpenStack Ansible SIG
+description:
+  - Get volume type details in OpenStack.
+  - Get volume type encryption details in OpenStack
+options:
+  name:
+    description:
+      - Volume type name or id.
+    required: true
+    type: str
+extends_documentation_fragment:
+  - openstack.cloud.openstack
+'''
+
+EXAMPLES = r'''
+    - name: Get volume type details
+      openstack.cloud.volume_type_info:
+        name: test_type
+
+    - name: Get volume type details by id
+      openstack.cloud.volume_type_info:
+        name: fbadfa6b-5f17-4c26-948e-73b94de57b42
+'''
+
+RETURN = '''
+access_project_ids:
+  description:
+    - List of project IDs allowed to access volume type
+    - Public volume types returns 'null' value as it is not applicable
+  returned: On success when I(state) is 'present'
+  type: list
+  elements: str
+volume_type:
+  description: Dictionary describing volume type
+  returned: On success when I(state) is 'present'
+  type: dict
+  contains:
+    id:
+      description: volume_type uuid
+      returned: success
+      type: str
+      sample: b75d8c5c-a6d8-4a5d-8c86-ef4f1298525d
+    name:
+      description: volume type name
+      returned: success
+      type: str
+      sample: test_type
+    extra_specs:
+      description: volume type extra parameters
+      returned: success
+      type: dict
+      sample: null
+    is_public:
+      description: whether the volume type is public
+      returned: success
+      type: bool
+      sample: True
+    description:
+      description: volume type description
+      returned: success
+      type: str
+      sample: Unencrypted volume type
+encryption:
+  description: Dictionary describing volume type encryption
+  returned: On success when I(state) is 'present'
+  type: dict
+  contains:
+    cipher:
+      description: encryption cipher
+      returned: success
+      type: str
+      sample: aes-xts-plain64
+    control_location:
+      description: encryption location
+      returned: success
+      type: str
+      sample: front-end
+    created_at:
+      description: Resource creation date and time
+      returned: success
+      type: str
+      sample: "2023-08-04T10:23:03.000000"
+    deleted:
+      description: Boolean if the resource was deleted
+      returned: success
+      type: str
+      sample: false
+    deleted_at:
+      description: Resource delete date and time
+      returned: success
+      type: str
+      sample: null
+    encryption_id:
+      description: UUID of the volume type encryption
+      returned: success
+      type: str
+      sample: b75d8c5c-a6d8-4a5d-8c86-ef4f1298525d
+    id:
+      description: Alias to encryption_id
+      returned: success
+      type: str
+      sample: b75d8c5c-a6d8-4a5d-8c86-ef4f1298525d
+    key_size:
+      description: Size of the key
+      returned: success
+      type: str
+      sample: 256
+    provider:
+      description: Encryption provider
+      returned: success
+      type: str
+      sample: "nova.volume.encryptors.luks.LuksEncryptor"
+    updated_at:
+      description: Resource last update date and time
+      returned: success
+      type: str
+      sample: null
+'''
+
+from ansible_collections.openstack.cloud.plugins.module_utils.openstack import OpenStackModule
+
+
+class VolumeTypeModule(OpenStackModule):
+    argument_spec = dict(
+        name=dict(type='str', required=True)
+    )
+    module_kwargs = dict(
+        supports_check_mode=True,
+    )
+
+    @staticmethod
+    def _extract_result(details):
+        if details is not None:
+            return details.to_dict(computed=False)
+        return {}
+
+    def run(self):
+        name_or_id = self.params['name']
+        volume_type = self.conn.block_storage.find_type(name_or_id)
+
+        type_encryption = self.conn.block_storage.get_type_encryption(
+            volume_type.id)
+
+        if volume_type.is_public:
+            type_access = None
+        else:
+            type_access = [
+                proj['project_id']
+                for proj in self.conn.block_storage.get_type_access(
+                    volume_type.id)]
+
+        self.exit_json(
+            changed=False,
+            volume_type=self._extract_result(volume_type),
+            encryption=self._extract_result(type_encryption),
+            access_project_ids=type_access)
+
+
+def main():
+    module = VolumeTypeModule()
+    module()
+
+
+if __name__ == '__main__':
+    main()

tests/unit/mock/loader.py

@@ -28,7 +28,7 @@ class DictDataLoader(DataLoader):
 
     def __init__(self, file_mapping=None):
         file_mapping = {} if file_mapping is None else file_mapping
-        assert type(file_mapping) == dict
+        assert isinstance(file_mapping, dict)
 
         super(DictDataLoader, self).__init__()