-
Notifications
You must be signed in to change notification settings - Fork 180
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add unit tests for rpm metadata parsers
- Loading branch information
1 parent
f480cce
commit 43379fb
Showing
5 changed files
with
300 additions
and
38 deletions.
There are no files selected for viewing
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,54 +1,316 @@ | ||
# pylint: disable=missing-module-docstring | ||
import datetime | ||
import gzip | ||
import io | ||
import os.path | ||
from unittest.mock import patch, mock_open | ||
import unittest | ||
|
||
from lzreposync import Handler, RPMRepo | ||
from lzreposync.filelists_parser import FilelistsParser | ||
from lzreposync.primary_parser import ( | ||
PrimaryParser, | ||
complex_attrs, | ||
map_dependency_attribute, | ||
) | ||
from lzreposync.rpm_metadata_parser import MetadataParser | ||
|
||
|
||
def test_download_and_parse_metadata(): | ||
test_hash = "00c99a7e67dac325f1cbeeedddea3e4001a8f803c9bf43d9f50b5f1c756b0887" | ||
test_name = "Leap15" | ||
test_cache_dir = ".cache" | ||
test_repo = "https://download.opensuse.org/update/leap/15.5/oss/repodata/" | ||
cache_file = os.path.join("{}/{}".format(test_cache_dir, test_name), test_name + ".hash") | ||
test_rpm_handler = Handler() | ||
"""TODO""" | ||
|
||
with patch("builtins.open", mock_open()) as mocked_file: | ||
rpm_repo = RPMRepo(test_name, test_cache_dir, | ||
test_repo, | ||
test_rpm_handler) | ||
rpm_repo.get_packages_metadata() | ||
|
||
mocked_file.assert_called_once_with(cache_file, 'w') | ||
mocked_file().write.assert_called_once_with(test_hash) | ||
def test_verify_signature(): | ||
"""TODO""" | ||
|
||
|
||
def test_parse_primary_missing_element_attributes(): | ||
""" | ||
Testing the parsing with a primary.xml file of 2 packages, one of which have missing element attributes: | ||
'epoch' and the 'ver' of the second package are missing. | ||
The current parser's behaviour is to ignore the missing attribute, and continue parsing | ||
""" | ||
test_primary_path = "primary_test_missing_element_attributes.xml.gz" | ||
test_cache_dir = ".cache" | ||
handler = Handler() | ||
class LazyRepoSyncTest(unittest.TestCase): | ||
|
||
with open(test_primary_path, "rb") as primary_gz: | ||
file_obj = io.BytesIO(primary_gz.read()) | ||
def test_parse_primary(self): | ||
""" | ||
Test the parsing functionality for rpm's primary.xml metadata file | ||
""" | ||
|
||
rpm_repo = RPMRepo(None, test_cache_dir, None, handler) | ||
parsed_packages_count = rpm_repo.parse_metadata_file(file_obj) | ||
# Prepare test data: valid rpm package metadata in primary.xml | ||
primary_xml = """<?xml version="1.0" encoding="UTF-8"?> | ||
<metadata xmlns="http://linux.duke.edu/metadata/common" xmlns:rpm="http://linux.duke.edu/metadata/rpm" packages="1"> | ||
<package type="rpm"> | ||
<name>gstreamer-plugins-bad</name> | ||
<arch>aarch64</arch> | ||
<version epoch="0" ver="1.22.0" rel="lp155.3.4.1"/> | ||
<checksum type="sha256" pkgid="YES">5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b</checksum> | ||
<summary>GStreamer Streaming-Media Framework Plug-Ins</summary> | ||
<description>GStreamer is a streaming media framework based on graphs of filters | ||
that operate on media data... | ||
</description> | ||
<packager>http://bugs.opensuse.org</packager> | ||
<url>https://gstreamer.freedesktop.org</url> | ||
<time file="1700989109" build="1696852591"/> | ||
<size package="2197472" installed="11946385" archive="11965968"/> | ||
<location href="aarch64/gstreamer-plugins-bad-1.22.0-lp155.3.4.1.aarch64.rpm"/> | ||
<format> | ||
<rpm:license>LGPL-2.1-or-later</rpm:license> | ||
<rpm:vendor>openSUSE</rpm:vendor> | ||
<rpm:group>Productivity/Multimedia/Other</rpm:group> | ||
<rpm:buildhost>armbuild26</rpm:buildhost> | ||
<rpm:sourcerpm>gstreamer-plugins-bad-1.22.0-lp155.3.4.1.src.rpm</rpm:sourcerpm> | ||
<rpm:header-range start="6200" end="149568"/> | ||
<rpm:provides> | ||
<rpm:entry name="gst-plugins-bad" flags="EQ" epoch="0" ver="1.22.0"/> | ||
<rpm:entry name="gstreamer-plugins-bad" flags="EQ" epoch="0" ver="1.22.0" rel="lp155.3.4.1"/> | ||
</rpm:provides> | ||
<rpm:requires> | ||
<rpm:entry name="glib2-tools" pre="1"/> | ||
<rpm:entry name="glib2-tools"/> | ||
</rpm:requires> | ||
<rpm:obsoletes> | ||
<rpm:entry name="libgstvdpau" flags="LT" epoch="0" ver="1.18.0"/> | ||
</rpm:obsoletes> | ||
<rpm:enhances> | ||
<rpm:entry name="gstreamer"/> | ||
</rpm:enhances> | ||
</format> | ||
</package> | ||
</metadata> | ||
""" | ||
package_data = { | ||
"package_size": "2197472", | ||
"checksum": "5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b", | ||
"checksum_type": "sha256", | ||
"header_start": "6200", | ||
"header_end": "149568", | ||
"remote_path": "aarch64/gstreamer-plugins-bad-1.22.0-lp155.3.4.1.aarch64.rpm", | ||
} | ||
package_header_data = { | ||
"name": "gstreamer-plugins-bad", | ||
"arch": "aarch64", | ||
"version": "1.22.0", | ||
"release": "lp155.3.4.1", | ||
"epoch": "0", | ||
"summary": "GStreamer Streaming-Media Framework Plug-Ins", | ||
"description": "GStreamer is a streaming media framework based on graphs of filters\n\ | ||
that operate on media data...", | ||
"packager": "http://bugs.opensuse.org", | ||
"url": "https://gstreamer.freedesktop.org", | ||
"buildtime": datetime.datetime.fromtimestamp(float(1696852591)), | ||
"installed_size": "11946385", | ||
"archivesize": int(11965968), | ||
"license": "LGPL-2.1-or-later", | ||
"vendor": "openSUSE", | ||
"group": "Productivity/Multimedia/Other", | ||
"buildhost": "armbuild26", | ||
"sourcerpm": "gstreamer-plugins-bad-1.22.0-lp155.3.4.1.src.rpm", | ||
} | ||
|
||
assert parsed_packages_count == 2 | ||
compressed_primary = gzip.compress(bytes(primary_xml, "utf-8")) | ||
primary_xml_file_obj = io.BytesIO(compressed_primary) | ||
|
||
# Parse the test package md | ||
primary_parser = PrimaryParser(primary_xml_file_obj) | ||
parsed_packages = list(primary_parser.parse_primary()) | ||
|
||
def test_verify_signature(): | ||
""" | ||
Note: this test case assumes that the repomd.xml file hasn't been altered, if so it will fail | ||
""" | ||
test_repo = "https://download.opensuse.org/update/leap/15.5/oss/repodata/" | ||
test_cache_dir = ".cache" | ||
# Assertions | ||
test_package = parsed_packages[0] | ||
test_package_elements = test_package.keys() | ||
test_package_hdr_elements = test_package["header"].keys() | ||
self.assertEqual(len(parsed_packages), 1) | ||
|
||
# Check package data | ||
for elt, val in package_data.items(): | ||
self.assertTrue(elt in test_package_elements) | ||
self.assertEqual(test_package[elt], val) | ||
|
||
# Check package's header data | ||
for elt, val in package_header_data.items(): | ||
self.assertTrue(elt in test_package_hdr_elements) | ||
self.assertEqual(test_package["header"][elt], val) | ||
|
||
# Check package's capabilities | ||
for dep in complex_attrs: | ||
for attribute in ( | ||
("name", "version", "flags") | ||
if dep != "changelog" | ||
else ("name", "text", "time") | ||
): | ||
dep_mapped_name = map_dependency_attribute(dep, attribute) | ||
if dep == "provides" or dep == "requires": | ||
self.assertEqual(len(test_package["header"][dep_mapped_name]), 2) | ||
continue | ||
if dep == "obsoletes" or dep == "enhances": | ||
self.assertEqual(len(test_package["header"][dep_mapped_name]), 1) | ||
continue | ||
|
||
# Else: non-existing attributes | ||
self.assertEqual(len(test_package["header"][dep_mapped_name]), 0) | ||
|
||
def test_parse_file_lists(self): | ||
""" | ||
Test the parsing functionality for rpm's filelists.xml metadata file | ||
""" | ||
|
||
file_lists_xml = """<?xml version="1.0" encoding="UTF-8"?> | ||
<filelists xmlns="http://linux.duke.edu/metadata/filelists" packages="2"> | ||
<package pkgid="5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b" name="gstreamer-plugins-bad" arch="aarch64"> | ||
<version epoch="0" ver="1.1.0" rel="lp155.3.4.1"/> | ||
<file>/usr/lib64/gstreamer-1.0/libgstaiff.so</file> | ||
<file>/usr/lib64/gstreamer-1.0/libgstaccurip.so</file> | ||
<file>/usr/lib64/gstreamer-1.0/libgstadpcmdec.so</file> | ||
<file>/usr/lib64/gstreamer-1.0/libgstadpcmenc.so</file> | ||
</package> | ||
<package pkgid="5aac91b3ec4b358b22fe50cf0a23d6ea6139ca2f1909f99b6c3b5734ca12530f" name="gstreamer-plugins-bad" arch="ppc64le"> | ||
<version epoch="0" ver="2.2.0" rel="lp155.3.4.1"/> | ||
<file>/usr/lib64/gstreamer-1.0/libgstaccurip.so</file> | ||
<file>/usr/lib64/gstreamer-1.0/libgstadpcmdec.so</file> | ||
<file>/usr/lib64/gstreamer-1.0/libgstadpcmenc.so</file> | ||
</package> | ||
</filelists>""" | ||
compressed_file_lists = gzip.compress(bytes(file_lists_xml, "utf-8")) | ||
file_lists_xml_obj_file = io.BytesIO(compressed_file_lists) | ||
|
||
file_lists_parser = FilelistsParser(file_lists_xml_obj_file) | ||
file_lists_parser.parse_filelists() | ||
|
||
package_files_5f32 = file_lists_parser.get_package_filelist( | ||
"5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b" | ||
) | ||
package_files_5aac = file_lists_parser.get_package_filelist( | ||
"5aac91b3ec4b358b22fe50cf0a23d6ea6139ca2f1909f99b6c3b5734ca12530f" | ||
) | ||
|
||
# Delete the cached files | ||
file_lists_parser.clear_cache() | ||
|
||
# First package | ||
self.assertEqual( | ||
package_files_5f32["pkgid"], | ||
"5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b", | ||
) | ||
self.assertEqual(len(package_files_5f32["files"]), 4) | ||
self.assertEqual(package_files_5f32["version"], "1.1.0") | ||
self.assertEqual(package_files_5f32["epoch"], "0") | ||
self.assertEqual(package_files_5f32["release"], "lp155.3.4.1") | ||
self.assertEqual( | ||
package_files_5f32["files"][0], "/usr/lib64/gstreamer-1.0/libgstaiff.so" | ||
) | ||
|
||
# Second package | ||
self.assertEqual( | ||
package_files_5aac["pkgid"], | ||
"5aac91b3ec4b358b22fe50cf0a23d6ea6139ca2f1909f99b6c3b5734ca12530f", | ||
) | ||
self.assertEqual(len(package_files_5aac["files"]), 3) | ||
self.assertEqual(package_files_5aac["version"], "2.2.0") | ||
self.assertEqual(package_files_5aac["epoch"], "0") | ||
self.assertEqual(package_files_5aac["release"], "lp155.3.4.1") | ||
self.assertEqual( | ||
package_files_5aac["files"][0], "/usr/lib64/gstreamer-1.0/libgstaccurip.so" | ||
) | ||
|
||
def test_parse_metadata(self): | ||
"""TODO: both valid primary and filelists | ||
Create header with full information | ||
""" | ||
|
||
# primary & filelists test file | ||
primary_parser = PrimaryParser("test-files/primary-sample-2.xml.gz") | ||
file_lists_parser = FilelistsParser("test-files/filelists-sample-2.xml.gz") | ||
|
||
rpm_metadata_parser = MetadataParser(primary_parser, file_lists_parser) | ||
package_gen = rpm_metadata_parser.parse_packages_metadata() | ||
|
||
package_gen = rpm_metadata_parser.parse_packages_metadata() | ||
package1 = next(package_gen) | ||
package2 = next(package_gen) | ||
|
||
# First pacakge | ||
pkgid = package1["checksum"] | ||
self.assertEqual( | ||
pkgid, "5f32047b55c0ca2dcc00a00270cd0b10df4df40c6cd9355eeff9b6aa0997657b" | ||
) | ||
self.assertEqual(package1["header"]["name"], "gstreamer-plugins-bad") | ||
# files | ||
self.assertEqual(len(package1["header"]["filenames"]), 8) | ||
self.assertEqual( | ||
package1["header"]["filenames"][0], | ||
file_lists_parser.get_package_filelist(pkgid)["files"][0], | ||
) | ||
# capabilities | ||
self.assertEqual(len(package1["header"]["provides"]), 5) | ||
self.assertEqual(len(package1["header"]["provideversion"]), 5) | ||
self.assertEqual(len(package1["header"]["provideflags"]), 5) | ||
|
||
self.assertEqual(len(package1["header"]["requirename"]), 7) | ||
self.assertEqual(len(package1["header"]["requireversion"]), 7) | ||
self.assertEqual(len(package1["header"]["requireflags"]), 7) | ||
|
||
self.assertEqual(len(package1["header"][5055]), 1) # 'enhancesname' | ||
self.assertEqual(len(package1["header"][5056]), 1) # 'enhancesversion' | ||
self.assertEqual(len(package1["header"][5057]), 1) # 'enhancesflags' | ||
|
||
self.assertEqual(len(package1["header"]["obsoletename"]), 1) # 'enhancesflags' | ||
self.assertEqual( | ||
len(package1["header"]["obsoleteversion"]), 1 | ||
) # 'enhancesflags' | ||
self.assertEqual(len(package1["header"]["obsoleteflags"]), 1) # 'enhancesflags' | ||
|
||
# Second pacakge | ||
pkgid2 = package2["checksum"] | ||
self.assertEqual( | ||
pkgid2, "5aac91b3ec4b358b22fe50cf0a23d6ea6139ca2f1909f99b6c3b5734ca12530f" | ||
) | ||
|
||
file_lists_parser.clear_cache() | ||
|
||
def test_arch_filter_primary(self): | ||
""" | ||
Test the primary.xml parsing functionality with the 'arch' filter | ||
""" | ||
|
||
primary_gz_test = "test-files/primary-sample-10.xml.gz" | ||
archs = { | ||
"x86_64": 1, | ||
"aarch64": 2, | ||
"ppc64le": 4, | ||
"noarch": 3, | ||
"(x86_64|aarch64)": 3, | ||
"(ppc64le|noarch)": 7, | ||
".*": 10, | ||
} | ||
|
||
for arch, count in archs.items(): | ||
primary_parser = PrimaryParser(primary_gz_test, arch_filter=arch) | ||
parsed_packages = list(primary_parser.parse_primary()) | ||
self.assertEqual(len(parsed_packages), count) | ||
|
||
def test_arch_filter_file_lists(self): | ||
""" | ||
Test the filelists.xml parsing functionality with the 'arch' filter | ||
""" | ||
file_lists_gz_test = "test-files/filelists-sample-10.xml.gz" | ||
archs = { | ||
"aarch64": 2, | ||
"ppc64le": 2, | ||
"src": 2, | ||
"x86_64": 2, | ||
"s390x": 2, | ||
"(ppc64le|s390x|src)": 6, | ||
".*": 10, | ||
} | ||
|
||
for arch, count in archs.items(): | ||
file_lists_parser = FilelistsParser(file_lists_gz_test, arch_filter=arch) | ||
file_lists_parser.parse_filelists() | ||
self.assertEqual(file_lists_parser.num_parsed_packages, count) | ||
|
||
file_lists_parser.clear_cache() | ||
|
||
def test_arch_filter_packages(self): | ||
""" | ||
Test the parse metadata functionality (both primary and filelists) with the 'arch' filter | ||
""" | ||
|
||
def test_insert_batch_into_db(self): | ||
""" | ||
TODO: create a sample batch of package and verify it has been inserted correctly | ||
""" | ||
|
||
rpm_repo = RPMRepo(None, test_cache_dir, test_repo, None) | ||
|
||
assert rpm_repo.verify_signature() | ||
if __name__ == "__main__": | ||
unittest.main() |