ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-02-04 07:51:50 +00:00
Code Activity

lxd_storage_pool_info, lxd_storage_volume_info: new modules (#11198)

Browse source 
* Fix mistaken rebase

* plugins/modules/lxd_storage_: include error codes, clean up notes

* plugins/modules/lxd_storage_: snap_url, ruff fix

* plugins/modules/lxd_storage_volume_info.py: remove checks on expected api returned bits

* plugins/modules/lxd_storage_volume_info.py: required: true

* tests/integration/targets/lxd_storage_volume_info/tasks/main.yaml: add Test fetching specific volume by name

* tests/unit/plugins/modules/test_lxd_storage_: add unit tests

* tests/integration/targets/lxd_storage_pool_info/tasks/main.yaml: add integratio tests

* tests/integration/targets/lxd_storage_: not required

* tests/integration/targets/lxd_storage_: not required perhaps, lxd_project has them

* tests/unit/plugins/modules/test_lxd_storage_volume_info.py: fix python3.8 tests

* tests/unit/plugins/modules/test_lxd_storage_pool_info.py: fix python3.8

* tests/integration/targets/lxd_storage_: correct paths for aliases

* tests/unit/plugins/modules/test_lxd_storage_volume_info.py: remove backticks

* tests/unit/plugins/modules/test_lxd_storage_volume_info.py: remove blank line

* tests/unit/plugins/modules/test_lxd_storage_: python3.8 changes

* tests/unit/plugins/modules/test_lxd_storage_: python3.8 changes

* tests/unit/plugins/lookup/test_github_app_access_token.py: restore

* tests/unit/plugins/connection/test_wsl.py: restore

* plugins/modules/lxd_storage_: use ANSIBLE_LXD_DEFAULT_SNAP_URL and put API version into const

* lxd_storage_volume_info: use recursion to gather all volume details

* tests/integration/targets/lxd_storage_volume_info/tasks/main.yaml: fix silet skipped failures

* tests/integration/targets/lxd_storage_pool_info/tasks/main.yaml: fix silet failures

* lxd_storage_pool_info: update to use recursion to gather all details in one shot

* Remove unnecessary change.

---------

Co-authored-by: Felix Fontein <felix@fontein.de>
This commit is contained in:
Sean McAvoy 2025-12-01 02:58:45 -03:00 • committed by GitHub
parent 16d51a8233
commit 6365b5a981
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 1243 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.github/BOTMETA.yml vendored
View file

@ -931,6 +931,10 @@ files:
maintainers: conloos
$modules/lxd_project.py:
maintainers: we10710aa
$modules/lxd_storage_pool_info.py:
maintainers: smcavoy
$modules/lxd_storage_volume_info.py:
maintainers: smcavoy
$modules/macports.py:
ignore: ryansb
keywords: brew cask darwin homebrew macosx macports osx

371
plugins/modules/lxd_storage_pool_info.py Normal file
View file

@ -0,0 +1,371 @@
# Copyright (c) 2025, Sean McAvoy (@smcavoy) <seanmcavoy@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
DOCUMENTATION = """
---
module: lxd_storage_pool_info
short_description: Retrieve information about LXD storage pools
version_added: 12.1.0
description:
- Retrieve information about LXD storage pools.
- This module returns details about all storage pools or a specific storage pool.
author: "Sean McAvoy (@smcavoy)"
extends_documentation_fragment:
- community.general.attributes
- community.general.attributes.info_module
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
name:
description:
- Name of a specific storage pool to retrieve information about.
- If not specified, information about all storage pools are returned.
type: str
type:
description:
- Filter storage pools by driver/type (for example V(dir), V(zfs), V(btrfs), V(lvm), or V(ceph)).
- If not specified, all storage pool types are returned.
- Can be a single type or a list of types.
type: list
elements: str
default: []
project:
description:
- 'Project of the storage pool.
See U(https://documentation.ubuntu.com/lxd/en/latest/projects/).'
type: str
url:
description:
- The Unix domain socket path or the https URL for the LXD server.
default: unix:/var/lib/lxd/unix.socket
type: str
snap_url:
description:
- The Unix domain socket path when LXD is installed by snap package manager.
default: unix:/var/snap/lxd/common/lxd/unix.socket
type: str
client_key:
description:
- The client certificate key file path.
- If not specified, it defaults to C($HOME/.config/lxc/client.key).
aliases: [key_file]
type: path
client_cert:
description:
- The client certificate file path.
- If not specified, it defaults to C($HOME/.config/lxc/client.crt).
aliases: [cert_file]
type: path
trust_password:
description:
- The client trusted password.
- 'You need to set this password on the LXD server before
running this module using the following command:
C(lxc config set core.trust_password <some random password>)
See U(https://www.stgraber.org/2016/04/18/lxd-api-direct-interaction/).'
- If O(trust_password) is set, this module sends a request for
authentication before sending any requests.
type: str
"""
EXAMPLES = """
- name: Get information about all storage pools
community.general.lxd_storage_pool_info:
register: result
- name: Get information about a specific storage pool
community.general.lxd_storage_pool_info:
name: default
register: result
- name: Get information about all ZFS storage pools
community.general.lxd_storage_pool_info:
type:
- zfs
- ceph
register: result
- name: Get information about ZFS and BTRFS storage pools
community.general.lxd_storage_pool_info:
type:
- zfs
- btrfs
register: result
- name: Get storage pool information via HTTPS connection
community.general.lxd_storage_pool_info:
url: https://127.0.0.1:8443
trust_password: mypassword
name: default
register: result
- name: Get storage pool information for a specific project
community.general.lxd_storage_pool_info:
project: myproject
register: result
"""
RETURN = """
storage_pools:
description: List of LXD storage pools.
returned: success
type: list
elements: dict
sample: [
{
"name": "default",
"driver": "dir",
"used_by": ["/1.0/instances/container1"],
"config": {
"source": "/var/lib/lxd/storage-pools/default"
},
"description": "Default storage pool",
"status": "Created",
"locations": ["none"]
}
]
contains:
name:
description: The name of the storage pool.
type: str
returned: success
driver:
description: The storage pool driver/type.
type: str
returned: success
used_by:
description: List of resources using this storage pool.
type: list
elements: str
returned: success
config:
description: Configuration of the storage pool.
type: dict
returned: success
description:
description: Description of the storage pool.
type: str
returned: success
status:
description: Current status of the storage pool.
type: str
returned: success
locations:
description: Cluster member locations for the pool.
type: list
elements: str
returned: success
logs:
description: The logs of requests and responses.
returned: when ansible-playbook is invoked with -vvvv.
type: list
elements: dict
sample: [
{
"type": "sent request",
"request": {
"method": "GET",
"url": "/1.0/storage-pools",
"json": null,
"timeout": null
},
"response": {
"json": {"type": "sync", "status": "Success"}
}
}
]
"""
import os
from urllib.parse import urlencode
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.lxd import (
LXDClient,
LXDClientException,
default_cert_file,
default_key_file,
)
# ANSIBLE_LXD_DEFAULT_URL is a default value of the lxd endpoint
ANSIBLE_LXD_DEFAULT_URL = "unix:/var/lib/lxd/unix.socket"
ANSIBLE_LXD_DEFAULT_SNAP_URL = "unix:/var/snap/lxd/common/lxd/unix.socket"
# API endpoints
LXD_API_VERSION = "1.0"
LXD_API_STORAGE_POOLS_ENDPOINT = f"/{LXD_API_VERSION}/storage-pools"
class LXDStoragePoolInfo:
def __init__(self, module: AnsibleModule) -> None:
"""Gather information about LXD storage pools.
:param module: Processed Ansible Module.
:type module: AnsibleModule
"""
self.module = module
self.name = self.module.params["name"]
# pool_type is guaranteed to be a list (may be empty) due to default=[]
self.pool_type = self.module.params["type"]
self.project = self.module.params["project"]
self.key_file = self.module.params["client_key"]
if self.key_file is None:
self.key_file = default_key_file()
self.cert_file = self.module.params["client_cert"]
if self.cert_file is None:
self.cert_file = default_cert_file()
self.debug = self.module._verbosity >= 4
snap_socket_path = self.module.params["snap_url"]
if snap_socket_path.startswith("unix:"):
snap_socket_path = snap_socket_path[5:]
if self.module.params["url"] != ANSIBLE_LXD_DEFAULT_URL:
self.url = self.module.params["url"]
elif os.path.exists(snap_socket_path):
self.url = self.module.params["snap_url"]
else:
self.url = self.module.params["url"]
try:
self.client = LXDClient(
self.url,
key_file=self.key_file,
cert_file=self.cert_file,
debug=self.debug,
)
except LXDClientException as e:
self._fail_from_lxd_exception(e)
self.trust_password = self.module.params["trust_password"]
def _fail_from_lxd_exception(self, exception: LXDClientException) -> None:
"""Build failure parameters from LXDClientException and fail.
:param exception: The LXDClientException instance
:type exception: LXDClientException
"""
fail_params = {
"msg": exception.msg,
"changed": False,
}
if self.client.debug and "logs" in exception.kwargs:
fail_params["logs"] = exception.kwargs["logs"]
self.module.fail_json(**fail_params)
def _build_url(self, endpoint: str) -> str:
"""Build URL with project parameter if specified."""
if self.project:
return f"{endpoint}?{urlencode({'project': self.project})}"
return endpoint
def _get_storage_pool_list(self, recursion: int = 0) -> list:
"""Get list of all storage pools.
:param recursion: API recursion level (0 for URLs only, 1 for full objects)
:type recursion: int
:return: List of pool names (recursion=0) or pool objects (recursion=1)
:rtype: list
"""
endpoint = LXD_API_STORAGE_POOLS_ENDPOINT
if recursion > 0:
endpoint = f"{endpoint}?recursion={recursion}"
url = self._build_url(endpoint)
resp_json = self.client.do("GET", url, ok_error_codes=[])
if resp_json["type"] == "error":
self.module.fail_json(
msg=f"Failed to retrieve storage pools: {resp_json.get('error', 'Unknown error')}",
error_code=resp_json.get("error_code"),
)
metadata = resp_json.get("metadata", [])
if recursion == 0:
# With recursion=0: list of pool URLs like ['/1.0/storage-pools/default']
# Extract the pool names from URLs
return [url.split("/")[-1] for url in metadata]
else:
# With recursion=1: list of pool objects with full metadata
return metadata
def _get_storage_pool_info(self, pool_name: str) -> dict:
"""Get detailed information about a specific storage pool."""
url = self._build_url(f"{LXD_API_STORAGE_POOLS_ENDPOINT}/{pool_name}")
resp_json = self.client.do("GET", url, ok_error_codes=[404])
if resp_json["type"] == "error":
if resp_json.get("error_code") == 404:
self.module.fail_json(msg=f'Storage pool "{pool_name}" not found')
else:
self.module.fail_json(
msg=f'Failed to retrieve storage pool "{pool_name}": {resp_json.get("error", "Unknown error")}',
error_code=resp_json.get("error_code"),
)
return resp_json.get("metadata", {})
def get_storage_pools(self) -> list[dict]:
"""Retrieve storage pool information based on module parameters."""
storage_pools = []
if self.name:
# Get information about a specific storage pool
pool_info = self._get_storage_pool_info(self.name)
storage_pools.append(pool_info)
else:
# Get information about all storage pools using recursion
# This retrieves all pool details in a single API call instead of one call per pool
storage_pools = self._get_storage_pool_list(recursion=1)
# Filter by type if specified
if self.pool_type:
storage_pools = [pool for pool in storage_pools if pool.get("driver") in self.pool_type]
return storage_pools
def run(self) -> None:
"""Run the main method."""
try:
if self.trust_password is not None:
self.client.authenticate(self.trust_password)
storage_pools = self.get_storage_pools()
result_json = {"storage_pools": storage_pools}
if self.client.debug:
result_json["logs"] = self.client.logs
self.module.exit_json(**result_json)
except LXDClientException as e:
self._fail_from_lxd_exception(e)
def main() -> None:
"""Ansible Main module."""
module = AnsibleModule(
argument_spec=dict(
name=dict(type="str"),
type=dict(type="list", elements="str", default=[]),
project=dict(type="str"),
url=dict(type="str", default=ANSIBLE_LXD_DEFAULT_URL),
snap_url=dict(type="str", default=ANSIBLE_LXD_DEFAULT_SNAP_URL),
client_key=dict(type="path", aliases=["key_file"]),
client_cert=dict(type="path", aliases=["cert_file"]),
trust_password=dict(type="str", no_log=True),
),
supports_check_mode=True,
)
lxd_info = LXDStoragePoolInfo(module=module)
lxd_info.run()
if __name__ == "__main__":
main()

407
plugins/modules/lxd_storage_volume_info.py Normal file
View file

@ -0,0 +1,407 @@
# Copyright (c) 2025, Sean McAvoy (@smcavoy) <seanmcavoy@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
DOCUMENTATION = """
---
module: lxd_storage_volume_info
short_description: Retrieve information about LXD storage volumes
version_added: 12.1.0
description:
- Retrieve information about LXD storage volumes in a specific storage pool.
- This module returns details about all volumes or a specific volume in a pool.
author: "Sean McAvoy (@smcavoy)"
extends_documentation_fragment:
- community.general.attributes
- community.general.attributes.info_module
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
pool:
description:
- Name of the storage pool to query for volumes.
- This parameter is required.
required: true
type: str
name:
description:
- Name of a specific storage volume to retrieve information about.
- If not specified, information about all volumes in the pool are returned.
type: str
type:
description:
- Filter volumes by type.
- Common types include V(container), V(virtual-machine), V(image), and V(custom).
- If not specified, all volume types are returned.
type: str
project:
description:
- 'Project of the storage volume.
See U(https://documentation.ubuntu.com/lxd/en/latest/projects/).'
type: str
url:
description:
- The Unix domain socket path or the https URL for the LXD server.
default: unix:/var/lib/lxd/unix.socket
type: str
snap_url:
description:
- The Unix domain socket path when LXD is installed by snap package manager.
default: unix:/var/snap/lxd/common/lxd/unix.socket
type: str
client_key:
description:
- The client certificate key file path.
- If not specified, it defaults to C($HOME/.config/lxc/client.key).
aliases: [key_file]
type: path
client_cert:
description:
- The client certificate file path.
- If not specified, it defaults to C($HOME/.config/lxc/client.crt).
aliases: [cert_file]
type: path
trust_password:
description:
- The client trusted password.
- 'You need to set this password on the LXD server before
running this module using the following command:
C(lxc config set core.trust_password <some random password>)
See U(https://www.stgraber.org/2016/04/18/lxd-api-direct-interaction/).'
- If O(trust_password) is set, this module sends a request for
authentication before sending any requests.
type: str
"""
EXAMPLES = """
- name: Get information about all volumes in the default storage pool
community.general.lxd_storage_volume_info:
pool: default
register: result
- name: Get information about a specific volume
community.general.lxd_storage_volume_info:
pool: default
name: my-volume
register: result
- name: Get information about all custom volumes in a pool
community.general.lxd_storage_volume_info:
pool: default
type: custom
register: result
- name: Get volume information via HTTPS connection
community.general.lxd_storage_volume_info:
url: https://127.0.0.1:8443
trust_password: mypassword
pool: default
name: my-volume
register: result
- name: Get volume information for a specific project
community.general.lxd_storage_volume_info:
project: myproject
pool: default
register: result
- name: Get container volumes only
community.general.lxd_storage_volume_info:
pool: default
type: container
register: result
"""
RETURN = """
storage_volumes:
description: List of LXD storage volumes.
returned: success
type: list
elements: dict
sample: [
{
"name": "my-volume",
"type": "custom",
"used_by": [],
"config": {
"size": "10GiB"
},
"description": "My custom volume",
"content_type": "filesystem",
"location": "none"
}
]
contains:
name:
description: The name of the storage volume.
type: str
returned: success
type:
description: The type of the storage volume.
type: str
returned: success
used_by:
description: List of resources using this storage volume.
type: list
elements: str
returned: success
config:
description: Configuration of the storage volume.
type: dict
returned: success
description:
description: Description of the storage volume.
type: str
returned: success
content_type:
description: Content type of the volume (filesystem or block).
type: str
returned: success
location:
description: Cluster member location for the volume.
type: str
returned: success
logs:
description: The logs of requests and responses.
returned: when ansible-playbook is invoked with -vvvv.
type: list
elements: dict
sample: [
{
"type": "sent request",
"request": {
"method": "GET",
"url": "/1.0/storage-pools/default/volumes",
"json": null,
"timeout": null
},
"response": {
"json": {"type": "sync", "status": "Success"}
}
}
]
"""
import os
from urllib.parse import quote, urlencode
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.community.general.plugins.module_utils.lxd import (
LXDClient,
LXDClientException,
default_cert_file,
default_key_file,
)
# ANSIBLE_LXD_DEFAULT_URL is a default value of the lxd endpoint
ANSIBLE_LXD_DEFAULT_URL = "unix:/var/lib/lxd/unix.socket"
ANSIBLE_LXD_DEFAULT_SNAP_URL = "unix:/var/snap/lxd/common/lxd/unix.socket"
# API endpoints
LXD_API_VERSION = "1.0"
LXD_API_STORAGE_POOLS_ENDPOINT = f"/{LXD_API_VERSION}/storage-pools"
class LXDStorageVolumeInfo:
def __init__(self, module: AnsibleModule) -> None:
"""Gather information about LXD storage volumes.
:param module: Processed Ansible Module.
:type module: AnsibleModule
"""
self.module = module
self.pool = self.module.params["pool"]
self.name = self.module.params["name"]
self.volume_type = self.module.params["type"]
self.project = self.module.params["project"]
self.key_file = self.module.params["client_key"]
if self.key_file is None:
self.key_file = default_key_file()
self.cert_file = self.module.params["client_cert"]
if self.cert_file is None:
self.cert_file = default_cert_file()
self.debug = self.module._verbosity >= 4
# check if domain socket to be used
snap_socket_path = self.module.params["snap_url"]
if snap_socket_path.startswith("unix:"):
snap_socket_path = snap_socket_path[5:]
if self.module.params["url"] != ANSIBLE_LXD_DEFAULT_URL:
self.url = self.module.params["url"]
elif os.path.exists(snap_socket_path):
self.url = self.module.params["snap_url"]
else:
self.url = self.module.params["url"]
try:
self.client = LXDClient(
self.url,
key_file=self.key_file,
cert_file=self.cert_file,
debug=self.debug,
)
except LXDClientException as e:
self._fail_from_lxd_exception(e)
self.trust_password = self.module.params["trust_password"]
def _fail_from_lxd_exception(self, exception: LXDClientException) -> None:
"""Build failure parameters from LXDClientException and fail.
:param exception: The LXDClientException instance
:type exception: LXDClientException
"""
fail_params = {
"msg": exception.msg,
"changed": False,
}
if self.client.debug and "logs" in exception.kwargs:
fail_params["logs"] = exception.kwargs["logs"]
self.module.fail_json(**fail_params)
def _build_url(self, endpoint: str) -> str:
"""Build URL with project parameter if specified."""
if self.project:
return f"{endpoint}?{urlencode({'project': self.project})}"
return endpoint
def _check_pool_exists(self) -> None:
"""Verify that the storage pool exists."""
url = self._build_url(f"{LXD_API_STORAGE_POOLS_ENDPOINT}/{quote(self.pool, safe='')}")
resp_json = self.client.do("GET", url, ok_error_codes=[404])
if resp_json["type"] == "error":
if resp_json.get("error_code") == 404:
self.module.fail_json(msg=f'Storage pool "{self.pool}" not found')
else:
self.module.fail_json(
msg=f'Failed to retrieve storage pool "{self.pool}": {resp_json.get("error", "Unknown error")}'
)
def _get_volume_list(self, recursion: int = 0) -> list:
"""Get list of all volumes in the storage pool.
:param recursion: API recursion level (0 for URLs only, 1 for full objects)
:type recursion: int
:return: List of volume URLs (recursion=0) or volume objects (recursion=1)
:rtype: list
"""
endpoint = f"{LXD_API_STORAGE_POOLS_ENDPOINT}/{quote(self.pool, safe='')}/volumes"
if recursion > 0:
endpoint = f"{endpoint}?recursion={recursion}"
url = self._build_url(endpoint)
resp_json = self.client.do("GET", url, ok_error_codes=[])
if resp_json["type"] == "error":
self.module.fail_json(
msg=f'Failed to retrieve volumes from pool "{self.pool}": {resp_json.get("error", "Unknown error")}',
error_code=resp_json.get("error_code"),
)
# With recursion=0: list of volume URLs like ['/1.0/storage-pools/default/volumes/custom/my-volume']
# With recursion=1: list of volume objects with full metadata
return resp_json.get("metadata", [])
def _get_volume_info(self, volume_type: str, volume_name: str) -> dict:
"""Get detailed information about a specific storage volume."""
url = self._build_url(
f"{LXD_API_STORAGE_POOLS_ENDPOINT}/{quote(self.pool, safe='')}/volumes/{quote(volume_type, safe='')}/{quote(volume_name, safe='')}"
)
resp_json = self.client.do("GET", url, ok_error_codes=[404])
if resp_json["type"] == "error":
if resp_json.get("error_code") == 404:
self.module.fail_json(
msg=f'Storage volume "{volume_name}" of type "{volume_type}" not found in pool "{self.pool}"'
)
else:
self.module.fail_json(
msg=f'Failed to retrieve volume "{volume_name}" of type "{volume_type}": {resp_json.get("error", "Unknown error")}'
)
return resp_json.get("metadata", {})
def get_storage_volumes(self) -> list[dict]:
"""Retrieve storage volume information based on module parameters."""
# First check if the pool exists
self._check_pool_exists()
storage_volumes = []
if self.name:
# Get information about a specific volume
# We need to determine the type if not specified
if self.volume_type:
volume_info = self._get_volume_info(self.volume_type, self.name)
storage_volumes.append(volume_info)
else:
# Try to find the volume by name using recursion for efficiency
# This gets all volume objects in a single API call
volumes = self._get_volume_list(recursion=1)
found = False
for volume in volumes:
if volume.get("name") == self.name:
storage_volumes.append(volume)
found = True
break
if not found:
self.module.fail_json(msg=f'Storage volume "{self.name}" not found in pool "{self.pool}"')
else:
# Get information about all volumes in the pool using recursion
# This retrieves all volume details in a single API call instead of one call per volume
volumes = self._get_volume_list(recursion=1)
for volume in volumes:
# Apply type filter if specified
if self.volume_type and volume.get("type") != self.volume_type:
continue
storage_volumes.append(volume)
return storage_volumes
def run(self) -> None:
"""Run the main method."""
try:
if self.trust_password is not None:
self.client.authenticate(self.trust_password)
storage_volumes = self.get_storage_volumes()
result_json = {"storage_volumes": storage_volumes}
if self.client.debug:
result_json["logs"] = self.client.logs
self.module.exit_json(**result_json)
except LXDClientException as e:
self._fail_from_lxd_exception(e)
def main() -> None:
"""Ansible Main module."""
module = AnsibleModule(
argument_spec=dict(
pool=dict(type="str", required=True),
name=dict(type="str"),
type=dict(type="str"),
project=dict(type="str"),
url=dict(type="str", default=ANSIBLE_LXD_DEFAULT_URL),
snap_url=dict(type="str", default=ANSIBLE_LXD_DEFAULT_SNAP_URL),
client_key=dict(type="path", aliases=["key_file"]),
client_cert=dict(type="path", aliases=["cert_file"]),
trust_password=dict(type="str", no_log=True),
),
supports_check_mode=True,
)
lxd_info = LXDStorageVolumeInfo(module=module)
lxd_info.run()
if __name__ == "__main__":
main()

5
tests/integration/targets/lxd_storage_pool_info/aliases Normal file
View file

@ -0,0 +1,5 @@
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
unsupported

69
tests/integration/targets/lxd_storage_pool_info/tasks/main.yaml Normal file
View file

@ -0,0 +1,69 @@
####################################################################
# WARNING: These are designed specifically for Ansible tests #
# and should not be used as examples of how to write Ansible roles #
####################################################################
# Copyright (c) Ansible project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: Get information about all storage pools
community.general.lxd_storage_pool_info:
register: pool_info_all
- name: Verify pools information was retrieved
assert:
that:
- pool_info_all is not changed
- pool_info_all is success
- pool_info_all.storage_pools is defined
- pool_info_all.storage_pools is iterable
- name: Get information about default pool
community.general.lxd_storage_pool_info:
name: default
register: pool_info_default
- name: Verify default pool information
assert:
that:
- pool_info_default is not changed
- pool_info_default is success
- pool_info_default.storage_pools | length == 1
- pool_info_default.storage_pools[0].name == 'default'
- name: Test filtering by pool type (dir)
community.general.lxd_storage_pool_info:
type:
- dir
register: pool_info_dir
- name: Verify type filtering works
assert:
that:
- pool_info_dir is not changed
- pool_info_dir is success
- pool_info_dir.storage_pools is defined
- name: Test with non-existent pool (should fail)
community.general.lxd_storage_pool_info:
name: nonexistent-pool-12345
register: pool_info_nonexistent
ignore_errors: true
- name: Verify error handling for non-existent pool
assert:
that:
- pool_info_nonexistent is failed
- name: Test check mode
community.general.lxd_storage_pool_info:
check_mode: true
register: pool_info_check
- name: Verify check mode works
assert:
that:
- pool_info_check is not changed
- pool_info_check is success
- pool_info_check.storage_pools is defined

5
tests/integration/targets/lxd_storage_volume_info/aliases Normal file
View file

@ -0,0 +1,5 @@
# Copyright (c) Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
unsupported

101
tests/integration/targets/lxd_storage_volume_info/tasks/main.yaml Normal file
View file

@ -0,0 +1,101 @@
####################################################################
# WARNING: These are designed specifically for Ansible tests #
# and should not be used as examples of how to write Ansible roles #
####################################################################
# Copyright (c) Ansible project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
- name: Get information about all volumes in default pool
community.general.lxd_storage_volume_info:
pool: default
register: volume_info_all
- name: Verify volumes information was retrieved
assert:
that:
- volume_info_all is not changed
- volume_info_all is success
- volume_info_all.storage_volumes is defined
- volume_info_all.storage_volumes is iterable
- name: Test filtering by volume type (custom)
community.general.lxd_storage_volume_info:
pool: default
type: custom
register: volume_info_custom
- name: Verify type filtering works
assert:
that:
- volume_info_custom is not changed
- volume_info_custom is success
- volume_info_custom.storage_volumes is defined
- volume_info_custom.storage_volumes is iterable
- name: Test filtering by volume type (container)
community.general.lxd_storage_volume_info:
pool: default
type: container
register: volume_info_container
- name: Verify container type filtering works
assert:
that:
- volume_info_container is not changed
- volume_info_container is success
- volume_info_container.storage_volumes is defined
- name: Test fetching specific volume by name
community.general.lxd_storage_volume_info:
pool: default
name: "{{ volume_info_all.storage_volumes[0].name }}"
register: volume_info_specific
when: volume_info_all.storage_volumes | length > 0
- name: Verify specific volume retrieval
assert:
that:
- volume_info_specific is not changed
- volume_info_specific is success
- volume_info_specific.storage_volumes | length == 1
- volume_info_specific.storage_volumes[0].name == volume_info_all.storage_volumes[0].name
when: volume_info_all.storage_volumes | length > 0
- name: Test with non-existent pool (should fail)
community.general.lxd_storage_volume_info:
pool: nonexistent-pool-12345
register: volume_info_nonexistent_pool
ignore_errors: true
- name: Verify error handling for non-existent pool
assert:
that:
- volume_info_nonexistent_pool is failed
- name: Test with non-existent volume name (should fail)
community.general.lxd_storage_volume_info:
pool: default
name: nonexistent-volume-12345
register: volume_info_nonexistent_volume
ignore_errors: true
- name: Verify error handling for non-existent volume
assert:
that:
- volume_info_nonexistent_volume is failed
when: volume_info_nonexistent_volume is defined
- name: Test check mode
community.general.lxd_storage_volume_info:
pool: default
check_mode: true
register: volume_info_check
- name: Verify check mode works
assert:
that:
- volume_info_check is not changed
- volume_info_check is success
- volume_info_check.storage_volumes is defined

125
tests/unit/plugins/modules/test_lxd_storage_pool_info.py Normal file
View file

@ -0,0 +1,125 @@
# Copyright (c) 2025, Sean McAvoy (@smcavoy) <seanmcavoy@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
from unittest.mock import patch
from ansible_collections.community.general.plugins.modules import lxd_storage_pool_info as module
from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import (
AnsibleExitJson,
AnsibleFailJson,
ModuleTestCase,
set_module_args,
)
class FakeLXDClient:
responses: dict[tuple[str, str], dict] = {}
def __init__(self, url, key_file=None, cert_file=None, debug=False, **kwargs):
self.url = url
self.key_file = key_file
self.cert_file = cert_file
self.debug = debug
self.logs = [{"type": "fake-request"}] if debug else []
def authenticate(self, trust_password):
self.trust_password = trust_password
def do(self, method, url, ok_error_codes=None, **kwargs):
try:
return self.responses[(method, url)]
except KeyError as exc:
raise AssertionError(f"Unexpected call: {method} {url}") from exc
class TestLXDStoragePoolInfo(ModuleTestCase):
def setUp(self):
super().setUp()
self.module = module
def test_returns_storage_pools(self):
"""Pool metadata from the API is returned unchanged using recursion."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools?recursion=1"): {
"type": "sync",
"metadata": [
{"name": "default", "driver": "dir"},
{"name": "fast", "driver": "zfs"},
],
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_pools"] == [
{"name": "default", "driver": "dir"},
{"name": "fast", "driver": "zfs"},
]
def test_returns_specific_storage_pool(self):
"""When name is specified, only that pool is returned."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default", "driver": "dir"}},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"name": "default"}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_pools"] == [
{"name": "default", "driver": "dir"},
]
def test_filters_storage_pools_by_type(self):
"""Pools can be filtered by driver type using recursion."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools?recursion=1"): {
"type": "sync",
"metadata": [
{"name": "default", "driver": "dir"},
{"name": "fast", "driver": "zfs"},
],
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"type": ["zfs"]}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_pools"] == [
{"name": "fast", "driver": "zfs"},
]
def test_error_code_returned_on_failure(self):
"""Failures surface the LXD error code for easier debugging."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools?recursion=1"): {
"type": "error",
"error": "unavailable",
"error_code": 503,
}
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleFailJson) as exc:
with set_module_args({}):
self.module.main()
result = exc.exception.args[0]
assert result["error_code"] == 503
assert "Failed to retrieve storage pools" in result["msg"]

156
tests/unit/plugins/modules/test_lxd_storage_volume_info.py Normal file
View file

@ -0,0 +1,156 @@
# Copyright (c) 2025, Sean McAvoy (@smcavoy) <seanmcavoy@gmail.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
from unittest.mock import patch
from ansible_collections.community.general.plugins.modules import lxd_storage_volume_info as module
from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import (
AnsibleExitJson,
AnsibleFailJson,
ModuleTestCase,
set_module_args,
)
class FakeLXDClient:
responses: dict[tuple[str, str], dict] = {}
def __init__(self, url, key_file=None, cert_file=None, debug=False, **kwargs):
self.url = url
self.key_file = key_file
self.cert_file = cert_file
self.debug = debug
self.logs = [{"type": "fake-request"}] if debug else []
def authenticate(self, trust_password):
self.trust_password = trust_password
def do(self, method, url, ok_error_codes=None, **kwargs):
try:
return self.responses[(method, url)]
except KeyError as exc:
raise AssertionError(f"Unexpected call: {method} {url}") from exc
class TestLXDStorageVolumeInfo(ModuleTestCase):
def setUp(self):
super().setUp()
self.module = module
def test_returns_all_volumes_for_pool(self):
"""Volume metadata is returned for every volume in the pool using recursion."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default"}},
("GET", "/1.0/storage-pools/default/volumes?recursion=1"): {
"type": "sync",
"metadata": [
{"name": "data", "type": "custom"},
{"name": "web", "type": "container"},
],
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"pool": "default"}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_volumes"] == [
{"name": "data", "type": "custom"},
{"name": "web", "type": "container"},
]
def test_returns_specific_storage_volume(self):
"""When name is specified without type, recursion is used to find it."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default"}},
("GET", "/1.0/storage-pools/default/volumes?recursion=1"): {
"type": "sync",
"metadata": [
{"name": "data", "type": "custom"},
{"name": "web", "type": "container"},
],
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"pool": "default", "name": "data"}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_volumes"] == [
{"name": "data", "type": "custom"},
]
def test_filters_storage_volumes_by_type(self):
"""Volumes can be filtered by type using recursion."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default"}},
("GET", "/1.0/storage-pools/default/volumes?recursion=1"): {
"type": "sync",
"metadata": [
{"name": "data", "type": "custom"},
{"name": "web", "type": "container"},
],
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"pool": "default", "type": "container"}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_volumes"] == [
{"name": "web", "type": "container"},
]
def test_returns_specific_volume_with_type_using_direct_request(self):
"""When both name and type are specified, a direct API call is made (no recursion)."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default"}},
("GET", "/1.0/storage-pools/default/volumes/custom/data"): {
"type": "sync",
"metadata": {"name": "data", "type": "custom", "config": {"size": "10GiB"}},
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleExitJson) as exc:
with set_module_args({"pool": "default", "name": "data", "type": "custom"}):
self.module.main()
result = exc.exception.args[0]
assert result["storage_volumes"] == [
{"name": "data", "type": "custom", "config": {"size": "10GiB"}},
]
def test_error_code_returned_when_listing_volumes_fails(self):
"""Errors from LXD are surfaced with the numeric code."""
FakeLXDClient.responses = {
("GET", "/1.0/storage-pools/default"): {"type": "sync", "metadata": {"name": "default"}},
("GET", "/1.0/storage-pools/default/volumes?recursion=1"): {
"type": "error",
"error": "service unavailable",
"error_code": 503,
},
}
with patch.object(self.module, "LXDClient", FakeLXDClient):
with patch.object(self.module.os.path, "exists", return_value=False):
with self.assertRaises(AnsibleFailJson) as exc:
with set_module_args({"pool": "default"}):
self.module.main()
result = exc.exception.args[0]
assert result["error_code"] == 503
assert "Failed to retrieve volumes from pool" in result["msg"]