mirror of
https://github.com/resf/distro-tools.git
synced 2024-11-14 09:11:26 +00:00
591 lines
18 KiB
Python
591 lines
18 KiB
Python
import tempfile
|
|
import shutil
|
|
import pathlib
|
|
import hashlib
|
|
from os import path, environ
|
|
|
|
import pytest
|
|
|
|
from apollo.publishing_tools import apollo_tree
|
|
|
|
from common.testing import MockResponse
|
|
|
|
data = [
|
|
"baseos__base__repomd__x86_64.xml",
|
|
"baseos__base__repomd__aarch64.xml",
|
|
"appstream__base__repomd__x86_64.xml",
|
|
"appstream__base__repomd__aarch64.xml",
|
|
]
|
|
|
|
|
|
async def _setup_test_baseos(directory: str):
|
|
file = data[0]
|
|
base_dir = path.join(
|
|
directory,
|
|
"BaseOS/x86_64/os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
return repos
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_path_valid_structure():
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
# Copy test data to temp dir
|
|
for file in data:
|
|
fsplit = file.split("__")
|
|
base_dir = path.join(
|
|
directory,
|
|
fsplit[0],
|
|
fsplit[-1].removesuffix(".xml"),
|
|
"os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "baseos" in repos
|
|
assert "appstream" in repos
|
|
assert len(repos["baseos"]) == 2
|
|
assert len(repos["appstream"]) == 2
|
|
|
|
for repo in repos["baseos"]:
|
|
assert repo["name"] == "baseos"
|
|
assert repo["arch"] in ["x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
"baseos",
|
|
repo["arch"],
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
for repo in repos["appstream"]:
|
|
assert repo["name"] == "appstream"
|
|
assert repo["arch"] in ["x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
"appstream",
|
|
repo["arch"],
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_path_multiple_formats():
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
# Copy test data to temp dir
|
|
for file in data:
|
|
fsplit = file.split("__")
|
|
base_dir = path.join(
|
|
directory,
|
|
fsplit[0],
|
|
fsplit[-1].removesuffix(".xml"),
|
|
"os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
file = data[0]
|
|
fsplit = file.split("__")
|
|
base_dir = path.join(
|
|
directory,
|
|
fsplit[0],
|
|
"source/tree/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "baseos" in repos
|
|
assert "appstream" in repos
|
|
assert len(repos["baseos"]) == 2
|
|
assert len(repos["appstream"]) == 2
|
|
|
|
for repo in repos["baseos"]:
|
|
assert repo["name"] == "baseos"
|
|
assert repo["arch"] in ["source", "x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
"baseos",
|
|
repo["arch"],
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
for repo in repos["appstream"]:
|
|
assert repo["name"] == "appstream"
|
|
assert repo["arch"] in ["x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
"appstream",
|
|
repo["arch"],
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
# Run scan_path for source
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/source/tree/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "baseos" in repos
|
|
assert len(repos["baseos"]) == 1
|
|
|
|
for repo in repos["baseos"]:
|
|
assert repo["name"] == "baseos"
|
|
assert repo["arch"] == "source"
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
"baseos",
|
|
"source",
|
|
"tree/repodata/repomd.xml",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_path_valid_structure_arch_first():
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
# Copy test data to temp dir
|
|
for file in data:
|
|
fsplit = file.split("__")
|
|
base_dir = path.join(
|
|
directory,
|
|
fsplit[-1].removesuffix(".xml"),
|
|
fsplit[0],
|
|
"os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$arch/$reponame/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "baseos" in repos
|
|
assert "appstream" in repos
|
|
assert len(repos["baseos"]) == 2
|
|
assert len(repos["appstream"]) == 2
|
|
|
|
for repo in repos["baseos"]:
|
|
assert repo["name"] == "baseos"
|
|
assert repo["arch"] in ["x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
repo["arch"],
|
|
"baseos",
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
for repo in repos["appstream"]:
|
|
assert repo["name"] == "appstream"
|
|
assert repo["arch"] in ["x86_64", "aarch64"]
|
|
assert repo["found_path"] == path.join(
|
|
directory,
|
|
repo["arch"],
|
|
"appstream",
|
|
"os/repodata/repomd.xml",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_updateinfo_from_apollo_live():
|
|
# This test is only run if the environment variable
|
|
# TEST_WITH_SIDE_EFFECTS is set to 1
|
|
if not environ.get("TEST_WITH_SIDE_EFFECTS"):
|
|
pytest.skip("Skipping test_fetch_updateinfo_from_apollo_live")
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
file = data[0]
|
|
base_dir = path.join(
|
|
directory,
|
|
"BaseOS/x86_64/os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "BaseOS" in repos
|
|
assert len(repos["BaseOS"]) == 1
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
assert updateinfo is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_updateinfo_from_apollo_live_no_updateinfo():
|
|
# This test is only run if the environment variable
|
|
# TEST_WITH_SIDE_EFFECTS is set to 1
|
|
if not environ.get("TEST_WITH_SIDE_EFFECTS"):
|
|
pytest.skip(
|
|
"Skipping test_fetch_updateinfo_from_apollo_live_no_updateinfo"
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
file = data[0]
|
|
base_dir = path.join(
|
|
directory,
|
|
"BaseOS/x86_64/os/repodata",
|
|
)
|
|
pathlib.Path(base_dir).mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(
|
|
path.join(path.dirname(__file__), "data", file),
|
|
path.join(base_dir, "repomd.xml"),
|
|
)
|
|
|
|
# Run scan_path
|
|
repos = await apollo_tree.scan_path(
|
|
directory,
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
[],
|
|
)
|
|
|
|
assert "BaseOS" in repos
|
|
assert len(repos["BaseOS"]) == 1
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64 NONEXISTENT",
|
|
)
|
|
|
|
assert updateinfo is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_updateinfo_from_apollo_mock(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
assert updateinfo == updateinfo_xml
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gzip_updateinfo(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
updateinfo = None
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
assert updateinfo == updateinfo_xml
|
|
break
|
|
|
|
# Run gzip_updateinfo
|
|
updateinfo_gz = await apollo_tree.gzip_updateinfo(updateinfo)
|
|
assert updateinfo_gz is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_updateinfo_to_file(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
updateinfo = None
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
assert updateinfo == updateinfo_xml
|
|
break
|
|
|
|
# Gzip first
|
|
gzipped = await apollo_tree.gzip_updateinfo(updateinfo)
|
|
|
|
# Run write_updateinfo_to_file
|
|
updateinfo_file = await apollo_tree.write_updateinfo_to_file(
|
|
repos["BaseOS"][0]["found_path"],
|
|
gzipped,
|
|
)
|
|
|
|
assert updateinfo_file is not None
|
|
assert path.exists(updateinfo_file)
|
|
assert path.isfile(updateinfo_file)
|
|
|
|
with open(updateinfo_file, "rb") as f:
|
|
updateinfo_file_contents = f.read()
|
|
|
|
# Check sha256sum against written file
|
|
actual_hexdigest = hashlib.sha256(updateinfo_file_contents).hexdigest()
|
|
expected_hexdigest = gzipped["gzipped_sha256sum"]
|
|
assert actual_hexdigest == expected_hexdigest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_repomd_xml(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
# Run fetch_updateinfo_from_apollo
|
|
updateinfo = None
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
updateinfo = await apollo_tree.fetch_updateinfo_from_apollo(
|
|
repo,
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
assert updateinfo == updateinfo_xml
|
|
break
|
|
|
|
# Gzip first
|
|
gzipped = await apollo_tree.gzip_updateinfo(updateinfo)
|
|
|
|
# Run write_updateinfo_to_file
|
|
updateinfo_file = await apollo_tree.write_updateinfo_to_file(
|
|
repos["BaseOS"][0]["found_path"],
|
|
gzipped,
|
|
)
|
|
|
|
assert updateinfo_file is not None
|
|
assert path.exists(updateinfo_file)
|
|
assert path.isfile(updateinfo_file)
|
|
|
|
# Run update_repomd_xml
|
|
# This will replace the repomd.xml file with the new one
|
|
mocker.patch("time.time", return_value=1674284973)
|
|
repomd_xml_path = repos["BaseOS"][0]["found_path"]
|
|
await apollo_tree.update_repomd_xml(
|
|
repomd_xml_path,
|
|
gzipped,
|
|
)
|
|
|
|
# Check that the repomd.xml file matches baseos__base__repomd__x86_64_with_updateinfo.xml from data
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__),
|
|
"data",
|
|
"baseos__base__repomd__x86_64_with_updateinfo.xml",
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
expected_repomd_xml = f.read()
|
|
|
|
with open(repomd_xml_path, "r", encoding="utf-8") as f:
|
|
actual_repomd_xml = f.read()
|
|
|
|
assert actual_repomd_xml == expected_repomd_xml
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_apollo_tree(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
mocker.patch("time.time", return_value=1674284973)
|
|
await apollo_tree.run_apollo_tree(
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
False,
|
|
True,
|
|
directory,
|
|
[],
|
|
"Rocky Linux 8 x86_64",
|
|
)
|
|
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
# Check that the repomd.xml file matches baseos__base__repomd__x86_64_with_updateinfo.xml from data
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__),
|
|
"data",
|
|
"baseos__base__repomd__x86_64_with_updateinfo.xml",
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
expected_repomd_xml = f.read()
|
|
|
|
with open(repo["found_path"], "r", encoding="utf-8") as f:
|
|
actual_repomd_xml = f.read()
|
|
|
|
assert actual_repomd_xml == expected_repomd_xml
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_apollo_tree_arch_in_product(mocker):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
repos = await _setup_test_baseos(directory)
|
|
|
|
# Read data/updateinfo__test__1.xml
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__), "data", "updateinfo__test__1.xml"
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
updateinfo_xml = f.read()
|
|
|
|
resp = MockResponse(updateinfo_xml, 200)
|
|
mocker.patch("aiohttp.ClientSession.get", return_value=resp)
|
|
|
|
mocker.patch("time.time", return_value=1674284973)
|
|
await apollo_tree.run_apollo_tree(
|
|
"$reponame/$arch/os/repodata/repomd.xml",
|
|
False,
|
|
True,
|
|
directory,
|
|
[],
|
|
"Rocky Linux 8 $arch",
|
|
)
|
|
|
|
for _, repo_variants in repos.items():
|
|
for repo in repo_variants:
|
|
# Check that the repomd.xml file matches baseos__base__repomd__x86_64_with_updateinfo.xml from data
|
|
with open(
|
|
path.join(
|
|
path.dirname(__file__),
|
|
"data",
|
|
"baseos__base__repomd__x86_64_with_updateinfo.xml",
|
|
),
|
|
"r",
|
|
encoding="utf-8",
|
|
) as f:
|
|
expected_repomd_xml = f.read()
|
|
|
|
with open(repo["found_path"], "r", encoding="utf-8") as f:
|
|
actual_repomd_xml = f.read()
|
|
|
|
assert actual_repomd_xml == expected_repomd_xml
|