mirror of
https://github.com/netinvent/npbackup.git
synced 2025-09-06 13:05:24 +08:00
tests: Add basic CLI tests
This commit is contained in:
Orsiris de Jong
parent
b3a5d56fbf
commit
120a00f8ba
4 changed files with 172 additions and 99 deletions
|
|
@ -4,9 +4,9 @@ repos:
|
|||
repo_uri: ./test
|
||||
repo_group: default_group
|
||||
backup_opts:
|
||||
paths: []
|
||||
paths: ../npbackup/npbackup
|
||||
tags: []
|
||||
minimum_backup_size_error: 80.0 Mib
|
||||
minimum_backup_size_error: 400 Kib
|
||||
exclude_files_larger_than: 0.0 Kib
|
||||
repo_opts:
|
||||
repo_password: test
|
||||
|
|
|
|||
|
|
@ -1,15 +1,18 @@
|
|||
conf_version: 3.0
|
||||
repos:
|
||||
default:
|
||||
repo_uri: ./test
|
||||
repo_uri:
|
||||
__NPBACKUP__cjA2cMpU/NXWeXGSQZrtgXVP1ZrKqQn4Hq41xxbaRTQxNzMyNzI0ODUzLjI0OTg5Nw8PDw8PDw8PDw8PDw8PDzxRwZTs6gTqdrEI6oFSnSWLureQSQ==__NPBACKUP__
|
||||
repo_group: default_group
|
||||
backup_opts:
|
||||
paths: []
|
||||
paths:
|
||||
- ../npbackup/npbackup
|
||||
tags: []
|
||||
minimum_backup_size_error: 80.0 Mib
|
||||
minimum_backup_size_error: 400 Kib
|
||||
exclude_files_larger_than: 0.0 Kib
|
||||
repo_opts:
|
||||
repo_password: test
|
||||
repo_password:
|
||||
__NPBACKUP__TfMLTaHdOBA7cUkggFqr8jQCX9BiPb+I00F7S5zQ088xNzMyNzI0ODUzLjI0OTg5Nw8PDw8PDw8PDw8PDw8PD7bNnnCJ3yvSLVsIxaxVMdWKbk0=__NPBACKUP__
|
||||
upload_speed: 100.0 Mib
|
||||
download_speed: 0.0 Kib
|
||||
retention_policy: {}
|
||||
|
|
@ -17,6 +20,7 @@ repos:
|
|||
env:
|
||||
env_variables: {}
|
||||
encrypted_env_variables: {}
|
||||
is_protected: false
|
||||
groups:
|
||||
default_group:
|
||||
backup_opts:
|
||||
|
|
@ -24,7 +28,7 @@ groups:
|
|||
source_type:
|
||||
tags: []
|
||||
compression: auto
|
||||
use_fs_snapshot: true
|
||||
use_fs_snapshot: false
|
||||
ignore_cloud_files: true
|
||||
exclude_caches: true
|
||||
one_file_system: true
|
||||
|
|
@ -71,6 +75,7 @@ groups:
|
|||
env:
|
||||
env_variables: {}
|
||||
encrypted_env_variables: {}
|
||||
is_protected: false
|
||||
identity:
|
||||
machine_id: ${HOSTNAME}__Zo6u
|
||||
machine_group:
|
||||
|
|
@ -90,3 +95,4 @@ global_options:
|
|||
auto_upgrade_server_password:
|
||||
auto_upgrade_host_identity: ${MACHINE_ID}
|
||||
auto_upgrade_group: ${MACHINE_GROUP}
|
||||
audience: private
|
||||
|
|
|
|||
|
|
@ -6,32 +6,47 @@ __intname__ = "npbackup_cli_tests"
|
|||
__author__ = "Orsiris de Jong"
|
||||
__copyright__ = "Copyright (C) 2022-2024 NetInvent"
|
||||
__license__ = "BSD-3-Clause"
|
||||
__build__ = "2024042301"
|
||||
__build__ = "2024112701"
|
||||
__compat__ = "python3.6+"
|
||||
|
||||
|
||||
"""
|
||||
Simple test where we launch the GUI and hope it doesn't die
|
||||
Simple test where we launch the CLI and hope it doesn't die
|
||||
Should be improved with much stronger tests
|
||||
Missing:
|
||||
- VSS test
|
||||
- backup minimum size tests
|
||||
- proper retention policy tests
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
from io import StringIO
|
||||
import json
|
||||
|
||||
sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
from npbackup import __main__
|
||||
from npbackup.path_helper import CURRENT_DIR, CURRENT_EXECUTABLE
|
||||
from npbackup.configuration import load_config, get_repo_config
|
||||
|
||||
|
||||
if os.name == 'nt':
|
||||
if os.name == "nt":
|
||||
CONF_FILE = "npbackup-cli-test-windows.yaml"
|
||||
else:
|
||||
CONF_FILE = "npbackup-cli-test-windows.yaml"
|
||||
CONF_FILE = os.path.join(CURRENT_DIR, CONF_FILE)
|
||||
CONF_FILE = "npbackup-cli-test-linux.yaml"
|
||||
|
||||
CONF_FILE = Path(CURRENT_DIR).absolute().joinpath(CONF_FILE)
|
||||
full_config = load_config(CONF_FILE)
|
||||
repo_config, _ = get_repo_config(full_config)
|
||||
|
||||
|
||||
class RedirectedStdout:
|
||||
"""
|
||||
Balantly copied from https://stackoverflow.com/a/45899925/2635443
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._stdout = None
|
||||
self._string_io = None
|
||||
|
|
@ -49,83 +64,137 @@ class RedirectedStdout:
|
|||
|
||||
|
||||
def test_npbackup_cli_no_config():
|
||||
sys.argv = [''] # Make sure we don't get any pytest args
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
assert 'CRITICAL :: Cannot run without configuration file' in str(logs), "There should be a critical error when config file is not given"
|
||||
|
||||
|
||||
def test_npbackup_cli_wrong_config_path():
|
||||
sys.argv = ['', '-c', 'npbackup-non-existent.conf']
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
assert 'Config file npbackup-non-existent.conf cannot be read' in str(logs), "There should be a critical error when config file is not given"
|
||||
|
||||
|
||||
def test_npbackup_cli_show_config():
|
||||
sys.argv = ['', '-c', CONF_FILE, '--show-config']
|
||||
sys.argv = [""] # Make sure we don't get any pytest args
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
assert "__(o_O)__" not in str(logs), "Obfuscation does not work"
|
||||
|
||||
|
||||
def _no_test_npbackup_cli_create_backup():
|
||||
sys.argv = ['', '-c' './npbackup-cli-test.conf', '-b']
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(logs)
|
||||
assert "CRITICAL :: Cannot run without configuration file" in str(
|
||||
logs
|
||||
), "There should be a critical error when config file is not given"
|
||||
|
||||
|
||||
def _no_test_npbackup_cli_snapshots():
|
||||
sys.argv = ['', '-c', 'npbackup-test.conf', '--snapshots']
|
||||
def test_npbackup_cli_wrong_config_path():
|
||||
sys.argv = ["", "-c", "npbackup-non-existent.conf"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
print(logs)
|
||||
print(str(logs))
|
||||
assert "Config file npbackup-non-existent.conf cannot be read" in str(
|
||||
logs
|
||||
), "There should be a critical error when config file is not given"
|
||||
|
||||
|
||||
def _no_test_npbackup_cli_restore():
|
||||
sys.argv = ['', '-c' './npbackup-cli-test.conf', '-r', './restored']
|
||||
def test_npbackup_cli_show_config():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "--show-config"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
assert "__(o_O)__" in str(logs), "Obfuscation does not work"
|
||||
|
||||
|
||||
def test_npbackup_cli_create_backup():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "-b"]
|
||||
# Make sure there is no existing repository
|
||||
|
||||
repo_uri = Path(repo_config.g("repo_uri"))
|
||||
if repo_uri.is_dir():
|
||||
shutil.rmtree(repo_uri)
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
|
||||
|
||||
def test_npbackup_cli_unlock():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "--unlock"]
|
||||
# Make sure there is no existing repository
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
assert "Repo successfully unlocked" in str(logs), "Could not unlock repo"
|
||||
|
||||
|
||||
def test_npbackup_cli_snapshots():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "--snapshots", "--json"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
__main__.main()
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
json_logs = json.loads(str(logs))
|
||||
assert json_logs["result"], "Bad snapshot result"
|
||||
assert (
|
||||
json_logs["operation"] == "snapshots"
|
||||
), "Bogus operation name for snapshots"
|
||||
assert len(json_logs["output"]) == 1, "More than one snapshot present"
|
||||
|
||||
|
||||
def test_npbackup_cli_restore():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "-r", "./restored"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(str(logs))
|
||||
assert "Successfully restored data" in str(
|
||||
logs
|
||||
), "Logs don't show successful restore"
|
||||
assert Path(
|
||||
"./restored/npbackup/npbackup/__version__.py"
|
||||
).is_file(), "Restored snapshot does not contain our data"
|
||||
|
||||
|
||||
def test_npbackup_cli_list():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "--ls", "latest", "--json"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(logs)
|
||||
json_logs = json.loads(str(logs))
|
||||
assert json_logs["result"], "Bad ls result"
|
||||
assert json_logs["operation"] == "ls", "Bogus operation name for ls"
|
||||
assert "/npbackup/npbackup/gui/__main__.py" in str(
|
||||
logs
|
||||
), "Missing main gui in list"
|
||||
|
||||
|
||||
def _no_test_npbackup_cli_list():
|
||||
sys.argv = ['', '-c' './npbackup-cli-test.conf', '--ls snapshots']
|
||||
def test_npbackup_cli_retention():
|
||||
sys.argv = ["", "-c", str(CONF_FILE), "--policy"]
|
||||
try:
|
||||
with RedirectedStdout() as logs:
|
||||
e = __main__.main()
|
||||
print(e)
|
||||
except SystemExit:
|
||||
print(logs)
|
||||
assert "Successfully applied retention policy" in str(
|
||||
logs
|
||||
), "Failed applying retention policy"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_npbackup_cli_no_config()
|
||||
test_npbackup_cli_wrong_config_path()
|
||||
test_npbackup_cli_show_config()
|
||||
# TODO
|
||||
#test_npbackup_cli_create_backup()
|
||||
#test_npbackup_cli_snapshots()
|
||||
#test_npbackup_cli_restore()
|
||||
#test_npbackup_cli_list()
|
||||
test_npbackup_cli_create_backup()
|
||||
test_npbackup_cli_unlock()
|
||||
test_npbackup_cli_snapshots()
|
||||
test_npbackup_cli_restore()
|
||||
test_npbackup_cli_list()
|
||||
# This one should is pretty hard to test without having repo with multiple different date snapshots
|
||||
# We need to create a "fake" repo starting in let's say 2020 and put our date back to 2023 to test our standard
|
||||
# policy
|
||||
# We can also have a forget test which should fail because of bogus permissions
|
||||
#test_npbackup_cli_forget()
|
||||
test_npbackup_cli_retention()
|
||||
|
|
|
|||
|
|
@ -7,7 +7,9 @@ __author__ = "Orsiris de Jong"
|
|||
__copyright__ = "Copyright (C) 2022-2024 NetInvent"
|
||||
__license__ = "BSD-3-Clause"
|
||||
__build__ = "2024010101"
|
||||
__description__ = "Converts restic command line output to a text file node_exporter can scrape"
|
||||
__description__ = (
|
||||
"Converts restic command line output to a text file node_exporter can scrape"
|
||||
)
|
||||
__compat__ = "python3.6+"
|
||||
|
||||
import sys
|
||||
|
|
@ -19,6 +21,7 @@ import json
|
|||
import tempfile
|
||||
from ofunctions.platform import os_arch
|
||||
from command_runner import command_runner
|
||||
|
||||
try:
|
||||
from npbackup.restic_metrics import *
|
||||
except ImportError: # would be ModuleNotFoundError in Python 3+
|
||||
|
|
@ -28,14 +31,16 @@ except ImportError: # would be ModuleNotFoundError in Python 3+
|
|||
from npbackup.core.restic_source_binary import get_restic_internal_binary
|
||||
|
||||
restic_json_outputs = {}
|
||||
restic_json_outputs["v0.16.2"] = \
|
||||
"""{"message_type":"summary","files_new":5,"files_changed":15,"files_unmodified":6058,"dirs_new":0,"dirs_changed":27,"dirs_unmodified":866,"data_blobs":17,"tree_blobs":28,"data_added":281097,"total_files_processed":6078,"total_bytes_processed":122342158,"total_duration":1.2836983,"snapshot_id":"360333437921660a5228a9c1b65a2d97381f0bc135499c6e851acb0ab84b0b0a"}
|
||||
restic_json_outputs[
|
||||
"v0.16.2"
|
||||
] = """{"message_type":"summary","files_new":5,"files_changed":15,"files_unmodified":6058,"dirs_new":0,"dirs_changed":27,"dirs_unmodified":866,"data_blobs":17,"tree_blobs":28,"data_added":281097,"total_files_processed":6078,"total_bytes_processed":122342158,"total_duration":1.2836983,"snapshot_id":"360333437921660a5228a9c1b65a2d97381f0bc135499c6e851acb0ab84b0b0a"}
|
||||
"""
|
||||
|
||||
restic_str_outputs = {}
|
||||
# log file from restic v0.16.2
|
||||
restic_str_outputs["v0.16.2"] = \
|
||||
"""repository 962d5924 opened (version 2, compression level auto)
|
||||
restic_str_outputs[
|
||||
"v0.16.2"
|
||||
] = """repository 962d5924 opened (version 2, compression level auto)
|
||||
using parent snapshot 325a2fa1
|
||||
[0:00] 100.00% 4 / 4 index files loaded
|
||||
|
||||
|
|
@ -47,9 +52,10 @@ processed 6073 files, 116.657 MiB in 0:03
|
|||
snapshot b28b0901 saved
|
||||
"""
|
||||
|
||||
# log file from restic v0.14.0
|
||||
restic_str_outputs["v0.14.0"] = \
|
||||
"""using parent snapshot df60db01
|
||||
# log file from restic v0.14.0
|
||||
restic_str_outputs[
|
||||
"v0.14.0"
|
||||
] = """using parent snapshot df60db01
|
||||
|
||||
Files: 1584 new, 269 changed, 235933 unmodified
|
||||
Dirs: 258 new, 714 changed, 37066 unmodified
|
||||
|
|
@ -58,9 +64,10 @@ Added to the repo: 493.649 MiB
|
|||
processed 237786 files, 85.487 GiB in 11:12"
|
||||
"""
|
||||
|
||||
# log file form restic v0.9.4
|
||||
restic_str_outputs["v0.9.4"] = \
|
||||
"""
|
||||
# log file form restic v0.9.4
|
||||
restic_str_outputs[
|
||||
"v0.9.4"
|
||||
] = """
|
||||
Files: 9 new, 32 changed, 110340 unmodified
|
||||
Dirs: 0 new, 2 changed, 0 unmodified
|
||||
Added to the repo: 196.568 MiB
|
||||
|
|
@ -106,84 +113,74 @@ def running_on_github_actions():
|
|||
def test_restic_str_output_2_metrics():
|
||||
instance = "test"
|
||||
backup_job = "some_nas"
|
||||
labels = "instance=\"{}\",backup_job=\"{}\"".format(instance, backup_job)
|
||||
labels = 'instance="{}",backup_job="{}"'.format(instance, backup_job)
|
||||
for version, output in restic_str_outputs.items():
|
||||
print(f"Testing V1 parser restic str output from version {version}")
|
||||
errors, prom_metrics = restic_output_2_metrics(True, output, labels)
|
||||
assert errors is False
|
||||
#print(f"Parsed result:\n{prom_metrics}")
|
||||
# print(f"Parsed result:\n{prom_metrics}")
|
||||
for expected_result in expected_results_V1:
|
||||
match_found = False
|
||||
#print("Searching for {}".format(expected_result))
|
||||
# print("Searching for {}".format(expected_result))
|
||||
for metric in prom_metrics:
|
||||
result = re.match(expected_result, metric)
|
||||
if result:
|
||||
match_found = True
|
||||
break
|
||||
assert match_found is True, 'No match found for {}'.format(expected_result)
|
||||
assert match_found is True, "No match found for {}".format(expected_result)
|
||||
|
||||
|
||||
def test_restic_str_output_to_json():
|
||||
labels = {
|
||||
"instance": "test",
|
||||
"backup_job": "some_nas"
|
||||
}
|
||||
labels = {"instance": "test", "backup_job": "some_nas"}
|
||||
for version, output in restic_str_outputs.items():
|
||||
print(f"Testing V2 parser restic str output from version {version}")
|
||||
json_metrics = restic_str_output_to_json(True, output)
|
||||
assert json_metrics["errors"] == False
|
||||
#print(json_metrics)
|
||||
# print(json_metrics)
|
||||
_, prom_metrics, _ = restic_json_to_prometheus(True, json_metrics, labels)
|
||||
|
||||
#print(f"Parsed result:\n{prom_metrics}")
|
||||
# print(f"Parsed result:\n{prom_metrics}")
|
||||
for expected_result in expected_results_V2:
|
||||
match_found = False
|
||||
#print("Searching for {}".format(expected_result))
|
||||
# print("Searching for {}".format(expected_result))
|
||||
for metric in prom_metrics:
|
||||
result = re.match(expected_result, metric)
|
||||
if result:
|
||||
match_found = True
|
||||
break
|
||||
assert match_found is True, 'No match found for {}'.format(expected_result)
|
||||
assert match_found is True, "No match found for {}".format(expected_result)
|
||||
|
||||
|
||||
def test_restic_json_output():
|
||||
labels = {
|
||||
"instance": "test",
|
||||
"backup_job": "some_nas"
|
||||
}
|
||||
labels = {"instance": "test", "backup_job": "some_nas"}
|
||||
for version, json_output in restic_json_outputs.items():
|
||||
print(f"Testing V2 direct restic --json output from version {version}")
|
||||
restic_json = json.loads(json_output)
|
||||
_, prom_metrics, _ = restic_json_to_prometheus(True, restic_json, labels)
|
||||
#print(f"Parsed result:\n{prom_metrics}")
|
||||
# print(f"Parsed result:\n{prom_metrics}")
|
||||
for expected_result in expected_results_V2:
|
||||
match_found = False
|
||||
#print("Searching for {}".format(expected_result))
|
||||
# print("Searching for {}".format(expected_result))
|
||||
for metric in prom_metrics:
|
||||
result = re.match(expected_result, metric)
|
||||
if result:
|
||||
match_found = True
|
||||
break
|
||||
assert match_found is True, 'No match found for {}'.format(expected_result)
|
||||
assert match_found is True, "No match found for {}".format(expected_result)
|
||||
|
||||
|
||||
def test_real_restic_output():
|
||||
# Don't do the real tests on github actions, since we don't have
|
||||
# Don't do the real tests on github actions, since we don't have
|
||||
# the binaries there.
|
||||
# TODO: Add download/unzip restic binaries so we can run these tests
|
||||
if running_on_github_actions():
|
||||
return
|
||||
labels = {
|
||||
"instance": "test",
|
||||
"backup_job": "some_nas"
|
||||
}
|
||||
labels = {"instance": "test", "backup_job": "some_nas"}
|
||||
restic_binary = get_restic_internal_binary(os_arch())
|
||||
print(f"Testing real restic output, Running with restic {restic_binary}")
|
||||
assert restic_binary is not None, "No restic binary found"
|
||||
|
||||
for api_arg in ['', ' --json']:
|
||||
|
||||
for api_arg in ["", " --json"]:
|
||||
# Setup repo and run a quick backup
|
||||
repo_path = Path(tempfile.gettempdir()) / "repo"
|
||||
if repo_path.is_dir():
|
||||
|
|
@ -193,8 +190,9 @@ def test_real_restic_output():
|
|||
os.environ["RESTIC_REPOSITORY"] = str(repo_path)
|
||||
os.environ["RESTIC_PASSWORD"] = "TEST"
|
||||
|
||||
|
||||
exit_code, output = command_runner(f"{restic_binary} init --repository-version 2", live_output=True)
|
||||
exit_code, output = command_runner(
|
||||
f"{restic_binary} init --repository-version 2", live_output=True
|
||||
)
|
||||
# Just backend current directory
|
||||
cmd = f"{restic_binary} backup {api_arg} ."
|
||||
exit_code, output = command_runner(cmd, timeout=120, live_output=True)
|
||||
|
|
@ -204,7 +202,7 @@ def test_real_restic_output():
|
|||
else:
|
||||
restic_json = output
|
||||
_, prom_metrics, _ = restic_json_to_prometheus(True, restic_json, labels)
|
||||
#print(f"Parsed result:\n{prom_metrics}")
|
||||
# print(f"Parsed result:\n{prom_metrics}")
|
||||
for expected_result in expected_results_V2:
|
||||
match_found = False
|
||||
print("Searching for {}".format(expected_result))
|
||||
|
|
@ -213,11 +211,11 @@ def test_real_restic_output():
|
|||
if result:
|
||||
match_found = True
|
||||
break
|
||||
assert match_found is True, 'No match found for {}'.format(expected_result)
|
||||
|
||||
assert match_found is True, "No match found for {}".format(expected_result)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_restic_str_output_2_metrics()
|
||||
test_restic_str_output_to_json()
|
||||
test_restic_json_output()
|
||||
test_real_restic_output()
|
||||
test_real_restic_output()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue