@@ 350,8 350,7 @@ class BaseMakerBase(abc.ABC):
with importlib.resources.as_file(path) as fp:
LOG.debug("Reading %s", fp)
self._read_repofile_new(fp)
- LOG.debug("Enabling repos: %s", release.repos)
- self.enable_repos(release.repos)
+ release.repog.load(self, release.config, release)
@abc.abstractmethod
def create_repo(self, repoid: str, **kwargs) -> None:
@@ 17,19 17,19 @@ from enum import auto as auto_enum
from importlib.abc import Traversable
from pathlib import Path
-from fedrq._compat import StrEnum
-from fedrq.backends.base import BaseMakerBase
-
if sys.version_info < (3, 11):
import tomli as tomllib
else:
import tomllib
-from pydantic import BaseModel, Field, validator
+from pydantic import BaseModel, Field, PrivateAttr, validator
+from fedrq._compat import StrEnum
from fedrq._config import ConfigError
from fedrq._utils import merge_dict, mklog
from fedrq.backends import BACKENDS, get_default_backend
+from fedrq.backends.base import BaseMakerBase
+from fedrq.release_repo import AliasRepoG, DefaultRepoGs, RepoG, Repos
if t.TYPE_CHECKING:
import dnf
@@ 72,6 72,18 @@ class ReleaseConfig(BaseModel):
copr_chroot_fmt: t.Optional[str] = None
full_def_paths: t.ClassVar[list[t.Union[Traversable, Path]]] = []
+ repo_aliases: dict[str, str] = {}
+ _repogs = PrivateAttr()
+
+ @property
+ def repogs(self) -> Repos:
+ return self._repogs
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self._repogs = DefaultRepoGs.new(
+ self.defs | AliasRepoG.from_str_mapping(self.repo_aliases)
+ )
@validator("defpaths")
def v_defpaths(cls, value, values) -> dict[str, t.Any]:
@@ 98,10 110,12 @@ class ReleaseConfig(BaseModel):
return bool(re.match(self.matcher, val))
def is_valid_repo(self, val: str) -> bool:
- return val in self.defs
-
- def release(self, branch: str, repo_name: str = "base") -> Release:
- return Release(self, branch, repo_name)
+ try:
+ self.repogs.get_repo(val)
+ except ConfigError:
+ return False
+ else:
+ return True
@staticmethod
def _repo_dir_iterator(
@@ 161,32 175,40 @@ class ReleaseConfig(BaseModel):
return full_defpaths
def get_release(
- self, config: RQConfig, branch: str, repo_name: str = "base" # noqa: ARG002
+ self, config: RQConfig, branch: str, repo_name: str = "base"
) -> Release:
- return Release(release_config=self, branch=branch, repo_name=repo_name)
+ return Release(
+ config=config, release_config=self, branch=branch, repo_name=repo_name
+ )
class Release:
+ """
+ Encapsulates a ReleaseConfig with a specific version and repo name.
+ This SHOULD NOT be instantiated directly.
+ The __init__() has no stability promises.
+ Use the RQConfig.get_config() factory instead.
+ """
+
def __init__(
self,
+ config: RQConfig,
release_config: ReleaseConfig,
branch: str,
repo_name: str = "base",
) -> None:
+ self.config = config
self.release_config = release_config
if not self.release_config.is_match(branch):
raise ConfigError(
f"Branch {branch} does not match {self.release_config.name}"
)
- if not self.release_config.is_valid_repo(repo_name):
- raise ConfigError(
- "{repo} is not a valid repo type for {name}".format(
- repo=repo_name, name=self.release_config.name
- )
- + " Valid repos are: {}".format(tuple(release_config.defs))
- )
self.branch = branch
self.repo_name = repo_name
+ self.repog: RepoG = self.get_repog(repo_name)
+
+ def get_repog(self, key: str) -> RepoG:
+ return self.release_config.repogs.get_repo(key)
@property
def version(self) -> str:
@@ 195,10 217,6 @@ class Release:
raise ValueError(f"{self.branch} does not match {self.release_config.name}")
@property
- def repos(self) -> tuple[str, ...]:
- return tuple(self.release_config.defs[self.repo_name])
-
- @property
def copr_chroot_fmt(self) -> str | None:
return self.release_config.copr_chroot_fmt
@@ 31,6 31,7 @@ def test_checkconfig_dump(run_command2, patch_config_dirs):
defs = {"base": ["testrepo1"]}
expected = {
"matcher": "^(tester)$",
+ "repo_aliases": {},
"defpaths": ["testrepo1.repo"],
"system_repos": False,
"defs": defs,
@@ 9,20 9,11 @@ from fedrq.backends.base import PackageCompat, PackageQueryCompat, RepoqueryBase
def test_make_base_rawhide_repos() -> None:
config = rqconfig.get_config()
rawhide = config.get_release("rawhide")
- base = rawhide.make_base(config, fill_sack=False)
- backend: str = config.backend_mod.BACKEND
- if backend == "dnf":
- assert len(tuple(base.repos.iter_enabled())) == len(rawhide.repos)
- assert set(repo.id for repo in base.repos.iter_enabled()) == set(rawhide.repos)
- elif backend == "libdnf5":
- import libdnf5
-
- repoq = libdnf5.repo.RepoQuery(base)
- repoq.filter_enabled(True)
- assert len(tuple(repoq)) == len(rawhide.repos)
- assert set(repo.get_id() for repo in repoq) == set(rawhide.repos)
- else:
- raise TypeError
+ bm = config.backend_mod.BaseMaker()
+ base = rawhide.make_base(config, fill_sack=False, base_maker=bm) # noqa: F841
+ repos = bm.repolist(True)
+ assert len(repos) == len(rawhide.repog.repos)
+ assert set(repos) == set(rawhide.repog.repos)
def test_package_protocol(repo_test_rq: RepoqueryBase):