zfs-snapshot-backup/compare_checksums.py

#!/usr/local/bin/python

import pathlib
import subprocess

from copy_snapshots_to_remote import get_member_name
from create_snapshot_backup import call, remote_call

SSH_KEY_FILE = "/mnt/Datenspeicher/snap-backup-dataset/backup_key"
SSH_REMOTE = "zfs_snap_backup@etha.cpi.imtek.uni-freiburg.de"

REMOTE_PATH = "zfs-backups"

ZFS_POOL = "Datenspeicher"
ZFS_ELAB_PREFIX = "elabfs-"

TMP_BACKUP_FOLDER = "/mnt/Datenspeicher/snap-backup-dataset/temporary-backups"


def xremote_call(arguments):
    """ makes runs an command on the remote backup server

    :params arguments: list of command line arguments and parameters
    :returns: string of the command output
    :raises subprocess.CalledProcessError: if command has not an exit value of 0
    """
    cmd = ["ssh", "-i", SSH_KEY_FILE, SSH_REMOTE]
    cmd.extend(arguments)
    result = subprocess.run(" ".join(cmd), check=True,
        stdout=subprocess.PIPE,
        universal_newlines=True, shell=True)
    return result.stdout


def get_local_backups():
    tmp_folder = pathlib.Path(TMP_BACKUP_FOLDER)
    return [i for i in tmp_folder.iterdir() if i.suffix==".gz"]


def get_remote_checksum(local_backup_path):
    member = get_member_name(local_backup_path.name)
    remote_path =  f"{REMOTE_PATH}/{member}/{local_backup_path.name}"
    try:
        result = remote_call(["sha256sum", remote_path])
        parts = result.strip().split()
        return parts[0].strip()
    except subprocess.CalledProcessError as e:
        return None

def get_local_checksum(local_backup_path):
    result = call(["sha256", str(local_backup_path)], as_text=True)
    parts = result.split("=")
    return parts[1].strip()

def batch_check():
    for local_backup_path in get_local_backups():
        remote_checksum = get_remote_checksum(local_backup_path)
        local_checksum = get_local_checksum(local_backup_path)
        if remote_checksum == local_checksum:
            print(f"     {local_checksum} -> {remote_checksum}")
        else:
            print(f"XXX  {local_checksum} -> {remote_checksum}")


if __name__ == "__main__":
    batch_check()