LVM hook end-to-end tests, not quite working yet (#80).

This commit is contained in:
Dan Helfman 2024-12-05 22:46:50 -08:00
parent 03bbe77dd9
commit ec9e1a8223
10 changed files with 293 additions and 10 deletions

View File

@ -2333,11 +2333,11 @@ properties:
description: |
Command to use instead of "lvs".
example: /usr/local/bin/lvs
lsbrk_command:
lsblk_command:
type: string
description: |
Command to use instead of "lsbrk".
example: /usr/local/bin/lsbrk
Command to use instead of "lsblk".
example: /usr/local/bin/lsblk
mount_command:
type: string
description: |

View File

@ -39,10 +39,10 @@ def get_logical_volumes(lsblk_command, source_directories=None):
'''
try:
devices_info = json.loads(
subprocess.check_output(
(
# Use lsblk instead of lvs here because lvs can't show active mounts.
lsblk_command,
borgmatic.execute.execute_command_and_capture_output(
# Use lsblk instead of lvs here because lvs can't show active mounts.
tuple(lsblk_command.split(' '))
+ (
'--output',
'name,path,mountpoint,type',
'--json',
@ -229,7 +229,7 @@ def unmount_snapshot(umount_command, snapshot_mount_path): # pragma: no cover
)
def delete_snapshot(lvremove_command, snapshot_device_path): # pragma: no cover
def remove_snapshot(lvremove_command, snapshot_device_path): # pragma: no cover
'''
Given an lvremove command to run and the device path of a snapshot, remove it it.
'''
@ -362,7 +362,7 @@ def remove_data_source_dumps(hook_config, config, log_prefix, borgmatic_runtime_
logger.debug(f'{log_prefix}: Deleting LVM snapshot {snapshot.name}{dry_run_label}')
if not dry_run:
delete_snapshot(lvremove_command, snapshot.device_path)
remove_snapshot(lvremove_command, snapshot.device_path)
def make_data_source_dump_patterns(

View File

@ -144,6 +144,8 @@ def mount_snapshot(mount_command, full_snapshot_name, snapshot_mount_path): # p
+ (
'-t',
'zfs',
'-o',
'ro',
full_snapshot_name,
snapshot_mount_path,
),

View File

@ -25,5 +25,5 @@ python3 -m pip install --no-cache --upgrade pip==24.2 setuptools==72.1.0
pip3 install --ignore-installed tox==4.11.3
export COVERAGE_FILE=/tmp/.coverage
#tox --workdir /tmp/.tox --sitepackages
tox --workdir /tmp/.tox --sitepackages
tox --workdir /tmp/.tox --sitepackages -e end-to-end

View File

@ -0,0 +1,79 @@
import argparse
import json
import sys
def parse_arguments(*unparsed_arguments):
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('--output', required=True)
parser.add_argument('--json', action='store_true', required=True)
parser.add_argument('--list', action='store_true', required=True)
return parser.parse_args(unparsed_arguments)
BUILTIN_BLOCK_DEVICES = {
'blockdevices': [
{
'name': 'loop0',
'path': '/dev/loop0',
'mountpoint': None,
'type': 'loop'
},
{
'name': 'cryptroot',
'path': '/dev/mapper/cryptroot',
'mountpoint': '/',
'type': 'crypt'
},{
'name': 'vgroup-lvolume',
'path': '/dev/mapper/vgroup-lvolume',
'mountpoint': '/mnt/lvolume',
'type': 'lvm'
},
{
'name': 'vgroup-lvolume-real',
'path': '/dev/mapper/vgroup-lvolume-real',
'mountpoint': None,
'type': 'lvm'
},
]
}
def load_snapshots():
try:
return json.load(open('/tmp/fake_lvm.json'))
except FileNotFoundError:
return []
def print_logical_volumes_json(arguments, snapshots):
data = dict(BUILTIN_BLOCK_DEVICES)
for snapshot in snapshots:
data['blockdevices'].extend(
{
'name': snapshot['lv_name'],
'path': snapshot['lv_path'],
'mountpoint': None,
'type': 'lvm'
}
for snapshot in snapshots
)
print(json.dumps(data))
def main():
arguments = parse_arguments(*sys.argv[1:])
snapshots = load_snapshots()
assert arguments.output == 'name,path,mountpoint,type'
print_logical_volumes_json(arguments, snapshots)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,43 @@
import argparse
import json
import sys
def parse_arguments(*unparsed_arguments):
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('--snapshot', action='store_true', required=True)
parser.add_argument('--extents')
parser.add_argument('--size')
parser.add_argument('--name', dest='snapshot_name', required=True)
parser.add_argument('logical_volume_device')
return parser.parse_args(unparsed_arguments)
def load_snapshots():
try:
return json.load(open('/tmp/fake_lvm.json'))
except FileNotFoundError:
return []
def save_snapshots(snapshots):
json.dump(snapshots, open('/tmp/fake_lvm.json', 'w'))
def main():
arguments = parse_arguments(*sys.argv[1:])
snapshots = load_snapshots()
assert arguments.extents or arguments.size
snapshots.append(
{'lv_name': arguments.snapshot_name, 'lv_path': f'/dev/vgroup/{arguments.snapshot_name}'},
)
save_snapshots(snapshots)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,38 @@
import argparse
import json
import sys
def parse_arguments(*unparsed_arguments):
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('--force', action='store_true', required=True)
parser.add_argument('snapshot_device')
return parser.parse_args(unparsed_arguments)
def load_snapshots():
try:
return json.load(open('/tmp/fake_lvm.json'))
except FileNotFoundError:
return []
def save_snapshots(snapshots):
json.dump(snapshots, open('/tmp/fake_lvm.json', 'w'))
def main():
arguments = parse_arguments(*sys.argv[1:])
snapshots = [
snapshot for snapshot in load_snapshots()
if snapshot['lv_path'] == arguments.snapshot_device
]
save_snapshots(snapshots)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,52 @@
import argparse
import json
import sys
def parse_arguments(*unparsed_arguments):
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('--report-format', required=True)
parser.add_argument('--options', required=True)
parser.add_argument('--select', required=True)
return parser.parse_args(unparsed_arguments)
def load_snapshots():
try:
return json.load(open('/tmp/fake_lvm.json'))
except FileNotFoundError:
return []
def print_snapshots_json(arguments, snapshots):
assert arguments.report_format == 'json'
assert arguments.options == 'lv_name,lv_path'
assert arguments.select == 'lv_attr =~ ^s'
print(
json.dumps(
{
'report': [
{
'lv': snapshots,
}
]
,
'log': [
]
}
)
)
def main():
arguments = parse_arguments(*sys.argv[1:])
snapshots = load_snapshots()
print_snapshots_json(arguments, snapshots)
if __name__ == '__main__':
main()

View File

@ -6,6 +6,7 @@ import sys
def parse_arguments(*unparsed_arguments):
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('-t', dest='type')
parser.add_argument('-o', dest='options')
parser.add_argument('snapshot_name')
parser.add_argument('mount_point')
@ -15,6 +16,8 @@ def parse_arguments(*unparsed_arguments):
def main():
arguments = parse_arguments(*sys.argv[1:])
assert arguments.options == 'ro'
subdirectory = os.path.join(arguments.mount_point, 'subdir')
os.mkdir(subdirectory)
test_file = open(os.path.join(subdirectory, 'file.txt'), 'w')

View File

@ -0,0 +1,66 @@
import json
import os
import shutil
import subprocess
import sys
import tempfile
def generate_configuration(config_path, repository_path):
'''
Generate borgmatic configuration into a file at the config path, and update the defaults so as
to work for testing (including injecting the given repository path and tacking on an encryption
passphrase).
'''
subprocess.check_call(f'borgmatic config generate --destination {config_path}'.split(' '))
config = (
open(config_path)
.read()
.replace('ssh://user@backupserver/./sourcehostname.borg', repository_path)
.replace('- path: /mnt/backup', '')
.replace('label: local', '')
.replace('- /home', f'- {config_path}')
.replace('- /etc', '- /mnt/lvolume/subdir')
.replace('- /var/log/syslog*', '')
+ 'encryption_passphrase: "test"\n'
+ 'lvm:\n'
+ ' lsblk_command: python3 /app/tests/end-to-end/commands/fake_lsblk.py\n'
+ ' lvcreate_command: python3 /app/tests/end-to-end/commands/fake_lvcreate.py\n'
+ ' lvremove_command: python3 /app/tests/end-to-end/commands/fake_lvremove.py\n'
+ ' lvs_command: python3 /app/tests/end-to-end/commands/fake_lvs.py\n'
+ ' mount_command: python3 /app/tests/end-to-end/commands/fake_mount.py\n'
+ ' umount_command: python3 /app/tests/end-to-end/commands/fake_umount.py\n'
)
config_file = open(config_path, 'w')
config_file.write(config)
config_file.close()
def test_lvm_create_and_list():
temporary_directory = tempfile.mkdtemp()
repository_path = os.path.join(temporary_directory, 'test.borg')
try:
config_path = os.path.join(temporary_directory, 'test.yaml')
generate_configuration(config_path, repository_path)
subprocess.check_call(
f'borgmatic -v 2 --config {config_path} repo-create --encryption repokey'.split(' ')
)
# Run a create action to exercise LVM snapshotting and backup.
subprocess.check_call(f'borgmatic --config {config_path} create'.split(' '))
# List the resulting archive and assert that the snapshotted files are there.
output = subprocess.check_output(
f'borgmatic --config {config_path} list --archive latest'.split(' ')
).decode(sys.stdout.encoding)
assert 'mnt/lvolume/subdir/file.txt' in output
# Assert that the snapshot has been deleted.
assert not json.loads(subprocess.check_output(
'python3 /app/tests/end-to-end/commands/fake_lvs.py --report-format json --options lv_name,lv_path --select'.split(' ') + ['lv_attr =~ ^s']
))['report'][0]['lv']
finally:
shutil.rmtree(temporary_directory)