tests: Improve restic download strategy

This commit is contained in:
deajan 2024-12-10 23:37:14 +01:00
parent bbc19f749d
commit cb89b3de36
2 changed files with 43 additions and 30 deletions

View file

@ -23,7 +23,7 @@ sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(__file__), ".."
from npbackup.path_helper import BASEDIR from npbackup.path_helper import BASEDIR
def download_restic_binaries(arch: str = "amd64", move_is_fatal: bool = True) -> bool: def download_restic_binaries(arch: str = "amd64") -> bool:
""" """
We must first download latest restic binaries to make sure we can run all tests and/or compile We must first download latest restic binaries to make sure we can run all tests and/or compile
""" """
@ -36,29 +36,54 @@ def download_restic_binaries(arch: str = "amd64", move_is_fatal: bool = True) ->
# print("RESPONSE: ", response) # print("RESPONSE: ", response)
json_response = json.loads(response.text) json_response = json.loads(response.text)
current_version = json_response["tag_name"].lstrip("v") current_version = json_response["tag_name"].lstrip("v")
# print("JSON RESPONSE") # print("JSON RESPONSE")
# pprint(json_response, indent=5) # pprint(json_response, indent=5)
dest_dir = Path(BASEDIR).absolute().parent.joinpath("RESTIC_SOURCE_FILES") dest_dir = Path(BASEDIR).absolute().parent.joinpath("RESTIC_SOURCE_FILES")
if os.name == "nt": if os.name == "nt":
fname = f"_windows_{arch}.zip" fname = f"_windows_{arch}"
suffix = ".exe" suffix = ".exe"
arch_suffix = ".zip"
else: else:
fname = f"_linux_{arch}.bz2" fname = f"_linux_{arch}"
suffix = "" suffix = ""
arch_suffix = ".bz2"
if not dest_dir.joinpath("ARCHIVES").is_dir():
os.makedirs(dest_dir.joinpath("ARCHIVES"))
dest_file = dest_dir.joinpath("restic_" + current_version + fname + arch_suffix)
dest_file = dest_dir.joinpath("restic_" + current_version + fname).with_suffix(
suffix
)
if dest_file.is_file(): if dest_file.is_file():
print(f"RESTIC SOURCE ALREADY PRESENT. NOT DOWNLOADING {dest_file}") print(f"RESTIC SOURCE ALREADY PRESENT. NOT DOWNLOADING {dest_file}")
return True return True
else: else:
print(f"DOWNALOADING RESTIC {dest_file}") print(f"DOWNALOADING RESTIC {dest_file}")
# Also we need to move any earlier file that may not be current version to archives
for file in dest_dir.glob(f"restic_*{fname}{suffix}"):
# We need to keep legacy binary for Windows 7 32 bits
if (
not file == "restic_0.16.2_windows_386.exe"
and not file == "restic_0.16.2_windows_amd64.exe"
):
try:
archive_file = dest_dir.joinpath("ARCHIVES").joinpath(file.name)
if archive_file.is_file():
archive_file.unlink()
shutil.move(
file,
archive_file,
)
except OSError as exc:
print(
f"CANNOT MOVE OLD FILES ARCHIVE: {file} to {archive_file}: {exc}"
)
return False
downloaded = False downloaded = False
for entry in json_response["assets"]: for entry in json_response["assets"]:
if fname in entry["browser_download_url"]: if f"{fname}{arch_suffix}" in entry["browser_download_url"]:
file_request = requests.get( file_request = requests.get(
entry["browser_download_url"], allow_redirects=True entry["browser_download_url"], allow_redirects=True
) )
@ -66,26 +91,25 @@ def download_restic_binaries(arch: str = "amd64", move_is_fatal: bool = True) ->
filename = entry["browser_download_url"].rsplit("/", 1)[1] filename = entry["browser_download_url"].rsplit("/", 1)[1]
full_path = dest_dir.joinpath(filename) full_path = dest_dir.joinpath(filename)
print("PATH TO DOWNLOADED ARCHIVE: ", full_path) print("PATH TO DOWNLOADED ARCHIVE: ", full_path)
if fname.endswith("bz2"): if arch_suffix == ".bz2":
with open(full_path.with_suffix(""), "wb") as fp: with open(str(full_path).rstrip("arch_suffix"), "wb") as fp:
fp.write(bz2.decompress(file_request.content)) fp.write(bz2.decompress(file_request.content))
# We also need to make that file executable # We also need to make that file executable
os.chmod(full_path.with_suffix(""), 0o775) os.chmod(str(full_path).rstrip(arch_suffix), 0o775)
else: else:
with open(full_path, "wb") as fp: with open(full_path, "wb") as fp:
fp.write(file_request.content) fp.write(file_request.content)
# Assume we have a zip or tar.gz # Assume we have a zip or tar.gz
shutil.unpack_archive(full_path, dest_dir) shutil.unpack_archive(full_path, dest_dir)
try: try:
if not dest_dir.joinpath("ARCHIVES").is_dir(): if dest_dir.joinpath("ARCHIVES").joinpath(filename).is_file():
os.makedirs(dest_dir.joinpath("ARCHIVES")) dest_dir.joinpath("ARCHIVES").joinpath(filename).unlink()
shutil.move(full_path, dest_dir.joinpath("ARCHIVES").joinpath(filename)) shutil.move(full_path, dest_dir.joinpath("ARCHIVES").joinpath(filename))
except OSError: except OSError as exc:
if move_is_fatal: print(
print( f'CANNOT MOVE TO ARCHIVE: {full_path} to {dest_dir.joinpath("ARCHIVES").joinpath(filename)}: {[exc]}'
f'CANNOT MOVE TO ARCHIVE: {full_path} to {dest_dir.joinpath("ARCHIVES").joinpath(filename)}' )
) return False
return False
print(f"DOWNLOADED {dest_dir}") print(f"DOWNLOADED {dest_dir}")
downloaded = True downloaded = True
break break

View file

@ -65,15 +65,6 @@ repo_config, _ = get_repo_config(full_config)
DUMP_FILE = "__version__.py" DUMP_FILE = "__version__.py"
def running_on_github_actions():
"""
This is set in github actions workflow with
env:
RUNNING_ON_GITHUB_ACTIONS: true
"""
return os.environ.get("RUNNING_ON_GITHUB_ACTIONS", "False").lower() == "true"
class RedirectedStdout: class RedirectedStdout:
""" """
Balantly copied from https://stackoverflow.com/a/45899925/2635443 Balantly copied from https://stackoverflow.com/a/45899925/2635443
@ -100,9 +91,7 @@ def test_download_restic_binaries():
We must first download latest restic binaries to make sure we can run all tests We must first download latest restic binaries to make sure we can run all tests
Currently we only run these on amd64 Currently we only run these on amd64
""" """
assert download_restic_binaries( assert download_restic_binaries("amd64"), "Could not download restic binaries"
"amd64", move_is_fatal=not running_on_github_actions()
), "Could not download restic binaries"
def test_npbackup_cli_no_config(): def test_npbackup_cli_no_config():