| |
@@ -11,15 +11,21 @@
|
| |
# the full text of the license.
|
| |
|
| |
import json
|
| |
+ import os
|
| |
import re
|
| |
+ import subprocess
|
| |
+ import tempfile
|
| |
from datetime import datetime, timezone
|
| |
|
| |
import git
|
| |
import requests
|
| |
+ import yaml
|
| |
+ from git import Repo
|
| |
from pyrpkg import rpkgError
|
| |
from requests.exceptions import ConnectionError
|
| |
from configparser import NoOptionError, NoSectionError
|
| |
from urllib.parse import urlparse
|
| |
+ from xml.etree import ElementTree
|
| |
|
| |
|
| |
def query_bodhi(server_url, timeout=60):
|
| |
@@ -657,3 +663,185 @@
|
| |
raise rpkgError(base_error_msg.format(rv_error))
|
| |
|
| |
logger.info("Monitoring of the project was sucessfully disabled.")
|
| |
+
|
| |
+ def get_copackages_of_pkg(config, pkg_name, cli_name) -> list:
|
| |
+ """
|
| |
+ Getting list of packages provided by a package.
|
| |
+ :param config: config
|
| |
+ :param pkg_name: package name
|
| |
+ :param cli_name: cli name
|
| |
+ :return: list of package names
|
| |
+ """
|
| |
+ mdapi_url = f"https://mdapi.fedoraproject.org/rawhide/pkg/{pkg_name}"
|
| |
+ try:
|
| |
+ mdapi_url = config.get('{0}.mdapi'.format(cli_name),
|
| |
+ 'package_info',
|
| |
+ vars={'branch': 'rawhide',
|
| |
+ 'repo_name': pkg_name})
|
| |
+ except (ValueError, NoOptionError, NoSectionError) as e:
|
| |
+ raise rpkgError('Could not get mdapi endpoint for repository'
|
| |
+ '({0}): {1}.'.format(pkg_name, str(e)))
|
| |
+
|
| |
+ try:
|
| |
+ rv = requests.get(mdapi_url, timeout=60)
|
| |
+ except ConnectionError as error:
|
| |
+ error_msg = ('The connection to Mdapi failed.'
|
| |
+ ' The error was: {0}'.format(str(error)))
|
| |
+ raise rpkgError(error_msg)
|
| |
+
|
| |
+ if rv.status_code == 404:
|
| |
+ # release wasn't found
|
| |
+ return None
|
| |
+ elif not rv.ok:
|
| |
+ base_error_msg = ('The following error occurred while trying to '
|
| |
+ 'get the information about package on mdapi')
|
| |
+ raise rpkgError(base_error_msg.format(rv.text))
|
| |
+
|
| |
+ copackages = rv.json()["co-packages"]
|
| |
+ return copackages
|
| |
+
|
| |
+ def get_manifested_packages(fedora_version):
|
| |
+ """
|
| |
+ Getting list of packages that are listed in
|
| |
+ comps, or kickstarts, or workstation-ostree-config, or FCOS manifests, or lorax's lists.
|
| |
+ No need to get packages listed in kickstarts and lorax's manifests, because it
|
| |
+ use packages listed in comps.
|
| |
+
|
| |
+ :param fedora_version: fedora version
|
| |
+ :return: list of package names listed in manifests
|
| |
+ """
|
| |
+ manifested_packages = []
|
| |
+ with tempfile.TemporaryDirectory() as tmp_dir:
|
| |
+ # TODO write error handling for cloning repo
|
| |
+ print(f"Temporary directory created: {tmp_dir}")
|
| |
+
|
| |
+ # Creating subdirectories
|
| |
+ comps_dir = os.path.join(tmp_dir, "fedora-comps")
|
| |
+ workstation_ostree_dir = os.path.join(tmp_dir, "workstation-ostree-config")
|
| |
+ fcos_manifests_dir = os.path.join(tmp_dir, "fcos-manifests")
|
| |
+
|
| |
+ os.mkdir(comps_dir)
|
| |
+ os.mkdir(workstation_ostree_dir)
|
| |
+ os.mkdir(fcos_manifests_dir)
|
| |
+
|
| |
+ # getting comps packages
|
| |
+ comps_repo_url = "https://pagure.io/fedora-comps.git"
|
| |
+ Repo.clone_from(comps_repo_url, comps_dir)
|
| |
+
|
| |
+ file_name = "comps-{0}.xml.in".format(fedora_version)
|
| |
+ file_path = os.path.join(comps_dir, file_name)
|
| |
+
|
| |
+ if os.path.exists(file_path):
|
| |
+ with open(file_path, "r") as f:
|
| |
+ content = f.read()
|
| |
+ else:
|
| |
+ print(f"{file_path} does not exist in the repository.")
|
| |
+
|
| |
+ xml_parse = ElementTree.fromstring(content)
|
| |
+
|
| |
+ # find all packagereq
|
| |
+ package_requirements = xml_parse.findall(".//packagereq")
|
| |
+ comps_packages = [pkg.text for pkg in package_requirements]
|
| |
+
|
| |
+ manifested_packages.extend(comps_packages)
|
| |
+
|
| |
+ # getting workstation-ostree-config packages
|
| |
+ workstation_ostree_repo_url = "https://pagure.io/workstation-ostree-config.git"
|
| |
+ Repo.clone_from(workstation_ostree_repo_url, workstation_ostree_dir)
|
| |
+
|
| |
+ # Walk through the repo and list files
|
| |
+ file_list = os.listdir(workstation_ostree_dir)
|
| |
+
|
| |
+ files_with_packages_common = [file for file in file_list
|
| |
+ if file.endswith("common.yaml")]
|
| |
+ print("Files with packages common:")
|
| |
+ print(files_with_packages_common)
|
| |
+
|
| |
+ files_with_packages = [file for file in file_list
|
| |
+ if file.endswith("packages.yaml")]
|
| |
+ print("Files with packages:")
|
| |
+ print(files_with_packages)
|
| |
+
|
| |
+ workstation_ostree_packages = []
|
| |
+
|
| |
+ def get_packages_from_worksation_ostree(files):
|
| |
+ packages = []
|
| |
+ for file_name in files:
|
| |
+ file_path = os.path.join(workstation_ostree_dir, file_name)
|
| |
+ if os.path.exists(file_path):
|
| |
+ with open(file_path, "r") as f:
|
| |
+ content = yaml.safe_load(f)
|
| |
+ packages.extend(content["packages"])
|
| |
+ return packages
|
| |
+
|
| |
+ packages_from_common = get_packages_from_worksation_ostree(files_with_packages_common)
|
| |
+ packages_from_package_files = get_packages_from_worksation_ostree(files_with_packages)
|
| |
+
|
| |
+ workstation_ostree_packages.extend(packages_from_common)
|
| |
+ workstation_ostree_packages.extend(packages_from_package_files)
|
| |
+ workstation_ostree_packages_clean = list(set(workstation_ostree_packages))
|
| |
+
|
| |
+ workstation_ostree_unique_packages = []
|
| |
+
|
| |
+ for package in workstation_ostree_packages_clean:
|
| |
+ if package not in manifested_packages:
|
| |
+ workstation_ostree_unique_packages.append(package)
|
| |
+
|
| |
+ manifested_packages.extend(workstation_ostree_unique_packages)
|
| |
+
|
| |
+ # Getting packages from FCOS manifest
|
| |
+ fcos_repo_url = "https://github.com/coreos/fedora-coreos-config.git"
|
| |
+ Repo.clone_from(fcos_repo_url, fcos_manifests_dir)
|
| |
+
|
| |
+ folder_name = "manifests"
|
| |
+ folder_path = os.path.join(fcos_manifests_dir, folder_name)
|
| |
+ file_list = os.listdir(folder_path)
|
| |
+ yaml_files = [file for file in file_list if file.endswith(".yaml")]
|
| |
+
|
| |
+ fcos_manifest_packages = []
|
| |
+
|
| |
+ for file_name in yaml_files:
|
| |
+ file_path = os.path.join(folder_path, file_name)
|
| |
+ if os.path.exists(file_path):
|
| |
+ with open(file_path, "r") as f:
|
| |
+ content = yaml.safe_load(f)
|
| |
+ if "packages" in content.keys():
|
| |
+ fcos_manifest_packages.extend(content["packages"])
|
| |
+
|
| |
+
|
| |
+ fcos_manifest_packages_clean = list(set(fcos_manifest_packages))
|
| |
+
|
| |
+ fcos_manifest_unique_packages = []
|
| |
+
|
| |
+ for package in fcos_manifest_packages_clean:
|
| |
+ if package not in manifested_packages:
|
| |
+ fcos_manifest_unique_packages.append(package)
|
| |
+
|
| |
+ print(len(fcos_manifest_unique_packages))
|
| |
+
|
| |
+ manifested_packages.extend(fcos_manifest_unique_packages)
|
| |
+
|
| |
+ return manifested_packages
|
| |
+
|
| |
+ def dnf_repoquery(pkg_name, option):
|
| |
+ """
|
| |
+ Getting list of packages by repoquery request.
|
| |
+ :param pkg_name: a string of the repository name
|
| |
+ :param option: a string of the option for requests
|
| |
+ :return: a list of package names
|
| |
+ """
|
| |
+ try:
|
| |
+ # Run the dnf repoquery command
|
| |
+ result = subprocess.run(
|
| |
+ ['dnf', 'repoquery', option, pkg_name],
|
| |
+ stdout=subprocess.PIPE,
|
| |
+ stderr=subprocess.PIPE,
|
| |
+ text=True,
|
| |
+ check=True
|
| |
+ )
|
| |
+ # Split the output into lines and return as a list
|
| |
+ provided_packages = result.stdout.strip().split('\n')
|
| |
+ return provided_packages
|
| |
+ except subprocess.CalledProcessError as e:
|
| |
+ print(f"Error: {e.stderr.strip()}")
|
| |
+ return []
|
| |
amedvede
commented 3 months ago
I added the util function that gets list of branches provided by package, by the idea it will be called to get tree of dependencies and notify user about all packages affected.