301 lines
9.8 KiB
Python
301 lines
9.8 KiB
Python
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from random import Random
|
|
from tempfile import TemporaryDirectory
|
|
from unittest import TestCase
|
|
|
|
from helpers import (
|
|
DirectoryDescriptor,
|
|
generate_random_dir,
|
|
randomly_mutate_directory_in_descriptor,
|
|
scan_dir,
|
|
)
|
|
from helpers.datman_helpers import (
|
|
MULTI_PILES_SECTION,
|
|
filter_descriptor_by_label,
|
|
generate_labels,
|
|
get_hostname,
|
|
save_labelling_rules,
|
|
set_up_simple_datman,
|
|
)
|
|
from helpers.yama_helpers import set_up_simple_yama
|
|
|
|
|
|
class TestBackupAndExtract(TestCase):
|
|
def test_backup_one_full_no_labelling_skip_metadata(self):
|
|
td = TemporaryDirectory("test_backup_one_full")
|
|
tdpath = Path(td.name)
|
|
|
|
datman_path = tdpath.joinpath("datman")
|
|
src_path = datman_path.joinpath("srca")
|
|
yama_path = datman_path.joinpath("main")
|
|
|
|
set_up_simple_datman(datman_path)
|
|
set_up_simple_yama(yama_path)
|
|
|
|
rng = Random()
|
|
seed = rng.randint(0, 9001)
|
|
print(f"seed: {seed}")
|
|
rng.seed(seed)
|
|
later_expected_descriptor, _ = generate_random_dir(rng, src_path, 32)
|
|
|
|
print("storing")
|
|
subprocess.check_call(("datman", "backup-one", "srca", "main"), cwd=datman_path)
|
|
|
|
print("extracting")
|
|
dest_path = tdpath.joinpath("desta")
|
|
subprocess.check_call(
|
|
(
|
|
"datman",
|
|
"extract",
|
|
"--skip-metadata",
|
|
"--accept-partial",
|
|
"main",
|
|
"../desta",
|
|
),
|
|
cwd=datman_path,
|
|
)
|
|
|
|
# this will be wrapped in a directory that starts with the name srca+
|
|
extracted_dir_descriptor_wrapper = scan_dir(dest_path)
|
|
|
|
contents = extracted_dir_descriptor_wrapper.contents
|
|
self.assertEqual(len(contents), 1)
|
|
key, value = next(iter(contents.items()))
|
|
self.assertTrue(key.startswith("srca+"))
|
|
|
|
self.assertIsInstance(value, DirectoryDescriptor)
|
|
key, value = next(iter(value.contents.items()))
|
|
self.assertEqual(key, "srca")
|
|
|
|
self.assertEqual(
|
|
value.ignore_metadata(), later_expected_descriptor.ignore_metadata()
|
|
)
|
|
|
|
td.cleanup()
|
|
|
|
def test_backup_failure_is_loud(self):
|
|
"""
|
|
Tests that backup failure is noticeable.
|
|
"""
|
|
|
|
td = TemporaryDirectory("test_backup_failure_is_loud")
|
|
tdpath = Path(td.name)
|
|
|
|
datman_path = tdpath.joinpath("datman")
|
|
src_path = datman_path.joinpath("srca")
|
|
yama_path = datman_path.joinpath("main")
|
|
|
|
set_up_simple_datman(
|
|
datman_path,
|
|
f"""
|
|
[sources.srcimpossible]
|
|
directory = "/path/to/absolutely/nowhere"
|
|
hostname = "{get_hostname()}"
|
|
|
|
[sources.srcimpossible2]
|
|
directory = "/path/to/absolutely/nowhere"
|
|
hostname = "notmymachine"
|
|
|
|
[sources.srcimpossible3]
|
|
helper = "failedhelper"
|
|
label = "precious"
|
|
kind = {{ stdout = "blahblah.txt" }}
|
|
|
|
""",
|
|
)
|
|
set_up_simple_yama(yama_path)
|
|
|
|
rng = Random()
|
|
seed = rng.randint(0, 9001)
|
|
print(f"seed: {seed}")
|
|
rng.seed(seed)
|
|
later_expected_descriptor, _ = generate_random_dir(rng, src_path, 32)
|
|
|
|
impossible_proc = subprocess.run(
|
|
("datman", "backup-one", "srcimpossible", "main"),
|
|
cwd=datman_path,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
self.assertNotEqual(impossible_proc.returncode, 0)
|
|
last_line = impossible_proc.stderr.decode().split("\n")[-2]
|
|
self.assertIn("\x1b[31m\x1b[1mFAILED", last_line)
|
|
|
|
# NOT YET SUPPORTED
|
|
print("TEST FOR REMOTE SRC LOUDNESS NOT YET SUPPORTED")
|
|
# impossible_proc = subprocess.run(("datman", "backup-one",
|
|
# "srcimpossible2", "main"),
|
|
# cwd=datman_path,
|
|
# stderr=subprocess.PIPE)
|
|
# self.assertNotEqual(impossible_proc.returncode, 0)
|
|
# last_line = impossible_proc.stderr.decode().split("\n")[-2]
|
|
# self.assertIn("\x1b[31m\x1b[1mFAILED", last_line)
|
|
|
|
impossible_proc = subprocess.run(
|
|
("datman", "backup-one", "srcimpossible3", "main"),
|
|
cwd=datman_path,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
self.assertNotEqual(impossible_proc.returncode, 0)
|
|
last_line = impossible_proc.stderr.decode().split("\n")[-2]
|
|
self.assertIn("\x1b[31m\x1b[1mFAILED", last_line)
|
|
|
|
td.cleanup()
|
|
|
|
def test_backup_incremental(self):
|
|
td = TemporaryDirectory("test_backup_incremental")
|
|
tdpath = Path(td.name)
|
|
|
|
datman_path = tdpath.joinpath("datman")
|
|
src_path = datman_path.joinpath("srca")
|
|
yama_path = datman_path.joinpath("main")
|
|
|
|
set_up_simple_datman(datman_path)
|
|
set_up_simple_yama(yama_path)
|
|
|
|
rng = Random()
|
|
seed = rng.randint(0, 9001)
|
|
print(f"seed: {seed}")
|
|
rng.seed(seed)
|
|
initial_descriptor, _ = generate_random_dir(rng, src_path, 32)
|
|
|
|
print("storing")
|
|
subprocess.check_call(("datman", "backup-one", "srca", "main"), cwd=datman_path)
|
|
|
|
# now mutate and store incremental
|
|
randomly_mutate_directory_in_descriptor(initial_descriptor, src_path, rng)
|
|
mutated_descriptor = scan_dir(src_path)
|
|
|
|
time.sleep(2)
|
|
timestamp_formatted = time.strftime("%Y-%m-%d %H:%M:%S")
|
|
time.sleep(2)
|
|
|
|
subprocess.check_call(("datman", "backup-one", "srca", "main"), cwd=datman_path)
|
|
|
|
print("extracting")
|
|
dest1_path = tdpath.joinpath("desta1")
|
|
dest2_path = tdpath.joinpath("desta2")
|
|
subprocess.check_call(
|
|
(
|
|
"datman",
|
|
"extract",
|
|
"--skip-metadata",
|
|
"--accept-partial",
|
|
"--before",
|
|
timestamp_formatted,
|
|
"main",
|
|
"../desta1",
|
|
),
|
|
cwd=datman_path,
|
|
)
|
|
subprocess.check_call(
|
|
(
|
|
"datman",
|
|
"extract",
|
|
"--skip-metadata",
|
|
"--accept-partial",
|
|
"main",
|
|
"../desta2",
|
|
),
|
|
cwd=datman_path,
|
|
)
|
|
|
|
# this will be wrapped in a directory that starts with the name srca+
|
|
extracted_dir_descriptor_wrapper = scan_dir(dest1_path)
|
|
|
|
contents = extracted_dir_descriptor_wrapper.contents
|
|
self.assertEqual(len(contents), 1)
|
|
key, value = next(iter(contents.items()))
|
|
self.assertTrue(key.startswith("srca+"))
|
|
|
|
self.assertIsInstance(value, DirectoryDescriptor)
|
|
key, value = next(iter(value.contents.items()))
|
|
self.assertEqual(key, "srca")
|
|
|
|
self.assertEqual(value.ignore_metadata(), initial_descriptor.ignore_metadata())
|
|
|
|
extracted_dir_descriptor_wrapper = scan_dir(dest2_path)
|
|
|
|
contents = extracted_dir_descriptor_wrapper.contents
|
|
self.assertEqual(len(contents), 1)
|
|
key, value = next(iter(contents.items()))
|
|
self.assertTrue(key.startswith("srca+"))
|
|
|
|
self.assertIsInstance(value, DirectoryDescriptor)
|
|
key, value = next(iter(value.contents.items()))
|
|
self.assertEqual(key, "srca")
|
|
|
|
self.assertEqual(value.ignore_metadata(), mutated_descriptor.ignore_metadata())
|
|
|
|
td.cleanup()
|
|
|
|
def test_labels_apply(self):
|
|
td = TemporaryDirectory("test_labels_apply")
|
|
tdpath = Path(td.name)
|
|
|
|
datman_path = tdpath.joinpath("datman")
|
|
labelling_path = datman_path.joinpath("labelling")
|
|
src_path = datman_path.joinpath("srca")
|
|
yama_precious_path = datman_path.joinpath("precious")
|
|
yama_pocket_path = datman_path.joinpath("pocket")
|
|
yama_bulky_path = datman_path.joinpath("bulky")
|
|
|
|
set_up_simple_datman(datman_path, piles_section=MULTI_PILES_SECTION)
|
|
set_up_simple_yama(yama_precious_path)
|
|
set_up_simple_yama(yama_pocket_path)
|
|
set_up_simple_yama(yama_bulky_path)
|
|
|
|
rng = Random()
|
|
seed = rng.randint(0, 9001)
|
|
seed = 7555
|
|
print(f"seed: {seed}")
|
|
rng.seed(seed)
|
|
# min_files is 8 because we need enough files to use each label for this test to succeed.
|
|
initial_descriptor, _ = generate_random_dir(rng, src_path, 32, min_files=8)
|
|
labellings = generate_labels(initial_descriptor, rng)
|
|
save_labelling_rules(labelling_path.joinpath("srca.zst"), labellings)
|
|
|
|
for label in ["precious", "pocket", "bulky"]:
|
|
output = subprocess.check_output(
|
|
("datman", "backup-one", "srca", label),
|
|
cwd=datman_path,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
self.assertNotIn(b"Unlabelled", output, "Labelling doesn't seem complete.")
|
|
|
|
for label in ["precious", "pocket", "bulky"]:
|
|
dest_path = tdpath.joinpath(f"dest_{label}")
|
|
subprocess.check_call(
|
|
(
|
|
"datman",
|
|
"extract",
|
|
"--skip-metadata",
|
|
"--accept-partial",
|
|
label,
|
|
str(dest_path),
|
|
),
|
|
cwd=datman_path,
|
|
)
|
|
|
|
extracted_dir_descriptor_wrapper = scan_dir(dest_path)
|
|
|
|
contents = extracted_dir_descriptor_wrapper.contents
|
|
|
|
self.assertEqual(len(contents), 1, contents)
|
|
key, value = next(iter(contents.items()))
|
|
self.assertTrue(key.startswith("srca+"))
|
|
|
|
self.assertIsInstance(value, DirectoryDescriptor)
|
|
key, value = next(iter(value.contents.items()))
|
|
self.assertEqual(key, "srca")
|
|
|
|
filtered_initial_descriptor = filter_descriptor_by_label(
|
|
{label}, initial_descriptor, labellings
|
|
)
|
|
self.assertEqual(
|
|
value.ignore_metadata(), filtered_initial_descriptor.ignore_metadata()
|
|
)
|
|
|
|
td.cleanup()
|