ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-07-05 18:18:53 +00:00
Code Activity

[PR #11752/d4031f36 backport][stable-13] kopia: Add kopia_repository module (#12127)

Browse source 
kopia: Add kopia_repository module (#11752)

* Add kopia module util

* fix pipeline suggestions

* add kopia repository module

* apply code review changes

* remove kopia_runner instance unit test

* update botmeta with kopia

* refactor docs and redundant state

* add kopia_info module and fix kopia_repository check mode support

- Add kopia_info module for read-only repository information gathering
  (kopia repository status, kopia repository throttle get) following
  the pacemaker_info pattern with ModuleHelper and info_module fragment
- Add _fmt_throttle to _kopia.py and register throttle format in
  kopia_runner; remove throttle_operation get option from
  kopia_repository per Ansible best practices (info ops belong in
  _info modules)
- Add throttle suboption dict to kopia_repository with all seven
  kopia repository throttle set flags
- Fix check_mode: support from full to actually full by implementing
  _predict_value() in kopia_repository; previously check_mode_skip
  caused changed to always be false in check mode
- Add check mode test cases to test_kopia_repository.yaml covering
  created and disconnected states for both connected and disconnected
  initial conditions
- Add BOTMETA.yml entry and full test fixture for kopia_info

* apply code review suggestions

(cherry picked from commit d4031f36e4)

Co-authored-by: munchtoast <45038532+munchtoast@users.noreply.github.com>
This commit is contained in:
patchback[bot] 2026-05-30 15:12:10 +02:00 • committed by GitHub
parent 6d7e44f14b
commit f7647b2131
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 1378 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.github/BOTMETA.yml vendored
View file

@ -389,6 +389,8 @@ files:
maintainers: $team_keycloak
$module_utils/_keycloak.py:
maintainers: $team_keycloak
$module_utils/_kopia.py:
maintainers: munchtoast
$module_utils/_lxc.py:
maintainers: russoz
$module_utils/_lxca_common.py:
@ -901,6 +903,10 @@ files:
maintainers: ahussey-redhat
$modules/kibana_plugin.py:
maintainers: barryib
$modules/kopia_repository_info.py:
maintainers: munchtoast
$modules/kopia_repository.py:
maintainers: munchtoast
$modules/krb_ticket.py:
maintainers: abakanovskii
$modules/launchd.py:

24
plugins/doc_fragments/_kopia.py Normal file
View file

@ -0,0 +1,24 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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
# Note that this doc fragment is **PRIVATE** to the collection. It can have breaking changes at any time.
# Do not use this from other collections or standalone plugins/modules!
from __future__ import annotations
class ModuleDocFragment:
# Common parameters for Kopia modules
DOCUMENTATION = r"""
options:
password:
description:
- Repository password used to encrypt and decrypt repository contents.
type: str
config:
description:
- Path to the Kopia config file for this repository connection.
- Defaults to the Kopia default config path when not set.
type: path
"""

162
plugins/module_utils/_kopia.py Normal file
View file

@ -0,0 +1,162 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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
# Note that this module util is **PRIVATE** to the collection. It can have breaking changes at any time.
# Do not use this from other collections or standalone plugins/modules!
from __future__ import annotations
import typing as t
from ansible_collections.community.general.plugins.module_utils._cmd_runner import CmdRunner, cmd_runner_fmt
if t.TYPE_CHECKING:
from ansible.module_utils.basic import AnsibleModule
# Maps kopia_repository module state values to kopia CLI subcommands.
# Used with cmd_runner_fmt.as_map() for the 'state' arg format.
REPOSITORY_STATE_MAP = {
"created": "create",
"connected": "connect",
"disconnected": "disconnect",
"synced": "sync-to",
"throttled": "throttle",
}
# Maps backend provider names to their CLI flag names.
# Each provider emits its name as a positional sub-subcommand, followed by
# --flag value pairs for each non-None param.
# The "server" provider is intentionally absent: `kopia repository connect server`
# uses top-level flags (--url, --server-cert-fingerprint) rather than backend flags,
# so fmt_backend() returns [] for it and those flags are passed separately.
_PROVIDER_BACKEND_MAP = {
"azure": {
"container": "--container",
"storage_account": "--storage-account",
"storage_key": "--storage-key",
"sas_token": "--sas-token",
"storage_domain": "--storage-domain",
"prefix": "--prefix",
},
"b2": {
"bucket": "--bucket",
"access_key": "--key-id",
"secret_access_key": "--key",
"prefix": "--prefix",
},
"filesystem": {
"path": "--path",
},
"gcs": {
"bucket": "--bucket",
"credentials_file": "--credentials-file",
"prefix": "--prefix",
},
"gdrive": {
"folder_id": "--folder-id",
"credentials_file": "--credentials-file",
},
"rclone": {
"path": "--remote-path",
},
"s3": {
"bucket": "--bucket",
"access_key": "--access-key",
"secret_access_key": "--secret-access-key",
"endpoint": "--endpoint",
"region": "--region",
"prefix": "--prefix",
"session_token": "--session-token",
},
"sftp": {
"path": "--path",
"host": "--host",
"username": "--username",
"port": "--port",
"keyfile": "--keyfile",
"known_hosts": "--known-hosts",
},
"webdav": {
"url": "--url",
"webdav_username": "--webdav-username",
"webdav_password": "--webdav-password",
},
}
# Argument spec entries shared by all kopia modules.
# Include this in each module's argument_spec via dict unpacking.
KOPIA_COMMON_ARGUMENT_SPEC = dict(
password=dict(type="str", no_log=True),
config=dict(type="path"),
)
def fmt_backend(value):
"""Format the backend dict into positional + flag arguments for kopia CLI.
For most providers the output is:
[provider_name, --flag1, value1, --flag2, value2, ...]
For the "server" provider, returns [] because server connect uses top-level
flags (--url, --server-cert-fingerprint) passed separately.
"""
provider = value["provider"]
if provider == "server":
return []
result = [provider]
for param_name, flag in _PROVIDER_BACKEND_MAP[provider].items():
param_value = value.get(param_name)
if param_value is not None:
result.append(f"{flag}={param_value}")
return result
def _fmt_throttle(value):
"""Format the throttle dict into --flag value arguments for kopia repository throttle set."""
if not value:
return []
flag_map = {
"download_bytes_per_second": "--download-bytes-per-second",
"upload_bytes_per_second": "--upload-bytes-per-second",
"read_requests_per_second": "--read-requests-per-second",
"write_requests_per_second": "--write-requests-per-second",
"list_requests_per_second": "--list-requests-per-second",
"concurrent_reads": "--concurrent-reads",
"concurrent_writes": "--concurrent-writes",
}
result = []
for param, flag in flag_map.items():
v = value.get(param)
if v is not None:
result.extend([flag, str(v)])
return result
def kopia_runner(module: AnsibleModule, extra_formats: dict | None = None, **kwargs) -> CmdRunner:
"""Create a CmdRunner for the kopia CLI.
Provides arg formats for all params shared across kopia modules.
Pass extra_formats to add module-specific arg formats on top of the shared ones.
"""
formats = dict(
cli_action=cmd_runner_fmt.as_list(),
status=cmd_runner_fmt.as_fixed("repository", "status"),
get_throttle=cmd_runner_fmt.as_fixed("repository", "throttle", "get"),
state=cmd_runner_fmt.as_map(REPOSITORY_STATE_MAP),
backend=cmd_runner_fmt.as_func(fmt_backend),
password=cmd_runner_fmt.as_opt_eq_val("--password"),
fingerprint_tls=cmd_runner_fmt.as_opt_eq_val("--server-cert-fingerprint"),
url=cmd_runner_fmt.as_opt_eq_val("--url"),
config=cmd_runner_fmt.as_opt_eq_val("--config-file"),
throttle_operation=cmd_runner_fmt.as_list(),
throttle=cmd_runner_fmt.as_func(_fmt_throttle),
)
if extra_formats:
formats.update(extra_formats)
return CmdRunner(
module,
command=["kopia"],
arg_formats=formats,
**kwargs,
)

460
plugins/modules/kopia_repository.py Normal file
View file

@ -0,0 +1,460 @@
#!/usr/bin/python
# Copyright (c) 2026, Dexter Le <dextersydney2001@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 = r"""
module: kopia_repository
short_description: Manage Kopia repository
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Manage a Kopia repository using the Kopia CLI.
- Supports creating, connecting, disconnecting, syncing, and throttling repositories.
extends_documentation_fragment:
- community.general._attributes
- community.general._kopia
attributes:
check_mode:
support: full
diff_mode:
support: full
options:
state:
description:
- Desired state of the Kopia repository.
type: str
choices:
created: Creates a new repository at the given backend.
connected: Connects to an existing repository or Kopia server.
disconnected: Disconnects from the current repository.
synced: Synchronizes the current repository to another backend location.
throttled: Sets throttle limits on the current repository.
default: created
fingerprint_tls:
description:
- TLS certificate fingerprint of the Kopia server.
- Required if O(state=connected) and O(backend.provider=server).
type: str
url:
description:
- URL of the Kopia server to connect to.
- Required if O(state=connected) and O(backend.provider=server).
type: str
throttle:
description:
- Throttle limits for the repository connection.
- Only used when O(state=throttled).
type: dict
suboptions:
download_bytes_per_second:
description:
- Maximum download speed in bytes per second. Set to V(0) to disable the limit.
type: int
upload_bytes_per_second:
description:
- Maximum upload speed in bytes per second. Set to V(0) to disable the limit.
type: int
read_requests_per_second:
description:
- Maximum number of read requests per second.
type: float
write_requests_per_second:
description:
- Maximum number of write requests per second.
type: float
list_requests_per_second:
description:
- Maximum number of list requests per second.
type: float
concurrent_reads:
description:
- Maximum number of concurrent read operations.
type: int
concurrent_writes:
description:
- Maximum number of concurrent write operations.
type: int
backend:
description:
- Backend storage configuration for the repository.
- Required if O(state=created), O(state=connected), or O(state=synced).
type: dict
suboptions:
provider:
description:
- Backend storage provider.
- Use V(server) to connect to a Kopia repository server instead of directly to storage.
type: str
required: true
choices: [azure, b2, filesystem, gcs, gdrive, rclone, s3, sftp, webdav, server]
bucket:
description:
- Bucket name for the backend.
- Required if O(backend.provider=b2), O(backend.provider=gcs), or O(backend.provider=s3).
type: str
container:
description:
- Azure Blob Storage container name.
- Required if O(backend.provider=azure).
type: str
storage_account:
description:
- Azure storage account name.
- Required if O(backend.provider=azure).
type: str
storage_key:
description:
- Azure storage account key used to authenticate.
- Optional if O(backend.provider=azure); omit when using managed identity or SAS tokens.
type: str
sas_token:
description:
- Azure Shared Access Signature token for authentication.
- Optional alternative to O(backend.storage_key) when O(backend.provider=azure).
type: str
storage_domain:
description:
- Azure storage domain override.
- Optional if O(backend.provider=azure).
type: str
access_key:
description:
- Access key ID for the backend.
- Required if O(backend.provider=b2) or O(backend.provider=s3).
type: str
secret_access_key:
description:
- Secret access key for the backend.
- Required if O(backend.provider=b2) or O(backend.provider=s3).
type: str
session_token:
description:
- Session token for temporary AWS credentials.
- Optional if O(backend.provider=s3).
type: str
endpoint:
description:
- S3-compatible endpoint URL.
- Optional if O(backend.provider=s3); defaults to C(s3.amazonaws.com).
type: str
region:
description:
- S3 bucket region.
- Optional if O(backend.provider=s3).
type: str
folder_id:
description:
- Google Drive folder ID to use as the backend root.
- Required if O(backend.provider=gdrive).
type: str
credentials_file:
description:
- Path to a JSON credentials file for authentication.
- Optional if O(backend.provider=gcs) or O(backend.provider=gdrive).
type: path
path:
description:
- Local file system path or remote path for the backend.
- Required if O(backend.provider=filesystem), O(backend.provider=rclone), or O(backend.provider=sftp).
type: path
host:
description:
- SFTP server hostname.
- Required if O(backend.provider=sftp).
type: str
username:
description:
- SFTP username for authentication.
- Required if O(backend.provider=sftp).
type: str
port:
description:
- SFTP server port.
- Optional if O(backend.provider=sftp); defaults to V(22).
type: int
keyfile:
description:
- Path to the SSH private key file for SFTP authentication.
- Optional if O(backend.provider=sftp).
type: path
known_hosts:
description:
- Path to a known_hosts file for SFTP host key verification.
- Optional if O(backend.provider=sftp).
type: path
url:
description:
- WebDAV server URL.
- Required if O(backend.provider=webdav).
type: str
webdav_username:
description:
- Username for WebDAV authentication.
- Optional if O(backend.provider=webdav).
type: str
webdav_password:
description:
- Password for WebDAV authentication.
- Optional if O(backend.provider=webdav).
type: str
prefix:
description:
- Object key prefix within the backend storage.
- Optional if O(backend.provider=azure), O(backend.provider=b2),
O(backend.provider=gcs), or O(backend.provider=s3).
type: str
"""
EXAMPLES = r"""
- name: Create a Kopia repository with S3 backend
community.general.kopia_repository:
state: created
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-bucket
access_key: myaccesskey
secret_access_key: mysecretaccesskey
- name: Create a Kopia repository on the local filesystem
community.general.kopia_repository:
state: created
password: secret
backend:
provider: filesystem
path: /mnt/backup/kopia
- name: Connect to a Kopia repository server
community.general.kopia_repository:
state: connected
password: secret
config: /etc/kopia/root.config
url: https://kopia.example.com:51515
fingerprint_tls: AA:BB:CC:DD:EE:FF
backend:
provider: server
- name: Connect directly to an Azure backend
community.general.kopia_repository:
state: connected
password: secret
backend:
provider: azure
container: my-container
storage_account: mystorageaccount
storage_key: mystoragekey
- name: Disconnect the Kopia repository
community.general.kopia_repository:
state: disconnected
config: /etc/kopia/root.config
- name: Sync Kopia repository to an S3 location
community.general.kopia_repository:
state: synced
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-synced-bucket
access_key: myaccesskey
secret_access_key: mysecretaccesskey
"""
RETURN = r"""
kopia_repository:
description: Output from the Kopia repository command.
type: str
sample: |-
Connected to repository: s3:/my-bucket/
Config file: /etc/kopia/root.config
...
returned: always
"""
from ansible_collections.community.general.plugins.module_utils._kopia import (
KOPIA_COMMON_ARGUMENT_SPEC,
kopia_runner,
)
from ansible_collections.community.general.plugins.module_utils._module_helper import StateModuleHelper
class KopiaRepository(StateModuleHelper):
module = dict(
supports_check_mode=True,
argument_spec=dict(
**KOPIA_COMMON_ARGUMENT_SPEC,
state=dict(
type="str",
default="created",
choices=["created", "connected", "disconnected", "synced", "throttled"],
),
fingerprint_tls=dict(type="str"),
url=dict(type="str"),
throttle=dict(
type="dict",
options=dict(
download_bytes_per_second=dict(type="int"),
upload_bytes_per_second=dict(type="int"),
read_requests_per_second=dict(type="float"),
write_requests_per_second=dict(type="float"),
list_requests_per_second=dict(type="float"),
concurrent_reads=dict(type="int"),
concurrent_writes=dict(type="int"),
),
),
backend=dict(
type="dict",
options=dict(
provider=dict(
type="str",
required=True,
choices=[
"azure",
"b2",
"filesystem",
"gcs",
"gdrive",
"rclone",
"s3",
"sftp",
"webdav",
"server",
],
),
bucket=dict(type="str"),
container=dict(type="str"),
storage_account=dict(type="str"),
storage_key=dict(type="str", no_log=True),
sas_token=dict(type="str", no_log=True),
storage_domain=dict(type="str"),
access_key=dict(type="str", no_log=True),
secret_access_key=dict(type="str", no_log=True),
session_token=dict(type="str", no_log=True),
endpoint=dict(type="str"),
region=dict(type="str"),
folder_id=dict(type="str"),
credentials_file=dict(type="path"),
path=dict(type="path"),
host=dict(type="str"),
username=dict(type="str"),
port=dict(type="int"),
keyfile=dict(type="path"),
known_hosts=dict(type="path"),
url=dict(type="str"),
webdav_username=dict(type="str"),
webdav_password=dict(type="str", no_log=True),
prefix=dict(type="str"),
),
required_if=[
("provider", "azure", ["container", "storage_account"]),
("provider", "b2", ["bucket", "access_key", "secret_access_key"]),
("provider", "filesystem", ["path"]),
("provider", "gcs", ["bucket"]),
("provider", "gdrive", ["folder_id"]),
("provider", "rclone", ["path"]),
("provider", "s3", ["bucket", "access_key", "secret_access_key"]),
("provider", "sftp", ["path", "host", "username"]),
("provider", "webdav", ["url"]),
],
),
),
required_if=[
("state", "created", ["backend"]),
("state", "connected", ["backend"]),
("state", "synced", ["backend"]),
],
)
def __init_module__(self):
self.runner = kopia_runner(self.module)
self.vars.set("previous_value", self._get()["out"])
self.vars.set("value", self.vars.previous_value, change=True, diff=True)
def __quit_module__(self):
if self.module.check_mode:
self.vars.set("value", self._predict_value())
else:
self.vars.set("value", self._get()["out"])
def _predict_value(self):
"""Predict the post-operation repository status for check mode change detection."""
state = self.module.params["state"]
previous = self.vars.previous_value
if state in ("created", "connected"):
return previous if previous is not None else "Connected to repository."
if state == "disconnected":
return None if previous is not None else previous
return previous
def _get(self):
with self.runner("status config") as ctx:
result = ctx.run()
return dict(
rc=result[0],
out=(result[1].rstrip() if result[1] else None),
err=result[2],
)
def _process_command_output(self, fail_on_err, ignore_err_msg=""):
def process(rc, out, err):
if fail_on_err and rc != 0 and err and ignore_err_msg not in err:
self.do_raise(f"kopia failed with error (rc={rc}): {err}")
out = out.rstrip()
return None if out == "" else out
return process
def state_created(self):
with self.runner(
"cli_action state backend password config",
output_process=self._process_command_output(True, "already exists"),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="repository")
def state_connected(self):
with self.runner(
"cli_action state backend password fingerprint_tls url config",
output_process=self._process_command_output(True, "already connected"),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="repository")
def state_disconnected(self):
with self.runner(
"cli_action state password config",
output_process=self._process_command_output(True, "does not exist"),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="repository")
def state_synced(self):
with self.runner(
"cli_action state backend password config",
output_process=self._process_command_output(True, "already synced"),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="repository")
def state_throttled(self):
with self.runner(
"cli_action state throttle_operation throttle config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="repository", throttle_operation="set")
def main():
KopiaRepository.execute()
if __name__ == "__main__":
main()

109
plugins/modules/kopia_repository_info.py Normal file
View file

@ -0,0 +1,109 @@
#!/usr/bin/python
# Copyright (c) 2026, Dexter Le <dextersydney2001@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 = r"""
module: kopia_repository_info
short_description: Gather information about a Kopia repository
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Gather read-only information about the current Kopia repository connection and throttle settings.
- Runs C(kopia repository status) and C(kopia repository throttle get).
extends_documentation_fragment:
- community.general._attributes
- community.general._attributes.info_module
- community.general._kopia
"""
EXAMPLES = r"""
- name: Gather Kopia repository info
community.general.kopia_repository_info:
config: /etc/kopia/root.config
register: result
- name: Show repository status
ansible.builtin.debug:
msg: "{{ result.repository_status }}"
- name: Show throttle settings
ansible.builtin.debug:
msg: "{{ result.throttle }}"
"""
RETURN = r"""
repository_status:
description: Output of C(kopia repository status).
type: str
returned: always
sample: |-
Connected to repository: s3:/my-bucket/
Config file: /etc/kopia/root.config
...
throttle:
description: Output of C(kopia repository throttle get) showing current throttle limits.
type: str
returned: always
sample: |-
upload-bytes-per-second: 0
download-bytes-per-second: 0
"""
from ansible_collections.community.general.plugins.module_utils._kopia import (
KOPIA_COMMON_ARGUMENT_SPEC,
kopia_runner,
)
from ansible_collections.community.general.plugins.module_utils._module_helper import ModuleHelper
class KopiaRepositoryInfo(ModuleHelper):
module = dict(
argument_spec=dict(**KOPIA_COMMON_ARGUMENT_SPEC),
supports_check_mode=True,
)
output_params = ["repository_status", "throttle"]
def __init_module__(self):
self.runner = kopia_runner(self.module)
def _process_command_output(self, cli_action=""):
def process(rc, out, err):
if rc != 0:
self.do_raise(f"kopia {cli_action} failed with error (rc={rc}): {err}")
return out.rstrip() if out else None
return process
def __run__(self):
with self.runner(
"status config",
output_process=self._process_command_output("repository status"),
) as ctx:
self.vars.set(
"repository_status",
ctx.run(),
output=True,
)
with self.runner(
"get_throttle config",
output_process=self._process_command_output("repository throttle get"),
) as ctx:
self.vars.set(
"throttle",
ctx.run(),
output=True,
)
def main():
KopiaRepositoryInfo.execute()
if __name__ == "__main__":
main()

176
tests/unit/plugins/module_utils/test__kopia.py Normal file
View file

@ -0,0 +1,176 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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
import pytest
from ansible_collections.community.general.plugins.module_utils._kopia import (
KOPIA_COMMON_ARGUMENT_SPEC,
REPOSITORY_STATE_MAP,
fmt_backend,
)
# ---------------------------------------------------------------------------
# KOPIA_COMMON_ARGUMENT_SPEC
# ---------------------------------------------------------------------------
def test_common_argument_spec_has_password():
assert "password" in KOPIA_COMMON_ARGUMENT_SPEC
spec = KOPIA_COMMON_ARGUMENT_SPEC["password"]
assert spec["type"] == "str"
assert spec["no_log"] is True
def test_common_argument_spec_has_config():
assert "config" in KOPIA_COMMON_ARGUMENT_SPEC
assert KOPIA_COMMON_ARGUMENT_SPEC["config"]["type"] == "path"
def test_common_argument_spec_only_two_keys():
assert set(KOPIA_COMMON_ARGUMENT_SPEC.keys()) == {"password", "config"}
# ---------------------------------------------------------------------------
# REPOSITORY_STATE_MAP
# ---------------------------------------------------------------------------
def test_repository_state_map_entries():
assert REPOSITORY_STATE_MAP["created"] == "create"
assert REPOSITORY_STATE_MAP["connected"] == "connect"
assert REPOSITORY_STATE_MAP["disconnected"] == "disconnect"
assert REPOSITORY_STATE_MAP["synced"] == "sync-to"
assert REPOSITORY_STATE_MAP["throttled"] == "throttle"
# ---------------------------------------------------------------------------
# fmt_backend
# ---------------------------------------------------------------------------
TC_FMT_BACKEND = dict(
server=(
{"provider": "server"},
[],
),
filesystem_path=(
{"provider": "filesystem", "path": "/mnt/backup"},
["filesystem", "--path=/mnt/backup"],
),
filesystem_no_path=(
{"provider": "filesystem"},
["filesystem"],
),
s3_full=(
{
"provider": "s3",
"bucket": "my-bucket",
"access_key": "keyid",
"secret_access_key": "secret",
"endpoint": "https://s3.example.com",
"region": "us-east-1",
"prefix": "backups/",
"session_token": "tok123",
},
[
"s3",
"--bucket=my-bucket",
"--access-key=keyid",
"--secret-access-key=secret",
"--endpoint=https://s3.example.com",
"--region=us-east-1",
"--prefix=backups/",
"--session-token=tok123",
],
),
s3_minimal=(
{"provider": "s3", "bucket": "my-bucket"},
["s3", "--bucket=my-bucket"],
),
azure_full=(
{
"provider": "azure",
"container": "my-container",
"storage_account": "myaccount",
"storage_key": "mykey",
"sas_token": "mytoken",
"storage_domain": "blob.core.windows.net",
"prefix": "data/",
},
[
"azure",
"--container=my-container",
"--storage-account=myaccount",
"--storage-key=mykey",
"--sas-token=mytoken",
"--storage-domain=blob.core.windows.net",
"--prefix=data/",
],
),
azure_minimal=(
{"provider": "azure", "container": "my-container", "storage_account": "myaccount"},
["azure", "--container=my-container", "--storage-account=myaccount"],
),
gcs_full=(
{"provider": "gcs", "bucket": "my-bucket", "credentials_file": "/etc/gcs.json", "prefix": "kopia/"},
["gcs", "--bucket=my-bucket", "--credentials-file=/etc/gcs.json", "--prefix=kopia/"],
),
gdrive=(
{"provider": "gdrive", "folder_id": "abc123", "credentials_file": "/etc/gdrive.json"},
["gdrive", "--folder-id=abc123", "--credentials-file=/etc/gdrive.json"],
),
b2_full=(
{"provider": "b2", "bucket": "my-b2-bucket", "access_key": "kid", "secret_access_key": "sec"},
["b2", "--bucket=my-b2-bucket", "--key-id=kid", "--key=sec"],
),
rclone=(
{"provider": "rclone", "path": "remote:backup"},
["rclone", "--remote-path=remote:backup"],
),
sftp_full=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"port": "22",
"keyfile": "/root/.ssh/id_rsa",
"known_hosts": "/root/.ssh/known_hosts",
},
[
"sftp",
"--path=/backup",
"--host=sftp.example.com",
"--username=admin",
"--port=22",
"--keyfile=/root/.ssh/id_rsa",
"--known-hosts=/root/.ssh/known_hosts",
],
),
webdav_full=(
{
"provider": "webdav",
"url": "https://dav.example.com",
"webdav_username": "user",
"webdav_password": "pass",
},
["webdav", "--url=https://dav.example.com", "--webdav-username=user", "--webdav-password=pass"],
),
none_values_skipped=(
{"provider": "s3", "bucket": "b", "access_key": None, "secret_access_key": None},
["s3", "--bucket=b"],
),
)
TC_FMT_BACKEND_IDS = sorted(TC_FMT_BACKEND.keys())
@pytest.mark.parametrize(
"backend, expected",
(TC_FMT_BACKEND[tc] for tc in TC_FMT_BACKEND_IDS),
ids=TC_FMT_BACKEND_IDS,
)
def test_fmt_backend(backend, expected):
assert fmt_backend(backend) == expected

11
tests/unit/plugins/modules/test_kopia_repository.py Normal file
View file

@ -0,0 +1,11 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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 ansible_collections.community.general.plugins.modules import kopia_repository
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_repository, __name__, mocks=[RunCommandMock])

346
tests/unit/plugins/modules/test_kopia_repository.yaml Normal file
View file

@ -0,0 +1,346 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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
---
anchors:
environ: &env-def {environ_update: {LANGUAGE: C, LC_ALL: C}, check_rc: false}
status_ok: &status-ok
command: [/testbin/kopia, repository, status, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: "Connected to repository."
err: ''
status_empty: &status-empty
command: [/testbin/kopia, repository, status, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 1
out: ''
err: 'repository not connected'
test_cases:
- id: create_s3
input:
state: created
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-bucket
access_key: myaccesskey
secret_access_key: mysecretkey
output:
changed: true
mocks:
run_command:
- *status-empty
- command:
- /testbin/kopia
- repository
- create
- s3
- --bucket=my-bucket
- --access-key=myaccesskey
- --secret-access-key=mysecretkey
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: create_s3_already_exists
input:
state: created
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-bucket
access_key: myaccesskey
secret_access_key: mysecretkey
output:
changed: false
mocks:
run_command:
- *status-ok
- command:
- /testbin/kopia
- repository
- create
- s3
- --bucket=my-bucket
- --access-key=myaccesskey
- --secret-access-key=mysecretkey
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'already exists'
- *status-ok
- id: create_filesystem
input:
state: created
password: secret
backend:
provider: filesystem
path: /mnt/backup/kopia
output:
changed: true
mocks:
run_command:
- command: [/testbin/kopia, repository, status]
environ: *env-def
rc: 1
out: ''
err: 'repository not connected'
- command:
- /testbin/kopia
- repository
- create
- filesystem
- --path=/mnt/backup/kopia
- --password=secret
environ: *env-def
rc: 0
out: ''
err: ''
- command: [/testbin/kopia, repository, status]
environ: *env-def
rc: 0
out: "Connected to repository."
err: ''
- id: connect_server
input:
state: connected
password: secret
config: /etc/kopia/root.config
url: https://kopia.example.com:51515
fingerprint_tls: "AA:BB:CC:DD"
backend:
provider: server
output:
changed: true
mocks:
run_command:
- *status-empty
- command:
- /testbin/kopia
- repository
- connect
- --password=secret
- --server-cert-fingerprint=AA:BB:CC:DD
- --url=https://kopia.example.com:51515
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: connect_azure
input:
state: connected
password: secret
config: /etc/kopia/root.config
backend:
provider: azure
container: my-container
storage_account: myaccount
storage_key: mykey
output:
changed: true
mocks:
run_command:
- *status-empty
- command:
- /testbin/kopia
- repository
- connect
- azure
- --container=my-container
- --storage-account=myaccount
- --storage-key=mykey
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: disconnect
input:
state: disconnected
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *status-ok
- command:
- /testbin/kopia
- repository
- disconnect
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-empty
- id: sync_s3
input:
state: synced
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-sync-bucket
access_key: myaccesskey
secret_access_key: mysecretkey
output:
changed: false
mocks:
run_command:
- *status-ok
- command:
- /testbin/kopia
- repository
- sync-to
- s3
- --bucket=my-sync-bucket
- --access-key=myaccesskey
- --secret-access-key=mysecretkey
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: throttle_set
input:
state: throttled
config: /etc/kopia/root.config
throttle:
upload_bytes_per_second: 1048576
download_bytes_per_second: 5242880
concurrent_reads: 4
concurrent_writes: 2
output:
changed: false
mocks:
run_command:
- *status-ok
- command:
- /testbin/kopia
- repository
- throttle
- set
- --download-bytes-per-second
- "5242880"
- --upload-bytes-per-second
- "1048576"
- --concurrent-reads
- "4"
- --concurrent-writes
- "2"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: throttle_set_rate_limits
input:
state: throttled
config: /etc/kopia/root.config
throttle:
read_requests_per_second: 10.0
write_requests_per_second: 5.0
list_requests_per_second: 2.5
output:
changed: false
mocks:
run_command:
- *status-ok
- command:
- /testbin/kopia
- repository
- throttle
- set
- --read-requests-per-second
- "10.0"
- --write-requests-per-second
- "5.0"
- --list-requests-per-second
- "2.5"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *status-ok
- id: check_mode_create_not_exists
flags: {check: true}
input:
state: created
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-bucket
access_key: myaccesskey
secret_access_key: mysecretkey
output:
changed: true
mocks:
run_command:
- *status-empty
- id: check_mode_create_already_exists
flags: {check: true}
input:
state: created
password: secret
config: /etc/kopia/root.config
backend:
provider: s3
bucket: my-bucket
access_key: myaccesskey
secret_access_key: mysecretkey
output:
changed: false
mocks:
run_command:
- *status-ok
- id: check_mode_disconnect_connected
flags: {check: true}
input:
state: disconnected
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *status-ok
- id: check_mode_disconnect_not_connected
flags: {check: true}
input:
state: disconnected
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *status-empty

11
tests/unit/plugins/modules/test_kopia_repository_info.py Normal file
View file

@ -0,0 +1,11 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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 ansible_collections.community.general.plugins.modules import kopia_repository_info
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_repository_info, __name__, mocks=[RunCommandMock])

73
tests/unit/plugins/modules/test_kopia_repository_info.yaml Normal file
View file

@ -0,0 +1,73 @@
# Copyright (c) 2026, Dexter Le <dextersydney2001@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
---
anchors:
environ: &env-def {environ_update: {LANGUAGE: C, LC_ALL: C}, check_rc: false}
test_cases:
- id: info_connected
input:
config: /etc/kopia/root.config
output:
changed: false
repository_status: |-
Connected to repository: s3:/my-bucket/
Config file: /etc/kopia/root.config
throttle: |-
upload-bytes-per-second: 0
download-bytes-per-second: 0
mocks:
run_command:
- command: [/testbin/kopia, repository, status, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: |-
Connected to repository: s3:/my-bucket/
Config file: /etc/kopia/root.config
err: ''
- command: [/testbin/kopia, repository, throttle, get, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: |-
upload-bytes-per-second: 0
download-bytes-per-second: 0
err: ''
- id: info_not_connected
input:
config: /etc/kopia/root.config
output:
failed: true
msg: "kopia repository status failed with error (rc=1): repository not connected"
mocks:
run_command:
- command: [/testbin/kopia, repository, status, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 1
out: ''
err: 'repository not connected'
- id: info_no_config
input: {}
output:
changed: false
repository_status: "Connected to repository: filesystem:/mnt/backup/kopia"
throttle: |-
upload-bytes-per-second: 1048576
download-bytes-per-second: 5242880
mocks:
run_command:
- command: [/testbin/kopia, repository, status]
environ: *env-def
rc: 0
out: "Connected to repository: filesystem:/mnt/backup/kopia"
err: ''
- command: [/testbin/kopia, repository, throttle, get]
environ: *env-def
rc: 0
out: |-
upload-bytes-per-second: 1048576
download-bytes-per-second: 5242880
err: ''