Skip to content
Snippets Groups Projects
Commit bdd9605e authored by Nicolas KAROLAK's avatar Nicolas KAROLAK
Browse files

change(tests): test_backup output format

parent 987d4ad9
No related branches found
No related tags found
No related merge requests found
...@@ -37,19 +37,19 @@ def test_ssh(host: str) -> bool: ...@@ -37,19 +37,19 @@ def test_ssh(host: str) -> bool:
) )
try: try:
subprocess.check_output(cmd, shell=True, timeout=5) subprocess.check_output(cmd, shell=True, timeout=5)
u.success("Logged in successfully") u.success("logged in successfully")
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
u.error("Failed to login using SSH public key authentication") u.error("failed to login using SSH public key authentication")
return False return False
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
try: try:
cmd_port = "nc -z -w2 {} 22".format(host) cmd_port = "nc -z -w2 {} 22".format(host)
subprocess.check_output(cmd_port, shell=True, timeout=5) subprocess.check_output(cmd_port, shell=True, timeout=5)
u.error("Failed to bind SSH port") u.error("failed to bind SSH port")
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
cmd_ping = "ping -c2 -w4 {}".format(host) cmd_ping = "ping -c2 -w4 {}".format(host)
subprocess.check_output(cmd_ping, shell=True, timeout=15) subprocess.check_output(cmd_ping, shell=True, timeout=15)
u.error("Failed to ping host") u.error("failed to ping host")
return False return False
return True return True
...@@ -76,13 +76,13 @@ def test_last_backup_is_recent(server: str) -> bool: ...@@ -76,13 +76,13 @@ def test_last_backup_is_recent(server: str) -> bool:
date = out.strip().split(" ")[-1] date = out.strip().split(" ")[-1]
pdate = datetime.strptime(date, "%Y-%m-%d-%H%M%S") pdate = datetime.strptime(date, "%Y-%m-%d-%H%M%S")
if (datetime.now() - pdate).days > MAX_AGE: if (datetime.now() - pdate).days > MAX_AGE:
u.error("Backup is older than {} days".format(MAX_AGE)) u.error("older than {} days".format(MAX_AGE))
return False return False
u.success("Backup is less than {} days old".format(MAX_AGE)) u.success("less than {} days old".format(MAX_AGE))
return True return True
out = out or "No output." out = out or "no output"
u.error("SSH access is not working (code: {}): {}".format(status, out)) u.error("SSH access not working (code: {}): {}".format(status, out))
return False return False
...@@ -96,21 +96,20 @@ def check_backup_is_incremental(path: str) -> bool: ...@@ -96,21 +96,20 @@ def check_backup_is_incremental(path: str) -> bool:
:rtype: bool :rtype: bool
""" """
print("Checking that backup is incremental:") all_ok = True
for directory in os.listdir(path):
dirs = os.listdir(path)
for directory in dirs:
files_count = 0 files_count = 0
folder_path = os.path.join(path, directory) folder_path = os.path.join(path, directory)
if os.path.isdir(folder_path): if os.path.isdir(folder_path):
files_count = len(os.listdir(folder_path)) files_count = len(os.listdir(folder_path))
if files_count == 0: if files_count == 0:
u.error("Folder {} is empty".format(folder_path)) u.error("folder {} is empty".format(folder_path))
os.rmdir(folder_path) os.rmdir(folder_path)
return False all_ok = False
u.success("Folder {} is not empty".format(folder_path)) if all_ok:
u.success("no incrementation issue found")
return True return all_ok
def check_local_backup(path: str) -> bool: def check_local_backup(path: str) -> bool:
...@@ -123,7 +122,7 @@ def check_local_backup(path: str) -> bool: ...@@ -123,7 +122,7 @@ def check_local_backup(path: str) -> bool:
""" """
backup_folder = os.path.dirname(path) backup_folder = os.path.dirname(path)
print("Checking %s" % backup_folder) print("Checking {}:".format(backup_folder))
all_ok = True all_ok = True
latest = os.path.join(backup_folder, "latest") latest = os.path.join(backup_folder, "latest")
...@@ -135,17 +134,17 @@ def check_local_backup(path: str) -> bool: ...@@ -135,17 +134,17 @@ def check_local_backup(path: str) -> bool:
now = datetime.now() now = datetime.now()
diff_seconds = (now - date).total_seconds() diff_seconds = (now - date).total_seconds()
if diff_seconds > MAX_AGE * 24 * 3600: if diff_seconds > MAX_AGE * 24 * 3600:
u.error("Backup {} is older than {} days".format(backup_folder, MAX_AGE)) u.error("older than {} days".format(MAX_AGE))
all_ok = False all_ok = False
else: else:
u.success("Backup {} is fine".format(backup_folder)) u.success("less than {} days old".format(MAX_AGE))
if not check_backup_is_incremental(backup_folder): if not check_backup_is_incremental(backup_folder):
all_ok = False all_ok = False
elif os.path.exists(os.path.join(backup_folder, "backup.inprogress")): elif os.path.exists(os.path.join(backup_folder, "backup.inprogress")):
u.error("Initial backup %s still running" % backup_folder) u.warning("still running")
all_ok = False all_ok = False
else: else:
u.error("Backup {} is not working".format(latest)) u.error("not working")
all_ok = False all_ok = False
return all_ok return all_ok
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment