ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-06-12 02:55:29 +00:00
Code Activity

Merge 24a6aefde5 into 3774ca20d2

Browse source
This commit is contained in:
munchtoast 2026-06-08 16:25:38 -04:00 • committed by GitHub
commit 45a99322a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 3374 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

8
.github/BOTMETA.yml vendored
View file

@ -905,10 +905,18 @@ files:
maintainers: ahussey-redhat
$modules/kibana_plugin.py:
maintainers: barryib
$modules/kopia_notification.py:
maintainers: munchtoast
$modules/kopia_policy.py:
maintainers: munchtoast
$modules/kopia_repository_info.py:
maintainers: munchtoast
$modules/kopia_repository.py:
maintainers: munchtoast
$modules/kopia_server.py:
maintainers: munchtoast
$modules/kopia_snapshot.py:
maintainers: munchtoast
$modules/krb_ticket.py:
maintainers: abakanovskii
$modules/launchd.py:

130
plugins/module_utils/_kopia.py
View file

@ -14,73 +14,102 @@ from ansible_collections.community.general.plugins.module_utils._cmd_runner impo
if t.TYPE_CHECKING:
from ansible.module_utils.basic import AnsibleModule
# Maps kopia_repository module state values to kopia CLI subcommands.
# Maps state values across all kopia modules to their kopia CLI subcommands.
# Used with cmd_runner_fmt.as_map() for the 'state' arg format.
REPOSITORY_STATE_MAP = {
STATE_MAP = {
# kopia_repository
"created": "create",
"connected": "connect",
"disconnected": "disconnect",
"synced": "sync-to",
"throttled": "throttle",
# kopia_snapshot
"deleted": "delete",
"expired": "expire",
"listed": "list",
"verified": "verify",
# kopia_policy
"set": "set",
"shown": "show",
}
# 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.
# Maps backend provider names to their CLI parameter definitions.
# Each provider maps param_name -> (flag, type) where type is:
# "str" - emit --flag=value (skip if None)
# "bool" - emit --flag only when value is True (skip if False or None)
# "list" - emit --flag item once per item in the list (skip if empty or None)
# 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 = {
_PROVIDER_BACKEND_MAP: dict[str, dict[str, tuple[str, str]]] = {
"azure": {
"container": "--container",
"storage_account": "--storage-account",
"storage_key": "--storage-key",
"sas_token": "--sas-token",
"storage_domain": "--storage-domain",
"prefix": "--prefix",
"container": ("--container", "str"),
"storage_account": ("--storage-account", "str"),
"storage_key": ("--storage-key", "str"),
"sas_token": ("--sas-token", "str"),
"storage_domain": ("--storage-domain", "str"),
"prefix": ("--prefix", "str"),
"client_id": ("--client-id", "str"),
"client_secret": ("--client-secret", "str"),
"tenant_id": ("--tenant-id", "str"),
"azure_federated_token_file": ("--azure-federated-token-file", "str"),
},
"b2": {
"bucket": "--bucket",
"access_key": "--key-id",
"secret_access_key": "--key",
"prefix": "--prefix",
"bucket": ("--bucket", "str"),
"access_key": ("--key-id", "str"),
"secret_access_key": ("--key", "str"),
"prefix": ("--prefix", "str"),
},
"filesystem": {
"path": "--path",
"path": ("--path", "str"),
},
"gcs": {
"bucket": "--bucket",
"credentials_file": "--credentials-file",
"prefix": "--prefix",
"bucket": ("--bucket", "str"),
"credentials_file": ("--credentials-file", "str"),
"prefix": ("--prefix", "str"),
"embed_credentials": ("--embed-credentials", "bool"),
"read_only": ("--read-only", "bool"),
},
"gdrive": {
"folder_id": "--folder-id",
"credentials_file": "--credentials-file",
"folder_id": ("--folder-id", "str"),
"credentials_file": ("--credentials-file", "str"),
"read_only": ("--read-only", "bool"),
},
"rclone": {
"path": "--remote-path",
"path": ("--remote-path", "str"),
"rclone_exe": ("--rclone-exe", "str"),
"rclone_args": ("--rclone-args", "list"),
"rclone_env": ("--rclone-env", "list"),
"embed_rclone_config": ("--embed-rclone-config", "bool"),
},
"s3": {
"bucket": "--bucket",
"access_key": "--access-key",
"secret_access_key": "--secret-access-key",
"endpoint": "--endpoint",
"region": "--region",
"prefix": "--prefix",
"session_token": "--session-token",
"bucket": ("--bucket", "str"),
"access_key": ("--access-key", "str"),
"secret_access_key": ("--secret-access-key", "str"),
"endpoint": ("--endpoint", "str"),
"region": ("--region", "str"),
"prefix": ("--prefix", "str"),
"session_token": ("--session-token", "str"),
},
"sftp": {
"path": "--path",
"host": "--host",
"username": "--username",
"port": "--port",
"keyfile": "--keyfile",
"known_hosts": "--known-hosts",
"path": ("--path", "str"),
"host": ("--host", "str"),
"username": ("--username", "str"),
"port": ("--port", "str"),
"keyfile": ("--keyfile", "str"),
"known_hosts": ("--known-hosts", "str"),
"sftp_password": ("--sftp-password", "str"),
"key_data": ("--key-data", "str"),
"known_hosts_data": ("--known-hosts-data", "str"),
"embed_credentials": ("--embed-credentials", "bool"),
"external": ("--external", "bool"),
"ssh_command": ("--ssh-command", "str"),
"ssh_args": ("--ssh-args", "list"),
},
"webdav": {
"url": "--url",
"webdav_username": "--webdav-username",
"webdav_password": "--webdav-password",
"url": ("--url", "str"),
"webdav_username": ("--webdav-username", "str"),
"webdav_password": ("--webdav-password", "str"),
},
}
@ -96,7 +125,12 @@ 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, ...]
[provider_name, --flag1=value1, --flag2, ...]
Param types:
str - emits --flag=value; skipped when value is None.
bool - emits --flag when True; skipped when False or None.
list - emits --flag item once per item; skipped when empty or None.
For the "server" provider, returns [] because server connect uses top-level
flags (--url, --server-cert-fingerprint) passed separately.
@ -105,10 +139,18 @@ def fmt_backend(value):
if provider == "server":
return []
result = [provider]
for param_name, flag in _PROVIDER_BACKEND_MAP[provider].items():
for param_name, (flag, kind) in _PROVIDER_BACKEND_MAP[provider].items():
param_value = value.get(param_name)
if param_value is not None:
result.append(f"{flag}={param_value}")
if kind == "str":
if param_value is not None:
result.append(f"{flag}={param_value}")
elif kind == "bool":
if param_value:
result.append(flag)
elif kind == "list":
if param_value:
for item in param_value:
result.extend([flag, item])
return result
@ -143,7 +185,7 @@ def kopia_runner(module: AnsibleModule, extra_formats: dict | None = None, **kwa
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),
state=cmd_runner_fmt.as_map(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"),

502
plugins/modules/kopia_notification.py Normal file
View file

@ -0,0 +1,502 @@
#!/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_notification
short_description: Manage Kopia notification profiles and templates
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Manage Kopia notification profiles and message templates using the Kopia CLI.
- Supports configuring email, Pushover, and webhook notification profiles,
as well as listing, showing, deleting, and testing profiles.
- Supports listing, showing, setting, and removing notification message templates.
extends_documentation_fragment:
- community.general._attributes
- community.general._kopia
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
state:
description:
- Desired state of the notification resource.
- V(profile_email) creates or updates an email notification profile.
Requires O(profile_name), O(smtp_server), O(mail_to), and O(mail_from).
- V(profile_pushover) creates or updates a Pushover notification profile.
Requires O(profile_name), O(pushover_app_token), and O(pushover_user_key).
- V(profile_webhook) creates or updates a webhook notification profile.
Requires O(profile_name) and O(webhook_endpoint).
- V(profile_deleted) removes a notification profile. Requires O(profile_name).
- V(profile_tested) sends a test notification via the named profile.
Requires O(profile_name).
- V(profiles_listed) lists all configured notification profiles.
- V(profile_shown) displays a specific notification profile. Requires O(profile_name).
- V(template_set) sets a notification message template.
Requires O(template_name) and O(template_file).
- V(template_removed) removes a notification message template. Requires O(template_name).
- V(templates_listed) lists all notification message templates.
- V(template_shown) displays a specific notification message template.
Requires O(template_name).
type: str
choices:
- profile_email
- profile_pushover
- profile_webhook
- profile_deleted
- profile_tested
- profiles_listed
- profile_shown
- template_set
- template_removed
- templates_listed
- template_shown
default: profiles_listed
profile_name:
description:
- Name of the notification profile to create, update, delete, test, or show.
- Required if O(state=profile_email), O(state=profile_pushover),
O(state=profile_webhook), O(state=profile_deleted), O(state=profile_tested),
or O(state=profile_shown).
type: str
min_severity:
description:
- Minimum notification severity level that triggers this profile.
- Optional for O(state=profile_email), O(state=profile_pushover),
or O(state=profile_webhook).
type: str
choices: [verbose, info, warning, error]
format:
description:
- Message format for notifications.
- Optional for O(state=profile_email), O(state=profile_pushover),
or O(state=profile_webhook).
type: str
choices: [html, md, txt]
send_test:
description:
- When V(true), send a test notification immediately after configuring the profile.
- Optional for O(state=profile_email), O(state=profile_pushover),
or O(state=profile_webhook).
type: bool
default: false
smtp_server:
description:
- SMTP server hostname or IP address.
- Required if O(state=profile_email).
type: str
smtp_port:
description:
- SMTP server port.
- Optional if O(state=profile_email).
type: int
smtp_username:
description:
- SMTP authentication username.
- Optional if O(state=profile_email).
type: str
smtp_password:
description:
- SMTP authentication password.
- Optional if O(state=profile_email).
type: str
smtp_identity:
description:
- SMTP SASL identity string.
- Optional if O(state=profile_email).
type: str
mail_from:
description:
- Sender email address.
- Required if O(state=profile_email).
type: str
mail_to:
description:
- Recipient email address.
- Required if O(state=profile_email).
type: str
mail_cc:
description:
- CC email address.
- Optional if O(state=profile_email).
type: str
pushover_app_token:
description:
- Pushover application token.
- Required if O(state=profile_pushover).
type: str
pushover_user_key:
description:
- Pushover user key.
- Required if O(state=profile_pushover).
type: str
webhook_endpoint:
description:
- Webhook destination URL.
- Required if O(state=profile_webhook).
type: str
webhook_method:
description:
- HTTP method to use when calling the webhook.
- Optional if O(state=profile_webhook).
type: str
choices: [GET, POST, PUT, PATCH]
webhook_headers:
description:
- List of HTTP headers to include with webhook requests, each in C(key:value) format.
- Optional if O(state=profile_webhook).
type: list
elements: str
template_name:
description:
- Name of the notification template to set, remove, or show.
- Required if O(state=template_set), O(state=template_removed),
or O(state=template_shown).
type: str
template_file:
description:
- Path to a file containing the template body to set.
- Required if O(state=template_set).
type: path
"""
EXAMPLES = r"""
- name: Configure an email notification profile
community.general.kopia_notification:
state: profile_email
profile_name: ops-email
smtp_server: smtp.example.com
smtp_port: 587
smtp_username: notify@example.com
smtp_password: smtpsecret
mail_from: notify@example.com
mail_to: ops@example.com
min_severity: warning
config: /etc/kopia/root.config
- name: Configure a Pushover notification profile
community.general.kopia_notification:
state: profile_pushover
profile_name: ops-pushover
pushover_app_token: "aToken123"
pushover_user_key: "uKey456"
min_severity: error
config: /etc/kopia/root.config
- name: Configure a webhook notification profile
community.general.kopia_notification:
state: profile_webhook
profile_name: ops-webhook
webhook_endpoint: https://hooks.example.com/kopia
webhook_method: POST
webhook_headers:
- "Authorization:Bearer mytoken"
format: html
config: /etc/kopia/root.config
- name: Send a test notification
community.general.kopia_notification:
state: profile_tested
profile_name: ops-email
config: /etc/kopia/root.config
- name: Show a notification profile
community.general.kopia_notification:
state: profile_shown
profile_name: ops-email
config: /etc/kopia/root.config
- name: List all notification profiles
community.general.kopia_notification:
state: profiles_listed
config: /etc/kopia/root.config
- name: Delete a notification profile
community.general.kopia_notification:
state: profile_deleted
profile_name: ops-email
config: /etc/kopia/root.config
- name: Set a notification template from a file
community.general.kopia_notification:
state: template_set
template_name: snapshot-complete
template_file: /etc/kopia/templates/snapshot-complete.html
config: /etc/kopia/root.config
- name: List all notification templates
community.general.kopia_notification:
state: templates_listed
config: /etc/kopia/root.config
- name: Show a notification template
community.general.kopia_notification:
state: template_shown
template_name: snapshot-complete
config: /etc/kopia/root.config
- name: Remove a notification template
community.general.kopia_notification:
state: template_removed
template_name: snapshot-complete
config: /etc/kopia/root.config
"""
RETURN = r"""
kopia_notification:
description: Output from the Kopia notification command.
type: str
sample: ""
returned: always
"""
from ansible_collections.community.general.plugins.module_utils._cmd_runner import cmd_runner_fmt
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
# Maps each module state to the kopia CLI words that follow `kopia notification`.
# profile configure email → ("profile", "configure", "email")
# profile delete → ("profile", "delete")
# template set → ("template", "set")
# etc.
_STATE_CLI_MAP = {
"profile_email": ("profile", "configure", "email"),
"profile_pushover": ("profile", "configure", "pushover"),
"profile_webhook": ("profile", "configure", "webhook"),
"profile_deleted": ("profile", "delete"),
"profile_tested": ("profile", "test"),
"profiles_listed": ("profile", "list"),
"profile_shown": ("profile", "show"),
"template_set": ("template", "set"),
"template_removed": ("template", "remove"),
"templates_listed": ("template", "list"),
"template_shown": ("template", "show"),
}
# Read-only states that must not be skipped in check mode.
_READONLY_STATES = frozenset(["profiles_listed", "profile_shown", "templates_listed", "template_shown"])
def _fmt_webhook_headers(value):
"""Expand a list of key:value strings into repeated --http-header flags."""
if not value:
return []
result = []
for header in value:
result.extend(["--http-header", header])
return result
class KopiaNotification(StateModuleHelper):
module = dict(
supports_check_mode=True,
argument_spec=dict(
**KOPIA_COMMON_ARGUMENT_SPEC,
state=dict(
type="str",
default="profiles_listed",
choices=[
"profile_email",
"profile_pushover",
"profile_webhook",
"profile_deleted",
"profile_tested",
"profiles_listed",
"profile_shown",
"template_set",
"template_removed",
"templates_listed",
"template_shown",
],
),
profile_name=dict(type="str"),
min_severity=dict(type="str", choices=["verbose", "info", "warning", "error"]),
format=dict(type="str", choices=["html", "md", "txt"]),
send_test=dict(type="bool", default=False),
# email
smtp_server=dict(type="str"),
smtp_port=dict(type="int"),
smtp_username=dict(type="str"),
smtp_password=dict(type="str", no_log=True),
smtp_identity=dict(type="str"),
mail_from=dict(type="str"),
mail_to=dict(type="str"),
mail_cc=dict(type="str"),
# pushover
pushover_app_token=dict(type="str", no_log=True),
pushover_user_key=dict(type="str", no_log=True),
# webhook
webhook_endpoint=dict(type="str"),
webhook_method=dict(type="str", choices=["GET", "POST", "PUT", "PATCH"]),
webhook_headers=dict(type="list", elements="str"),
# template
template_name=dict(type="str"),
template_file=dict(type="path"),
),
required_if=[
("state", "profile_email", ["profile_name", "smtp_server", "mail_from", "mail_to"]),
("state", "profile_pushover", ["profile_name", "pushover_app_token", "pushover_user_key"]),
("state", "profile_webhook", ["profile_name", "webhook_endpoint"]),
("state", "profile_deleted", ["profile_name"]),
("state", "profile_tested", ["profile_name"]),
("state", "profile_shown", ["profile_name"]),
("state", "template_set", ["template_name", "template_file"]),
("state", "template_removed", ["template_name"]),
("state", "template_shown", ["template_name"]),
],
)
def __init_module__(self):
self.runner = kopia_runner(
self.module,
extra_formats=dict(
list_profiles=cmd_runner_fmt.as_fixed("notification", "profile", "list"),
notif_group=cmd_runner_fmt.as_list(),
notif_subcommand=cmd_runner_fmt.as_list(),
notif_provider=cmd_runner_fmt.as_list(),
profile_name=cmd_runner_fmt.as_opt_val("--profile-name"),
min_severity=cmd_runner_fmt.as_opt_val("--min-severity"),
format=cmd_runner_fmt.as_opt_val("--format"),
send_test=cmd_runner_fmt.as_bool("--send-test-notification"),
smtp_server=cmd_runner_fmt.as_opt_val("--smtp-server"),
smtp_port=cmd_runner_fmt.as_opt_val("--smtp-port"),
smtp_username=cmd_runner_fmt.as_opt_val("--smtp-username"),
smtp_password=cmd_runner_fmt.as_opt_val("--smtp-password"),
smtp_identity=cmd_runner_fmt.as_opt_val("--smtp-identity"),
mail_from=cmd_runner_fmt.as_opt_val("--mail-from"),
mail_to=cmd_runner_fmt.as_opt_val("--mail-to"),
mail_cc=cmd_runner_fmt.as_opt_val("--mail-cc"),
pushover_app_token=cmd_runner_fmt.as_opt_val("--app-token"),
pushover_user_key=cmd_runner_fmt.as_opt_val("--user-key"),
webhook_endpoint=cmd_runner_fmt.as_opt_val("--endpoint"),
webhook_method=cmd_runner_fmt.as_opt_val("--method"),
webhook_headers=cmd_runner_fmt.as_func(_fmt_webhook_headers),
template_name=cmd_runner_fmt.as_list(),
template_file=cmd_runner_fmt.as_list(),
),
)
self.vars.set("previous_value", self._get()["out"])
self.vars.set("value", self.vars.previous_value, change=True, diff=True)
def __quit_module__(self):
self.vars.set("value", self._get()["out"])
def _get(self):
with self.runner("list_profiles 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() if out else ""
return None if out == "" else out
return process
def _run_notif_cmd(self, args_order, ignore_err_msg="", **run_kwargs):
cli_words = _STATE_CLI_MAP[self.vars.state]
notif_group = cli_words[0]
notif_subcommand = cli_words[1]
notif_provider = cli_words[2] if len(cli_words) == 3 else None
check_mode_skip = self.vars.state not in _READONLY_STATES
with self.runner(
args_order,
output_process=self._process_command_output(True, ignore_err_msg),
check_mode_skip=check_mode_skip,
) as ctx:
ctx.run(
cli_action="notification",
notif_group=notif_group,
notif_subcommand=notif_subcommand,
notif_provider=notif_provider,
**run_kwargs,
)
def state_profile_email(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand notif_provider"
" profile_name smtp_server smtp_port smtp_username smtp_password"
" smtp_identity mail_from mail_to mail_cc"
" min_severity format send_test config",
)
def state_profile_pushover(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand notif_provider"
" profile_name pushover_app_token pushover_user_key"
" min_severity format send_test config",
)
def state_profile_webhook(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand notif_provider"
" profile_name webhook_endpoint webhook_method webhook_headers"
" min_severity format send_test config",
)
def state_profile_deleted(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand profile_name config",
ignore_err_msg="no such profile",
)
def state_profile_tested(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand profile_name config",
)
def state_profiles_listed(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand config",
)
def state_profile_shown(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand profile_name config",
)
def state_template_set(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand template_name template_file config",
)
def state_template_removed(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand template_name config",
ignore_err_msg="no such template",
)
def state_templates_listed(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand config",
)
def state_template_shown(self):
self._run_notif_cmd(
"cli_action notif_group notif_subcommand template_name config",
)
def main():
KopiaNotification.execute()
if __name__ == "__main__":
main()

402
plugins/modules/kopia_policy.py Normal file
View file

@ -0,0 +1,402 @@
#!/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_policy
short_description: Manage Kopia snapshot policies
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Manage Kopia snapshot policies using the Kopia CLI.
- Supports setting, deleting, showing, and listing policies.
- Policies control retention, scheduling, file exclusions, and compression for snapshots.
extends_documentation_fragment:
- community.general._attributes
- community.general._kopia
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
state:
description:
- Desired state of the Kopia policy.
- V(set) creates or updates a policy. At least one policy option must be provided.
- V(deleted) removes the policy for O(target) or the global policy when O(global_policy=true).
- V(listed) lists all defined policies.
- V(shown) displays the effective policy for O(target), including inherited values.
type: str
choices: [set, deleted, listed, shown]
default: set
target:
description:
- Policy target in one of the following forms.
- A per-path policy C(user@host:/path).
- A per-host policy C(@host).
- A per-user policy C(user@host).
- Required if O(state=set), O(state=deleted), or O(state=shown), unless O(global_policy=true).
type: str
global_policy:
description:
- When V(true), operate on the global policy instead of a per-target policy.
- When V(true), O(target) must not be set.
type: bool
default: false
retention:
description:
- Snapshot retention settings. All sub-options accept an integer or the string C(inherit)
to remove an override and fall back to the parent policy.
type: dict
suboptions:
keep_latest:
description:
- Number of most-recent snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
keep_hourly:
description:
- Number of most-recent hourly snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
keep_daily:
description:
- Number of most-recent daily snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
keep_weekly:
description:
- Number of most-recent weekly snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
keep_monthly:
description:
- Number of most-recent monthly snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
keep_annual:
description:
- Number of most-recent annual snapshots to keep.
- Pass C(inherit) to remove this override.
type: str
ignore_identical:
description:
- Do not save a new snapshot if its contents are identical to the previous one.
- Accepts V(true), V(false), or C(inherit).
type: str
scheduling:
description:
- Snapshot scheduling settings.
type: dict
suboptions:
interval:
description:
- Time between automatic snapshots, for example C(1h), C(30m), or C(24h).
type: str
times:
description:
- List of times of day at which to take snapshots, in C(HH:mm) format.
- Pass C(inherit) as the only list entry to remove this override.
type: list
elements: str
manual:
description:
- When V(true), only create snapshots manually and disable automatic scheduling.
type: bool
files:
description:
- File and directory exclusion settings.
type: dict
suboptions:
add_ignore:
description:
- List of path patterns to add to the ignore list.
type: list
elements: str
remove_ignore:
description:
- List of path patterns to remove from the ignore list.
type: list
elements: str
max_file_size:
description:
- Exclude files larger than this size. Accepts a byte count or human-readable string
such as C(100MB).
type: str
one_file_system:
description:
- When V(true), do not cross filesystem boundaries during backup.
- Accepts V(true), V(false), or C(inherit).
type: str
ignore_cache_dirs:
description:
- When V(true), ignore directories containing a C(CACHEDIR.TAG) file.
- Accepts V(true), V(false), or C(inherit).
type: str
compression:
description:
- Compression algorithm to apply to snapshot data.
- Refer to C(kopia policy set --compression help) for the list of supported algorithms.
type: str
"""
EXAMPLES = r"""
- name: Set a retention policy for a path
community.general.kopia_policy:
state: set
target: "user@hostname:/home/user"
retention:
keep_latest: "10"
keep_daily: "7"
keep_weekly: "4"
keep_monthly: "6"
config: /etc/kopia/root.config
- name: Set a scheduled snapshot policy
community.general.kopia_policy:
state: set
target: "user@hostname:/var/www"
scheduling:
times:
- "02:00"
- "14:00"
config: /etc/kopia/root.config
- name: Set the global policy with compression and ignore rules
community.general.kopia_policy:
state: set
global_policy: true
compression: zstd
files:
add_ignore:
- "*.tmp"
- ".cache"
ignore_cache_dirs: "true"
config: /etc/kopia/root.config
- name: Inherit keep_daily from the parent policy
community.general.kopia_policy:
state: set
target: "user@hostname:/home/user"
retention:
keep_daily: inherit
config: /etc/kopia/root.config
- name: Show the effective policy for a target
community.general.kopia_policy:
state: shown
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
- name: List all defined policies
community.general.kopia_policy:
state: listed
config: /etc/kopia/root.config
- name: Delete a policy for a specific target
community.general.kopia_policy:
state: deleted
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
"""
RETURN = r"""
kopia_policy:
description: Output from the Kopia policy command.
type: str
sample: ""
returned: always
"""
from ansible_collections.community.general.plugins.module_utils._cmd_runner import cmd_runner_fmt
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
def _fmt_retention(value):
"""Expand the retention dict into --keep-* and --ignore-identical-snapshots flags."""
if not value:
return []
flag_map = {
"keep_latest": "--keep-latest",
"keep_hourly": "--keep-hourly",
"keep_daily": "--keep-daily",
"keep_weekly": "--keep-weekly",
"keep_monthly": "--keep-monthly",
"keep_annual": "--keep-annual",
"ignore_identical": "--ignore-identical-snapshots",
}
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 _fmt_scheduling(value):
"""Expand the scheduling dict into --snapshot-interval, --snapshot-time, --manual flags."""
if not value:
return []
result = []
if value.get("interval") is not None:
result.extend(["--snapshot-interval", value["interval"]])
if value.get("times") is not None:
result.extend(["--snapshot-time", ",".join(value["times"])])
if value.get("manual"):
result.append("--manual")
return result
def _fmt_files(value):
"""Expand the files dict into file-handling flags."""
if not value:
return []
result = []
for path in value.get("add_ignore") or []:
result.extend(["--add-ignore", path])
for path in value.get("remove_ignore") or []:
result.extend(["--remove-ignore", path])
if value.get("max_file_size") is not None:
result.extend(["--max-file-size", value["max_file_size"]])
if value.get("one_file_system") is not None:
result.extend(["--one-file-system", value["one_file_system"]])
if value.get("ignore_cache_dirs") is not None:
result.extend(["--ignore-cache-dirs", value["ignore_cache_dirs"]])
return result
class KopiaPolicy(StateModuleHelper):
module = dict(
supports_check_mode=True,
argument_spec=dict(
**KOPIA_COMMON_ARGUMENT_SPEC,
state=dict(
type="str",
default="set",
choices=["set", "deleted", "listed", "shown"],
),
target=dict(type="str"),
global_policy=dict(type="bool", default=False),
retention=dict(
type="dict",
options=dict(
keep_latest=dict(type="str"),
keep_hourly=dict(type="str"),
keep_daily=dict(type="str"),
keep_weekly=dict(type="str"),
keep_monthly=dict(type="str"),
keep_annual=dict(type="str"),
ignore_identical=dict(type="str"),
),
),
scheduling=dict(
type="dict",
options=dict(
interval=dict(type="str"),
times=dict(type="list", elements="str"),
manual=dict(type="bool"),
),
),
files=dict(
type="dict",
options=dict(
add_ignore=dict(type="list", elements="str"),
remove_ignore=dict(type="list", elements="str"),
max_file_size=dict(type="str"),
one_file_system=dict(type="str"),
ignore_cache_dirs=dict(type="str"),
),
),
compression=dict(type="str"),
),
required_if=[
("state", "set", ["target", "global_policy"], True),
("state", "deleted", ["target", "global_policy"], True),
("state", "shown", ["target", "global_policy"], True),
],
)
def __init_module__(self):
self.runner = kopia_runner(
self.module,
extra_formats=dict(
list_policies=cmd_runner_fmt.as_fixed("policy", "list"),
target=cmd_runner_fmt.as_list(),
global_policy=cmd_runner_fmt.as_bool("--global"),
retention=cmd_runner_fmt.as_func(_fmt_retention),
scheduling=cmd_runner_fmt.as_func(_fmt_scheduling),
files=cmd_runner_fmt.as_func(_fmt_files),
compression=cmd_runner_fmt.as_opt_val("--compression"),
),
)
self.vars.set("previous_value", self._get()["out"])
self.vars.set("value", self.vars.previous_value, change=True, diff=True)
def __quit_module__(self):
self.vars.set("value", self._get()["out"])
def _get(self):
with self.runner("list_policies 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() if out else ""
return None if out == "" else out
return process
def state_set(self):
with self.runner(
"cli_action state target global_policy retention scheduling files compression config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="policy")
def state_deleted(self):
with self.runner(
"cli_action state target global_policy config",
output_process=self._process_command_output(True, "no such policy"),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="policy")
def state_listed(self):
with self.runner(
"cli_action state config",
output_process=self._process_command_output(True),
) as ctx:
ctx.run(cli_action="policy")
def state_shown(self):
with self.runner(
"cli_action state target global_policy config",
output_process=self._process_command_output(True),
) as ctx:
ctx.run(cli_action="policy")
def main():
KopiaPolicy.execute()
if __name__ == "__main__":
main()

304
plugins/modules/kopia_server.py Normal file
View file

@ -0,0 +1,304 @@
#!/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_server
short_description: Manage Kopia server users and ACL entries
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Manage users and access control list (ACL) entries on a Kopia repository server.
- Supports creating, updating, and deleting server users, and adding, listing,
and deleting ACL rules.
- This module targets the repository-side user and ACL configuration stored inside
the repository itself, not the running server process.
- To manage the server process lifecycle use your system's service manager (for
example C(ansible.builtin.systemd)).
extends_documentation_fragment:
- community.general._attributes
- community.general._kopia
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
state:
description:
- Desired state of the resource.
- V(user_present) creates or updates a server user. Requires O(username).
- V(user_absent) removes a server user. Requires O(username).
- V(users_listed) lists all server users. No additional options required.
- V(acl_present) adds an ACL entry. Requires O(acl_user) and O(acl_access).
- V(acl_absent) removes an ACL entry. Requires O(acl_user).
- V(acl_listed) lists all ACL entries. No additional options required.
- V(acl_enabled) enables ACL enforcement and installs default entries.
No additional options required.
type: str
choices: [user_present, user_absent, users_listed, acl_present, acl_absent, acl_listed, acl_enabled]
default: user_present
username:
description:
- Repository server username in C(user@hostname) format.
- Required if O(state=user_present) or O(state=user_absent).
type: str
user_password:
description:
- Password for the server user.
- Required when O(state=user_present) and O(user_password_hash) is not set.
- Mutually exclusive with O(user_password_hash).
type: str
user_password_hash:
description:
- Pre-hashed password for the server user.
- Required when O(state=user_present) and O(user_password) is not set.
- Mutually exclusive with O(user_password).
type: str
acl_user:
description:
- User the ACL rule applies to, in C(user@hostname) format.
- Required if O(state=acl_present) or O(state=acl_absent).
type: str
acl_access:
description:
- Access level granted to O(acl_user).
- Required if O(state=acl_present).
- Refer to Kopia documentation for supported access level values.
type: str
acl_target:
description:
- Manifests targeted by the ACL rule, in C(type:T,key1:value1,...) format.
- Optional if O(state=acl_present).
type: str
acl_no_overwrite:
description:
- When V(true), do not overwrite an existing rule with the same user and target.
- Optional if O(state=acl_present).
type: bool
default: false
"""
EXAMPLES = r"""
- name: Add a repository server user
community.general.kopia_server:
state: user_present
username: alice@backuphost
user_password: secretpassword
config: /etc/kopia/root.config
- name: Update a user with a pre-hashed password
community.general.kopia_server:
state: user_present
username: alice@backuphost
user_password_hash: "$2a$12$..."
config: /etc/kopia/root.config
- name: Remove a repository server user
community.general.kopia_server:
state: user_absent
username: alice@backuphost
config: /etc/kopia/root.config
- name: List all server users
community.general.kopia_server:
state: users_listed
config: /etc/kopia/root.config
- name: Enable ACL enforcement with default entries
community.general.kopia_server:
state: acl_enabled
config: /etc/kopia/root.config
- name: Add an ACL entry granting full access
community.general.kopia_server:
state: acl_present
acl_user: alice@backuphost
acl_access: FULL
config: /etc/kopia/root.config
- name: Add a targeted ACL entry
community.general.kopia_server:
state: acl_present
acl_user: bob@backuphost
acl_access: READ
acl_target: "type:snapshot,username:bob"
config: /etc/kopia/root.config
- name: Delete an ACL entry
community.general.kopia_server:
state: acl_absent
acl_user: alice@backuphost
config: /etc/kopia/root.config
- name: List all ACL entries
community.general.kopia_server:
state: acl_listed
config: /etc/kopia/root.config
"""
RETURN = r"""
kopia_server:
description: Output from the Kopia server command.
type: str
sample: ""
returned: always
"""
from ansible_collections.community.general.plugins.module_utils._cmd_runner import cmd_runner_fmt
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
# Maps module states to (cli_group, cli_subcommand) pairs used in _run_server_cmd().
_STATE_CLI_MAP = {
"user_present": ("users", "set"),
"user_absent": ("users", "delete"),
"users_listed": ("users", "list"),
"acl_present": ("acl", "add"),
"acl_absent": ("acl", "delete"),
"acl_listed": ("acl", "list"),
"acl_enabled": ("acl", "enable"),
}
class KopiaServer(StateModuleHelper):
module = dict(
supports_check_mode=True,
argument_spec=dict(
**KOPIA_COMMON_ARGUMENT_SPEC,
state=dict(
type="str",
default="user_present",
choices=[
"user_present",
"user_absent",
"users_listed",
"acl_present",
"acl_absent",
"acl_listed",
"acl_enabled",
],
),
username=dict(type="str"),
user_password=dict(type="str", no_log=True),
user_password_hash=dict(type="str", no_log=True),
acl_user=dict(type="str"),
acl_access=dict(type="str"),
acl_target=dict(type="str"),
acl_no_overwrite=dict(type="bool", default=False),
),
required_if=[
("state", "user_present", ["username"]),
("state", "user_absent", ["username"]),
("state", "acl_present", ["acl_user", "acl_access"]),
("state", "acl_absent", ["acl_user"]),
],
mutually_exclusive=[
("user_password", "user_password_hash"),
],
)
def __init_module__(self):
self.runner = kopia_runner(
self.module,
extra_formats=dict(
list_users=cmd_runner_fmt.as_fixed("server", "users", "list"),
server_group=cmd_runner_fmt.as_list(),
server_subcommand=cmd_runner_fmt.as_list(),
username=cmd_runner_fmt.as_list(),
user_password=cmd_runner_fmt.as_opt_val("--user-password"),
user_password_hash=cmd_runner_fmt.as_opt_val("--user-password-hash"),
acl_user=cmd_runner_fmt.as_opt_val("--user"),
acl_access=cmd_runner_fmt.as_opt_val("--access"),
acl_target=cmd_runner_fmt.as_opt_val("--target"),
acl_no_overwrite=cmd_runner_fmt.as_bool("--no-overwrite"),
),
)
self.vars.set("previous_value", self._get()["out"])
self.vars.set("value", self.vars.previous_value, change=True, diff=True)
def __quit_module__(self):
self.vars.set("value", self._get()["out"])
def _get(self):
with self.runner("list_users 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() if out else ""
return None if out == "" else out
return process
def _run_server_cmd(self, args_order, ignore_err_msg="", check_mode_skip=True, **run_kwargs):
group, subcommand = _STATE_CLI_MAP[self.vars.state]
with self.runner(
args_order,
output_process=self._process_command_output(True, ignore_err_msg),
check_mode_skip=check_mode_skip,
) as ctx:
ctx.run(cli_action="server", server_group=group, server_subcommand=subcommand, **run_kwargs)
def state_user_present(self):
self._run_server_cmd(
"cli_action server_group server_subcommand username user_password user_password_hash config",
ignore_err_msg="already exists",
)
def state_user_absent(self):
self._run_server_cmd(
"cli_action server_group server_subcommand username config",
ignore_err_msg="no such user",
)
def state_users_listed(self):
self._run_server_cmd(
"cli_action server_group server_subcommand config",
check_mode_skip=False,
)
def state_acl_present(self):
self._run_server_cmd(
"cli_action server_group server_subcommand acl_user acl_access acl_target acl_no_overwrite config",
)
def state_acl_absent(self):
self._run_server_cmd(
"cli_action server_group server_subcommand acl_user config",
ignore_err_msg="no such rule",
)
def state_acl_listed(self):
self._run_server_cmd(
"cli_action server_group server_subcommand config",
check_mode_skip=False,
)
def state_acl_enabled(self):
self._run_server_cmd(
"cli_action server_group server_subcommand config",
)
def main():
KopiaServer.execute()
if __name__ == "__main__":
main()

299
plugins/modules/kopia_snapshot.py Normal file
View file

@ -0,0 +1,299 @@
#!/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_snapshot
short_description: Manage Kopia snapshots
author:
- Dexter Le (@munchtoast)
version_added: "13.1.0"
description:
- Manage Kopia snapshots using the Kopia CLI.
- Supports creating, deleting, and verifying snapshots, as well as listing and expiring them.
extends_documentation_fragment:
- community.general._attributes
- community.general._kopia
attributes:
check_mode:
support: full
diff_mode:
support: none
options:
state:
description:
- Desired state of the Kopia snapshot.
- V(created) takes a new snapshot of O(source).
- V(deleted) deletes the snapshot identified by O(snapshot_id).
- V(expired) applies retention policy and deletes snapshots that no longer meet it.
This is a dry-run unless O(delete=true).
- V(listed) lists snapshots for O(source) or all sources when O(source) is not set.
- V(verified) verifies the integrity of snapshots identified by O(snapshot_id),
or all snapshots when O(snapshot_id) is not set.
type: str
choices: [created, deleted, expired, listed, verified]
default: created
source:
description:
- Path of the source directory or file to snapshot.
- Required if O(state=created).
- When O(state=listed) or O(state=expired), limits the operation to this source path.
type: str
snapshot_id:
description:
- One or more snapshot manifest IDs to operate on.
- Required if O(state=deleted).
- Optional if O(state=verified); verifies all snapshots when omitted.
type: list
elements: str
description:
description:
- Free-form description to attach to the snapshot.
- Optional if O(state=created).
type: str
tags:
description:
- List of tags to attach to or filter snapshots by, in C(key:value) format.
- Optional if O(state=created) or O(state=listed).
type: list
elements: str
all_sources:
description:
- When V(true), operate on all snapshot sources rather than only the current user and host.
- Optional if O(state=listed) or O(state=expired).
type: bool
default: false
delete:
description:
- When V(true), actually delete snapshots that have expired according to the retention policy.
- When V(false) the expiration is a dry-run and no snapshots are removed.
- Required to be V(true) to perform deletion when O(state=expired).
- Required to be V(true) to confirm deletion when O(state=deleted).
type: bool
default: false
parallel:
description:
- Number of parallel upload or verification workers.
- Optional if O(state=created) or O(state=verified).
type: int
fail_fast:
description:
- When V(true), abort the snapshot on the first error encountered.
- Optional if O(state=created).
type: bool
default: false
ignore_identical:
description:
- When V(true), skip saving a new snapshot if the contents are identical to the previous one.
- Optional if O(state=created).
type: bool
default: false
verify_files_percent:
description:
- Randomly verify this percentage of files during verification.
- Value must be between C(0.0) and C(100.0).
- Optional if O(state=verified).
type: float
"""
EXAMPLES = r"""
- name: Create a snapshot of /home/user
community.general.kopia_snapshot:
state: created
source: /home/user
password: secret
config: /etc/kopia/root.config
- name: Create a tagged snapshot with a description
community.general.kopia_snapshot:
state: created
source: /var/www
password: secret
description: "pre-deploy backup"
tags:
- env:production
- app:web
- name: List all snapshots for a source
community.general.kopia_snapshot:
state: listed
source: /home/user
config: /etc/kopia/root.config
- name: List snapshots across all users and hosts
community.general.kopia_snapshot:
state: listed
all_sources: true
config: /etc/kopia/root.config
- name: Delete a specific snapshot
community.general.kopia_snapshot:
state: deleted
snapshot_id:
- abc1234def5678
delete: true
config: /etc/kopia/root.config
- name: Expire snapshots (dry run)
community.general.kopia_snapshot:
state: expired
source: /home/user
config: /etc/kopia/root.config
- name: Expire snapshots (apply deletion)
community.general.kopia_snapshot:
state: expired
source: /home/user
delete: true
config: /etc/kopia/root.config
- name: Verify all snapshots
community.general.kopia_snapshot:
state: verified
password: secret
config: /etc/kopia/root.config
- name: Verify specific snapshots and check 10 percent of files
community.general.kopia_snapshot:
state: verified
snapshot_id:
- abc1234def5678
- def5678abc1234
verify_files_percent: 10.0
config: /etc/kopia/root.config
"""
RETURN = r"""
kopia_snapshot:
description: Output from the Kopia snapshot command.
type: str
sample: ""
returned: always
"""
from ansible_collections.community.general.plugins.module_utils._cmd_runner import cmd_runner_fmt
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 KopiaSnapshot(StateModuleHelper):
module = dict(
supports_check_mode=True,
argument_spec=dict(
**KOPIA_COMMON_ARGUMENT_SPEC,
state=dict(
type="str",
default="created",
choices=["created", "deleted", "expired", "listed", "verified"],
),
source=dict(type="str"),
snapshot_id=dict(type="list", elements="str"),
description=dict(type="str"),
tags=dict(type="list", elements="str"),
all_sources=dict(type="bool", default=False),
delete=dict(type="bool", default=False),
parallel=dict(type="int"),
fail_fast=dict(type="bool", default=False),
ignore_identical=dict(type="bool", default=False),
verify_files_percent=dict(type="float"),
),
required_if=[
("state", "created", ["source"]),
("state", "deleted", ["snapshot_id"]),
],
)
def __init_module__(self):
self.runner = kopia_runner(
self.module,
extra_formats=dict(
list_snapshots=cmd_runner_fmt.as_fixed("snapshot", "list"),
source=cmd_runner_fmt.as_list(),
snapshot_id=cmd_runner_fmt.as_list(),
description=cmd_runner_fmt.as_opt_eq_val("--description"),
tags=cmd_runner_fmt.as_func(lambda v: [x for tag in v for x in ("--tags", tag)]),
all_sources=cmd_runner_fmt.as_bool("--all"),
delete=cmd_runner_fmt.as_bool("--delete"),
parallel=cmd_runner_fmt.as_opt_eq_val("--parallel"),
fail_fast=cmd_runner_fmt.as_bool("--fail-fast"),
ignore_identical=cmd_runner_fmt.as_bool("--ignore-identical-snapshots"),
verify_files_percent=cmd_runner_fmt.as_opt_eq_val("--verify-files-percent"),
),
)
self.vars.set("previous_value", self._get()["out"])
self.vars.set("value", self.vars.previous_value, change=True, diff=True)
def __quit_module__(self):
self.vars.set("value", self._get()["out"])
def _get(self):
with self.runner("list_snapshots 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() if out else ""
return None if out == "" else out
return process
def state_created(self):
with self.runner(
"cli_action state source description tags parallel fail_fast ignore_identical password config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="snapshot")
def state_deleted(self):
with self.runner(
"cli_action state snapshot_id delete config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="snapshot")
def state_expired(self):
with self.runner(
"cli_action state source all_sources delete config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="snapshot")
def state_listed(self):
with self.runner(
"cli_action state source all_sources tags config",
output_process=self._process_command_output(True),
) as ctx:
ctx.run(cli_action="snapshot")
def state_verified(self):
with self.runner(
"cli_action state snapshot_id parallel verify_files_percent password config",
output_process=self._process_command_output(True),
check_mode_skip=True,
) as ctx:
ctx.run(cli_action="snapshot")
def main():
KopiaSnapshot.execute()
if __name__ == "__main__":
main()

219
tests/unit/plugins/module_utils/test__kopia.py
View file

@ -8,7 +8,7 @@ import pytest
from ansible_collections.community.general.plugins.module_utils._kopia import (
KOPIA_COMMON_ARGUMENT_SPEC,
REPOSITORY_STATE_MAP,
STATE_MAP,
fmt_backend,
)
@ -34,22 +34,35 @@ def test_common_argument_spec_only_two_keys():
# ---------------------------------------------------------------------------
# REPOSITORY_STATE_MAP
# 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"
def test_state_map_repository_entries():
assert STATE_MAP["created"] == "create"
assert STATE_MAP["connected"] == "connect"
assert STATE_MAP["disconnected"] == "disconnect"
assert STATE_MAP["synced"] == "sync-to"
assert STATE_MAP["throttled"] == "throttle"
def test_state_map_snapshot_entries():
assert STATE_MAP["deleted"] == "delete"
assert STATE_MAP["expired"] == "expire"
assert STATE_MAP["listed"] == "list"
assert STATE_MAP["verified"] == "verify"
def test_state_map_policy_entries():
assert STATE_MAP["set"] == "set"
assert STATE_MAP["shown"] == "show"
# ---------------------------------------------------------------------------
# fmt_backend
# ---------------------------------------------------------------------------
TC_FMT_BACKEND = dict(
server=(
{"provider": "server"},
@ -113,29 +126,132 @@ TC_FMT_BACKEND = dict(
{"provider": "azure", "container": "my-container", "storage_account": "myaccount"},
["azure", "--container=my-container", "--storage-account=myaccount"],
),
azure_service_principal=(
{
"provider": "azure",
"container": "my-container",
"storage_account": "myaccount",
"client_id": "cid",
"client_secret": "csecret",
"tenant_id": "tid",
},
[
"azure",
"--container=my-container",
"--storage-account=myaccount",
"--client-id=cid",
"--client-secret=csecret",
"--tenant-id=tid",
],
),
azure_federated_token=(
{
"provider": "azure",
"container": "my-container",
"storage_account": "myaccount",
"azure_federated_token_file": "/var/run/secrets/azure/token",
},
[
"azure",
"--container=my-container",
"--storage-account=myaccount",
"--azure-federated-token-file=/var/run/secrets/azure/token",
],
),
gcs_full=(
{"provider": "gcs", "bucket": "my-bucket", "credentials_file": "/etc/gcs.json", "prefix": "kopia/"},
{
"provider": "gcs",
"bucket": "my-bucket",
"credentials_file": "/etc/gcs.json",
"prefix": "kopia/",
},
["gcs", "--bucket=my-bucket", "--credentials-file=/etc/gcs.json", "--prefix=kopia/"],
),
gcs_embed_credentials=(
{
"provider": "gcs",
"bucket": "my-bucket",
"embed_credentials": True,
},
["gcs", "--bucket=my-bucket", "--embed-credentials"],
),
gcs_embed_credentials_false=(
{
"provider": "gcs",
"bucket": "my-bucket",
"embed_credentials": False,
},
["gcs", "--bucket=my-bucket"],
),
gcs_read_only=(
{
"provider": "gcs",
"bucket": "my-bucket",
"read_only": True,
},
["gcs", "--bucket=my-bucket", "--read-only"],
),
gdrive=(
{"provider": "gdrive", "folder_id": "abc123", "credentials_file": "/etc/gdrive.json"},
["gdrive", "--folder-id=abc123", "--credentials-file=/etc/gdrive.json"],
),
gdrive_read_only=(
{"provider": "gdrive", "folder_id": "abc123", "read_only": True},
["gdrive", "--folder-id=abc123", "--read-only"],
),
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=(
rclone_minimal=(
{"provider": "rclone", "path": "remote:backup"},
["rclone", "--remote-path=remote:backup"],
),
rclone_with_exe=(
{"provider": "rclone", "path": "remote:backup", "rclone_exe": "/usr/local/bin/rclone"},
["rclone", "--remote-path=remote:backup", "--rclone-exe=/usr/local/bin/rclone"],
),
rclone_with_args=(
{
"provider": "rclone",
"path": "remote:backup",
"rclone_args": ["--transfers=4", "--checkers=8"],
},
[
"rclone",
"--remote-path=remote:backup",
"--rclone-args",
"--transfers=4",
"--rclone-args",
"--checkers=8",
],
),
rclone_with_env=(
{
"provider": "rclone",
"path": "remote:backup",
"rclone_env": ["RCLONE_CONFIG=/etc/rclone.conf", "HOME=/root"],
},
[
"rclone",
"--remote-path=remote:backup",
"--rclone-env",
"RCLONE_CONFIG=/etc/rclone.conf",
"--rclone-env",
"HOME=/root",
],
),
rclone_embed_config=(
{"provider": "rclone", "path": "remote:backup", "embed_rclone_config": True},
["rclone", "--remote-path=remote:backup", "--embed-rclone-config"],
),
sftp_full=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"port": "22",
"port": 22,
"keyfile": "/root/.ssh/id_rsa",
"known_hosts": "/root/.ssh/known_hosts",
},
@ -149,6 +265,79 @@ TC_FMT_BACKEND = dict(
"--known-hosts=/root/.ssh/known_hosts",
],
),
sftp_password=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"sftp_password": "s3cr3t",
},
[
"sftp",
"--path=/backup",
"--host=sftp.example.com",
"--username=admin",
"--sftp-password=s3cr3t",
],
),
sftp_key_data=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"key_data": "-----BEGIN RSA PRIVATE KEY-----\n...",
"known_hosts_data": "sftp.example.com ssh-rsa AAAA...",
},
[
"sftp",
"--path=/backup",
"--host=sftp.example.com",
"--username=admin",
"--key-data=-----BEGIN RSA PRIVATE KEY-----\n...",
"--known-hosts-data=sftp.example.com ssh-rsa AAAA...",
],
),
sftp_embed_credentials=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"embed_credentials": True,
},
[
"sftp",
"--path=/backup",
"--host=sftp.example.com",
"--username=admin",
"--embed-credentials",
],
),
sftp_external=(
{
"provider": "sftp",
"path": "/backup",
"host": "sftp.example.com",
"username": "admin",
"external": True,
"ssh_command": "/usr/bin/ssh",
"ssh_args": ["-o", "StrictHostKeyChecking=no"],
},
[
"sftp",
"--path=/backup",
"--host=sftp.example.com",
"--username=admin",
"--external",
"--ssh-command=/usr/bin/ssh",
"--ssh-args",
"-o",
"--ssh-args",
"StrictHostKeyChecking=no",
],
),
webdav_full=(
{
"provider": "webdav",
@ -162,6 +351,14 @@ TC_FMT_BACKEND = dict(
{"provider": "s3", "bucket": "b", "access_key": None, "secret_access_key": None},
["s3", "--bucket=b"],
),
bool_false_skipped=(
{"provider": "gcs", "bucket": "b", "embed_credentials": False, "read_only": False},
["gcs", "--bucket=b"],
),
empty_list_skipped=(
{"provider": "rclone", "path": "remote:b", "rclone_args": []},
["rclone", "--remote-path=remote:b"],
),
)
TC_FMT_BACKEND_IDS = sorted(TC_FMT_BACKEND.keys())

11
tests/unit/plugins/modules/test_kopia_notification.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_notification
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_notification, __name__, mocks=[RunCommandMock])

455
tests/unit/plugins/modules/test_kopia_notification.yaml Normal file
View file

@ -0,0 +1,455 @@
# 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}
profiles_list_ok: &profiles-list-ok
command: [/testbin/kopia, notification, profile, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: "ops-email"
err: ''
profiles_list_empty: &profiles-list-empty
command: [/testbin/kopia, notification, profile, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: ''
err: ''
test_cases:
- id: configure_email_profile
input:
state: profile_email
profile_name: ops-email
smtp_server: smtp.example.com
smtp_port: 587
smtp_username: notify@example.com
smtp_password: smtpsecret
mail_from: notify@example.com
mail_to: ops@example.com
min_severity: warning
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- configure
- email
- --profile-name
- ops-email
- --smtp-server
- smtp.example.com
- --smtp-port
- "587"
- --smtp-username
- notify@example.com
- --smtp-password
- smtpsecret
- --mail-from
- notify@example.com
- --mail-to
- ops@example.com
- --min-severity
- warning
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: configure_pushover_profile
input:
state: profile_pushover
profile_name: ops-pushover
pushover_app_token: "aToken123"
pushover_user_key: "uKey456"
min_severity: error
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- configure
- pushover
- --profile-name
- ops-pushover
- --app-token
- aToken123
- --user-key
- uKey456
- --min-severity
- error
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: configure_webhook_profile
input:
state: profile_webhook
profile_name: ops-webhook
webhook_endpoint: https://hooks.example.com/kopia
webhook_method: POST
webhook_headers:
- "Authorization:Bearer mytoken"
format: html
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- configure
- webhook
- --profile-name
- ops-webhook
- --endpoint
- https://hooks.example.com/kopia
- --method
- POST
- --http-header
- "Authorization:Bearer mytoken"
- --format
- html
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: delete_profile
input:
state: profile_deleted
profile_name: ops-email
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- profile
- delete
- --profile-name
- ops-email
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-empty
- id: delete_profile_no_such_profile
input:
state: profile_deleted
profile_name: ops-email
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- delete
- --profile-name
- ops-email
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'no such profile'
- *profiles-list-empty
- id: list_profiles
input:
state: profiles_listed
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- profile
- list
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "ops-email"
err: ''
- *profiles-list-ok
- id: show_profile
input:
state: profile_shown
profile_name: ops-email
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- profile
- show
- --profile-name
- ops-email
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Profile: ops-email"
err: ''
- *profiles-list-ok
- id: set_template
input:
state: template_set
template_name: snapshot-complete
template_file: /etc/kopia/templates/snapshot-complete.html
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- template
- set
- snapshot-complete
- /etc/kopia/templates/snapshot-complete.html
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: list_templates
input:
state: templates_listed
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- template
- list
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "snapshot-complete"
err: ''
- *profiles-list-ok
- id: remove_template
input:
state: template_removed
template_name: snapshot-complete
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- template
- remove
- snapshot-complete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: test_profile
input:
state: profile_tested
profile_name: ops-email
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- profile
- test
- --profile-name
- ops-email
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: show_template
input:
state: template_shown
template_name: snapshot-complete
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- template
- show
- snapshot-complete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Subject: Snapshot complete"
err: ''
- *profiles-list-ok
- id: remove_template_no_such_template
input:
state: template_removed
template_name: snapshot-complete
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *profiles-list-ok
- command:
- /testbin/kopia
- notification
- template
- remove
- snapshot-complete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'no such template'
- *profiles-list-ok
- id: configure_email_with_cc_and_identity
input:
state: profile_email
profile_name: ops-email
smtp_server: smtp.example.com
smtp_port: 587
smtp_username: notify@example.com
smtp_password: smtpsecret
smtp_identity: "notify"
mail_from: notify@example.com
mail_to: ops@example.com
mail_cc: manager@example.com
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- configure
- email
- --profile-name
- ops-email
- --smtp-server
- smtp.example.com
- --smtp-port
- "587"
- --smtp-username
- notify@example.com
- --smtp-password
- smtpsecret
- --smtp-identity
- notify
- --mail-from
- notify@example.com
- --mail-to
- ops@example.com
- --mail-cc
- manager@example.com
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok
- id: configure_webhook_with_send_test
input:
state: profile_webhook
profile_name: ops-webhook
webhook_endpoint: https://hooks.example.com/kopia
webhook_method: POST
format: md
send_test: true
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *profiles-list-empty
- command:
- /testbin/kopia
- notification
- profile
- configure
- webhook
- --profile-name
- ops-webhook
- --endpoint
- https://hooks.example.com/kopia
- --method
- POST
- --format
- md
- --send-test-notification
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *profiles-list-ok

11
tests/unit/plugins/modules/test_kopia_policy.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_policy
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_policy, __name__, mocks=[RunCommandMock])

359
tests/unit/plugins/modules/test_kopia_policy.yaml Normal file
View file

@ -0,0 +1,359 @@
# 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}
list_ok: &list-ok
command: [/testbin/kopia, policy, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: "user@host:/home/user"
err: ''
list_empty: &list-empty
command: [/testbin/kopia, policy, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: ''
err: ''
test_cases:
- id: set_retention_policy
input:
state: set
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
retention:
keep_latest: "10"
keep_daily: "7"
keep_weekly: "4"
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/home/user
- --keep-latest
- "10"
- --keep-daily
- "7"
- --keep-weekly
- "4"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: set_global_policy_with_compression
input:
state: set
global_policy: true
config: /etc/kopia/root.config
compression: zstd
files:
add_ignore:
- "*.tmp"
- ".cache"
ignore_cache_dirs: "true"
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- --global
- --add-ignore
- "*.tmp"
- --add-ignore
- .cache
- --ignore-cache-dirs
- "true"
- --compression
- zstd
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: set_scheduling_policy
input:
state: set
target: "user@hostname:/var/www"
config: /etc/kopia/root.config
scheduling:
times:
- "02:00"
- "14:00"
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/var/www
- --snapshot-time
- "02:00,14:00"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: show_policy
input:
state: shown
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- policy
- show
- user@hostname:/home/user
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Policy for user@hostname:/home/user"
err: ''
- *list-ok
- id: list_policies
input:
state: listed
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- policy
- list
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "user@host:/home/user"
err: ''
- *list-ok
- id: delete_policy
input:
state: deleted
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- policy
- delete
- user@hostname:/home/user
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-empty
- id: delete_policy_no_such_policy
input:
state: deleted
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- delete
- user@hostname:/home/user
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'no such policy'
- *list-empty
- id: set_scheduling_interval
input:
state: set
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
scheduling:
interval: "1h"
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/home/user
- --snapshot-interval
- "1h"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: set_scheduling_manual
input:
state: set
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
scheduling:
manual: true
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/home/user
- --manual
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: set_policy_files_remove_ignore
input:
state: set
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
files:
remove_ignore:
- "*.tmp"
max_file_size: "100MB"
one_file_system: "true"
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/home/user
- --remove-ignore
- "*.tmp"
- --max-file-size
- 100MB
- --one-file-system
- "true"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: set_retention_policy_inherit
input:
state: set
target: "user@hostname:/home/user"
config: /etc/kopia/root.config
retention:
keep_daily: inherit
keep_weekly: inherit
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- policy
- set
- user@hostname:/home/user
- --keep-daily
- inherit
- --keep-weekly
- inherit
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-ok
- id: delete_global_policy
input:
state: deleted
global_policy: true
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- policy
- delete
- --global
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-empty
- id: show_global_policy
input:
state: shown
global_policy: true
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- policy
- show
- --global
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Global policy settings"
err: ''
- *list-ok

11
tests/unit/plugins/modules/test_kopia_server.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_server
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_server, __name__, mocks=[RunCommandMock])

315
tests/unit/plugins/modules/test_kopia_server.yaml Normal file
View file

@ -0,0 +1,315 @@
# 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}
users_list_ok: &users-list-ok
command: [/testbin/kopia, server, users, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: "alice@backuphost"
err: ''
users_list_empty: &users-list-empty
command: [/testbin/kopia, server, users, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: ''
err: ''
test_cases:
- id: add_user
input:
state: user_present
username: alice@backuphost
user_password: secretpassword
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *users-list-empty
- command:
- /testbin/kopia
- server
- users
- set
- alice@backuphost
- --user-password
- secretpassword
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: add_user_with_hash
input:
state: user_present
username: alice@backuphost
user_password_hash: "$2a$12$hashedvalue"
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *users-list-empty
- command:
- /testbin/kopia
- server
- users
- set
- alice@backuphost
- --user-password-hash
- "$2a$12$hashedvalue"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: remove_user
input:
state: user_absent
username: alice@backuphost
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- users
- delete
- alice@backuphost
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-empty
- id: remove_user_no_such_user
input:
state: user_absent
username: alice@backuphost
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-empty
- command:
- /testbin/kopia
- server
- users
- delete
- alice@backuphost
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'no such user'
- *users-list-empty
- id: list_users
input:
state: users_listed
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- users
- list
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "alice@backuphost"
err: ''
- *users-list-ok
- id: enable_acl
input:
state: acl_enabled
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- enable
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: add_acl_entry
input:
state: acl_present
acl_user: alice@backuphost
acl_access: FULL
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- add
- --user
- alice@backuphost
- --access
- FULL
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: add_targeted_acl_entry
input:
state: acl_present
acl_user: bob@backuphost
acl_access: READ
acl_target: "type:snapshot,username:bob"
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- add
- --user
- bob@backuphost
- --access
- READ
- --target
- "type:snapshot,username:bob"
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: delete_acl_entry
input:
state: acl_absent
acl_user: alice@backuphost
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- delete
- --user
- alice@backuphost
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: list_acl_entries
input:
state: acl_listed
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- list
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "alice@backuphost FULL"
err: ''
- *users-list-ok
- id: add_acl_entry_no_overwrite
input:
state: acl_present
acl_user: alice@backuphost
acl_access: FULL
acl_no_overwrite: true
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- add
- --user
- alice@backuphost
- --access
- FULL
- --no-overwrite
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *users-list-ok
- id: delete_acl_entry_no_such_rule
input:
state: acl_absent
acl_user: alice@backuphost
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *users-list-ok
- command:
- /testbin/kopia
- server
- acl
- delete
- --user
- alice@backuphost
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 1
out: ''
err: 'no such rule'
- *users-list-ok

11
tests/unit/plugins/modules/test_kopia_snapshot.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_snapshot
from .uthelper import RunCommandMock, UTHelper
UTHelper.from_module(kopia_snapshot, __name__, mocks=[RunCommandMock])

392
tests/unit/plugins/modules/test_kopia_snapshot.yaml Normal file
View file

@ -0,0 +1,392 @@
# 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}
list_ok: &list-ok
command: [/testbin/kopia, snapshot, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: "abc123 /home/user 2026-01-01 00:00:00 UTC"
err: ''
list_empty: &list-empty
command: [/testbin/kopia, snapshot, list, --config-file=/etc/kopia/root.config]
environ: *env-def
rc: 0
out: ''
err: ''
test_cases:
- id: create_snapshot
input:
state: created
source: /home/user
password: secret
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- snapshot
- create
- /home/user
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Created snapshot abc123"
err: ''
- *list-ok
- id: create_snapshot_with_tags
input:
state: created
source: /var/www
password: secret
config: /etc/kopia/root.config
description: "pre-deploy backup"
tags:
- env:production
- app:web
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- snapshot
- create
- /var/www
- --description=pre-deploy backup
- --tags
- env:production
- --tags
- app:web
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Created snapshot def456"
err: ''
- *list-ok
- id: list_snapshots
input:
state: listed
source: /home/user
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- list
- /home/user
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "abc123 /home/user 2026-01-01 00:00:00 UTC"
err: ''
- *list-ok
- id: list_all_sources
input:
state: listed
all_sources: true
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- list
- --all
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "abc123 /home/user 2026-01-01 00:00:00 UTC"
err: ''
- *list-ok
- id: delete_snapshot
input:
state: deleted
snapshot_id:
- abc1234def5678
delete: true
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- delete
- abc1234def5678
- --delete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-empty
- id: expire_snapshots_dry_run
input:
state: expired
source: /home/user
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- expire
- /home/user
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Would delete 2 snapshots"
err: ''
- *list-ok
- id: expire_snapshots_apply
input:
state: expired
source: /home/user
delete: true
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- expire
- /home/user
- --delete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Deleted 2 snapshots"
err: ''
- *list-ok
- id: verify_all_snapshots
input:
state: verified
password: secret
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- verify
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Verified 5 files."
err: ''
- *list-ok
- id: verify_specific_snapshot
input:
state: verified
snapshot_id:
- abc1234def5678
verify_files_percent: 10.0
config: /etc/kopia/root.config
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- verify
- abc1234def5678
- --verify-files-percent=10.0
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Verified 1 snapshot."
err: ''
- *list-ok
- id: create_snapshot_with_parallel
input:
state: created
source: /home/user
password: secret
config: /etc/kopia/root.config
parallel: 4
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- snapshot
- create
- /home/user
- --parallel=4
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Created snapshot abc123"
err: ''
- *list-ok
- id: create_snapshot_fail_fast
input:
state: created
source: /home/user
password: secret
config: /etc/kopia/root.config
fail_fast: true
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- snapshot
- create
- /home/user
- --fail-fast
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Created snapshot abc123"
err: ''
- *list-ok
- id: create_snapshot_ignore_identical
input:
state: created
source: /home/user
password: secret
config: /etc/kopia/root.config
ignore_identical: true
output:
changed: true
mocks:
run_command:
- *list-empty
- command:
- /testbin/kopia
- snapshot
- create
- /home/user
- --ignore-identical-snapshots
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Created snapshot abc123"
err: ''
- *list-ok
- id: delete_multiple_snapshots
input:
state: deleted
snapshot_id:
- abc1234def5678
- def5678abc1234
delete: true
config: /etc/kopia/root.config
output:
changed: true
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- delete
- abc1234def5678
- def5678abc1234
- --delete
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: ''
err: ''
- *list-empty
- id: verify_with_parallel
input:
state: verified
password: secret
config: /etc/kopia/root.config
parallel: 8
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- verify
- --parallel=8
- --password=secret
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "Verified 5 files."
err: ''
- *list-ok
- id: list_snapshots_with_tags
input:
state: listed
config: /etc/kopia/root.config
tags:
- env:production
output:
changed: false
mocks:
run_command:
- *list-ok
- command:
- /testbin/kopia
- snapshot
- list
- --tags
- env:production
- --config-file=/etc/kopia/root.config
environ: *env-def
rc: 0
out: "abc123 /home/user 2026-01-01 00:00:00 UTC"
err: ''
- *list-ok