Add an end-to-end test for the Btrfs hook using a fake Btrfs binary (#80).

borgmatic/hooks/data_source/btrfs.py

@@ -26,7 +26,8 @@ def get_filesystem_mount_points(findmnt_command):
     findmnt_output = borgmatic.execute.execute_command_and_capture_output(
         tuple(findmnt_command.split(' '))
         + (
-            '-nt',
+            '-n',  # No headings.
+            '-t',  # Filesystem type.
             'btrfs',
         )
     )

scripts/run-full-tests

@@ -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

tests/end-to-end/commands/fake_btrfs.py

@@ -0,0 +1,86 @@
+import argparse
+import json
+import os
+import shutil
+import sys
+
+
+def parse_arguments(*unparsed_arguments):
+    global_parser = argparse.ArgumentParser(add_help=False)
+    action_parsers = global_parser.add_subparsers(dest='action')
+
+    subvolume_parser = action_parsers.add_parser('subvolume')
+    subvolume_subparser = subvolume_parser.add_subparsers(dest='subaction')
+
+    list_parser = subvolume_subparser.add_parser('list')
+    list_parser.add_argument('-s', dest='snapshots_only', action='store_true')
+    list_parser.add_argument('subvolume_path')
+
+    snapshot_parser = subvolume_subparser.add_parser('snapshot')
+    snapshot_parser.add_argument('-r', dest='read_only', action='store_true')
+    snapshot_parser.add_argument('subvolume_path')
+    snapshot_parser.add_argument('snapshot_path')
+
+    delete_parser = subvolume_subparser.add_parser('delete')
+    delete_parser.add_argument('snapshot_path')
+
+    return global_parser.parse_args(unparsed_arguments)
+
+
+BUILTIN_SUBVOLUME_LIST_LINES = (
+    '261 gen 29 top level 5 path sub',
+    '262 gen 29 top level 5 path other',
+)
+SUBVOLUME_LIST_LINE_PREFIX = '263 gen 29 top level 5 path '
+
+
+def load_snapshots():
+    try:
+        return json.load(open('/tmp/fake_btrfs.json'))
+    except FileNotFoundError:
+        return []
+
+
+def save_snapshots(snapshot_paths):
+    json.dump(snapshot_paths, open('/tmp/fake_btrfs.json', 'w'))
+
+
+def print_subvolume_list(arguments, snapshot_paths):
+    assert arguments.subvolume_path == '/mnt/subvolume'
+
+    if not arguments.snapshots_only:
+        for line in BUILTIN_SUBVOLUME_LIST_LINES:
+            print(line)
+
+    for snapshot_path in snapshot_paths:
+        print(SUBVOLUME_LIST_LINE_PREFIX + snapshot_path[snapshot_path.index('.borgmatic-snapshot-'):])
+
+
+def main():
+    arguments = parse_arguments(*sys.argv[1:])
+    snapshot_paths = load_snapshots()
+
+    if arguments.subaction == 'list':
+        print_subvolume_list(arguments, snapshot_paths)
+    elif arguments.subaction == 'snapshot':
+        snapshot_paths.append(arguments.snapshot_path)
+        save_snapshots(snapshot_paths)
+
+        subdirectory = os.path.join(arguments.snapshot_path, 'subdir')
+        os.makedirs(subdirectory, mode=0o700, exist_ok=True)
+        test_file = open(os.path.join(subdirectory, 'file.txt'), 'w')
+        test_file.write('contents')
+        test_file.close()
+    elif arguments.subaction == 'delete':
+        subdirectory = os.path.join(arguments.snapshot_path, 'subdir')
+        shutil.rmtree(subdirectory)
+
+        snapshot_paths = [
+            snapshot_path for snapshot_path in snapshot_paths
+            if snapshot_path.endswith('/' + arguments.snapshot_path)
+        ]
+        save_snapshots(snapshot_paths)
+
+
+if __name__ == '__main__':
+    main()

tests/end-to-end/commands/fake_findmnt.py

@@ -0,0 +1,34 @@
+import argparse
+import json
+import sys
+
+
+def parse_arguments(*unparsed_arguments):
+    parser = argparse.ArgumentParser(add_help=False)
+    parser.add_argument('-n', dest='headings', action='store_false', default=True)
+    parser.add_argument('-t', dest='type')
+
+    return parser.parse_args(unparsed_arguments)
+
+
+BUILTIN_FILESYSTEM_MOUNT_LINES = (
+    '/mnt/subvolume /dev/loop1 btrfs rw,relatime,ssd,space_cache=v2,subvolid=5,subvol=/',
+)
+
+
+def print_filesystem_mounts(arguments):
+    for line in BUILTIN_FILESYSTEM_MOUNT_LINES:
+        print(line)
+
+
+def main():
+    arguments = parse_arguments(*sys.argv[1:])
+
+    assert not arguments.headings
+    assert arguments.type == 'btrfs'
+
+    print_filesystem_mounts(arguments)
+
+
+if __name__ == '__main__':
+    main()

tests/end-to-end/test_btrfs.py

@@ -0,0 +1,61 @@
+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/subvolume/subdir')
+        .replace('- /var/log/syslog*', '')
+        + 'encryption_passphrase: "test"\n'
+        + 'btrfs:\n'
+        + '    btrfs_command: python3 /app/tests/end-to-end/commands/fake_btrfs.py\n'
+        + '    findmnt_command: python3 /app/tests/end-to-end/commands/fake_findmnt.py\n'
+    )
+    config_file = open(config_path, 'w')
+    config_file.write(config)
+    config_file.close()
+
+
+def test_btrfs_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 Btrfs 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/subvolume/subdir/file.txt' in output
+
+        # Assert that the snapshot has been deleted.
+        assert not subprocess.check_output(
+            f'python3 /app/tests/end-to-end/commands/fake_btrfs.py subvolume list -s /mnt/subvolume'.split(' ')
+        )
+    finally:
+        shutil.rmtree(temporary_directory)

tests/end-to-end/test_zfs.py

@@ -19,7 +19,7 @@ def generate_configuration(config_path, repository_path):
         .replace('- path: /mnt/backup', '')
         .replace('label: local', '')
         .replace('- /home', f'- {config_path}')
-        .replace('- /etc', '- /pool/dataset')
+        .replace('- /etc', '- /pool/dataset/subdir')
         .replace('- /var/log/syslog*', '')
         + 'encryption_passphrase: "test"\n'
         + 'zfs:\n'