arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-04-24 07:17:17 +00:00
Code Issues Projects Releases Wiki Activity

split application class into traits

Browse Source
This commit is contained in:
Evgenii Alekseev 2021-10-21 03:10:49 +03:00
parent 4d4e5e9dde
commit 35df354c14
24 changed files with 1019 additions and 702 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docs/architecture.md
View File

@ -12,7 +12,11 @@ Full dependency diagram:
## `ahriman.application` package ## `ahriman.application` package
This package contains application (aka executable) related classes and everything for that. It also contains package called `ahriman.application.handlers` in which all available subcommands are described as separated classes derived from base `ahriman.application.handlers.handler.Handler` class. `ahriman.application.ahriman` contains only command line parses and executes specified `Handler` on success, `ahriman.application.application.Application` is a god class which provides interfaces for all repository related actions. `ahriman.application.lock.Lock` is additional class which provides file-based lock and also performs some common checks. This package contains application (aka executable) related classes and everything for that. It also contains package called `ahriman.application.handlers` in which all available subcommands are described as separated classes derived from base `ahriman.application.handlers.handler.Handler` class.
`ahriman.application.application.application.Application` (god class) is used for any interaction from parsers with repository, web etc. It is divided into multiple traits by functions (package related and repository related) in the same package.
`ahriman.application.ahriman` contains only command line parses and executes specified `Handler` on success, `ahriman.application.lock.Lock` is additional class which provides file-based lock and also performs some common checks.
## `ahriman.core` package ## `ahriman.core` package

12
src/ahriman/application/ahriman.py
View File

@ -295,12 +295,12 @@ def _set_repo_clean_parser(root: SubParserAction) -> argparse.ArgumentParser:
"you should not run this command manually. Also in case if you are going to clear " "you should not run this command manually. Also in case if you are going to clear "
"the chroot directories you will need root privileges.", "the chroot directories you will need root privileges.",
formatter_class=_formatter) formatter_class=_formatter)
parser.add_argument("--no-build", help="do not clear directory with package sources", action="store_true") parser.add_argument("--build", help="clear directory with package sources", action="store_true")
parser.add_argument("--no-cache", help="do not clear directory with package caches", action="store_true") parser.add_argument("--cache", help="clear directory with package caches", action="store_true")
parser.add_argument("--no-chroot", help="do not clear build chroot", action="store_true") parser.add_argument("--chroot", help="clear build chroot", action="store_true")
parser.add_argument("--no-manual", help="do not clear directory with manually added packages", action="store_true") parser.add_argument("--manual", help="clear directory with manually added packages", action="store_true")
parser.add_argument("--no-packages", help="do not clear directory with built packages", action="store_true") parser.add_argument("--packages", help="clear directory with built packages", action="store_true")
parser.add_argument("--no-patches", help="do not clear directory with patches", action="store_true") parser.add_argument("--patches", help="clear directory with patches", action="store_true")
parser.set_defaults(handler=handlers.Clean, quiet=True, unsafe=True) parser.set_defaults(handler=handlers.Clean, quiet=True, unsafe=True)
return parser return parser

281
src/ahriman/application/application.py
View File

@ -1,281 +0,0 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
import requests
import shutil
from pathlib import Path
from typing import Callable, Iterable, List, Set
from ahriman.core.build_tools.sources import Sources
from ahriman.core.configuration import Configuration
from ahriman.core.repository.repository import Repository
from ahriman.core.tree import Tree
from ahriman.core.util import package_like
from ahriman.models.package import Package
from ahriman.models.package_source import PackageSource
class Application:
"""
base application class
:ivar architecture: repository architecture
:ivar configuration: configuration instance
:ivar logger: application logger
:ivar repository: repository instance
"""
def __init__(self, architecture: str, configuration: Configuration, no_report: bool) -> None:
"""
default constructor
:param architecture: repository architecture
:param configuration: configuration instance
:param no_report: force disable reporting
"""
self.logger = logging.getLogger("root")
self.configuration = configuration
self.architecture = architecture
self.repository = Repository(architecture, configuration, no_report)
def _finalize(self, built_packages: Iterable[Package]) -> None:
"""
generate report and sync to remote server
"""
self.report([], built_packages)
self.sync([], built_packages)
def _known_packages(self) -> Set[str]:
"""
load packages from repository and pacman repositories
:return: list of known packages
"""
known_packages: Set[str] = set()
# local set
for base in self.repository.packages():
for package, properties in base.packages.items():
known_packages.add(package)
known_packages.update(properties.provides)
known_packages.update(self.repository.pacman.all_packages())
return known_packages
def get_updates(self, filter_packages: List[str], no_aur: bool, no_manual: bool, no_vcs: bool,
log_fn: Callable[[str], None]) -> List[Package]:
"""
get list of packages to run update process
:param filter_packages: do not check every package just specified in the list
:param no_aur: do not check for aur updates
:param no_manual: do not check for manual updates
:param no_vcs: do not check VCS packages
:param log_fn: logger function to log updates
:return: list of out-of-dated packages
"""
updates = []
if not no_aur:
updates.extend(self.repository.updates_aur(filter_packages, no_vcs))
if not no_manual:
updates.extend(self.repository.updates_manual())
for package in updates:
log_fn(f"{package.base} = {package.version}")
return updates
def add(self, names: Iterable[str], source: PackageSource, without_dependencies: bool) -> None:
"""
add packages for the next build
:param names: list of package bases to add
:param source: package source to add
:param without_dependencies: if set, dependency check will be disabled
"""
known_packages = self._known_packages()
aur_url = self.configuration.get("alpm", "aur_url")
def add_archive(src: Path) -> None:
dst = self.repository.paths.packages / src.name
shutil.copy(src, dst)
def add_aur(src: str) -> Path:
package = Package.load(src, self.repository.pacman, aur_url)
Sources.load(self.repository.paths.manual_for(package.base), package.git_url,
self.repository.paths.patches_for(package.base))
return self.repository.paths.manual_for(package.base)
def add_directory(path: Path) -> None:
for full_path in filter(package_like, path.iterdir()):
add_archive(full_path)
def add_local(path: Path) -> Path:
package = Package.load(path, self.repository.pacman, aur_url)
cache_dir = self.repository.paths.cache_for(package.base)
shutil.copytree(path, cache_dir) # copy package to store in caches
Sources.init(cache_dir) # we need to run init command in directory where we do have permissions
shutil.copytree(cache_dir, self.repository.paths.manual_for(package.base)) # copy package for the build
return self.repository.paths.manual_for(package.base)
def add_remote(src: str) -> None:
dst = self.repository.paths.packages / Path(src).name # URL is path, is not it?
response = requests.get(src, stream=True)
response.raise_for_status()
with dst.open("wb") as local_file:
for chunk in response.iter_content(chunk_size=1024):
local_file.write(chunk)
def process_dependencies(path: Path) -> None:
if without_dependencies:
return
dependencies = Package.dependencies(path)
self.add(dependencies.difference(known_packages), PackageSource.AUR, without_dependencies)
def process_single(src: str) -> None:
resolved_source = source.resolve(src)
if resolved_source == PackageSource.Archive:
add_archive(Path(src))
elif resolved_source == PackageSource.AUR:
path = add_aur(src)
process_dependencies(path)
elif resolved_source == PackageSource.Directory:
add_directory(Path(src))
elif resolved_source == PackageSource.Local:
path = add_local(Path(src))
process_dependencies(path)
elif resolved_source == PackageSource.Remote:
add_remote(src)
for name in names:
process_single(name)
def clean(self, no_build: bool, no_cache: bool, no_chroot: bool, no_manual: bool, no_packages: bool,
no_patches: bool) -> None:
"""
run all clean methods. Warning: some functions might not be available under non-root
:param no_build: do not clear directory with package sources
:param no_cache: do not clear directory with package caches
:param no_chroot: do not clear build chroot
:param no_manual: do not clear directory with manually added packages
:param no_packages: do not clear directory with built packages
:param no_patches: do not clear directory with patches
"""
if not no_build:
self.repository.clear_build()
if not no_cache:
self.repository.clear_cache()
if not no_chroot:
self.repository.clear_chroot()
if not no_manual:
self.repository.clear_manual()
if not no_packages:
self.repository.clear_packages()
if not no_patches:
self.repository.clear_patches()
def remove(self, names: Iterable[str]) -> None:
"""
remove packages from repository
:param names: list of packages (either base or name) to remove
"""
self.repository.process_remove(names)
self._finalize([])
def report(self, target: Iterable[str], built_packages: Iterable[Package]) -> None:
"""
generate report
:param target: list of targets to run (e.g. html)
:param built_packages: list of packages which has just been built
"""
targets = target or None
self.repository.process_report(targets, built_packages)
def sign(self, packages: Iterable[str]) -> None:
"""
sign packages and repository
:param packages: only sign specified packages
"""
# copy to prebuilt directory
for package in self.repository.packages():
# no one requested this package
if packages and package.base not in packages:
continue
for archive in package.packages.values():
if archive.filepath is None:
self.logger.warning("filepath is empty for %s", package.base)
continue # avoid mypy warning
src = self.repository.paths.repository / archive.filepath
dst = self.repository.paths.packages / archive.filepath
shutil.copy(src, dst)
# run generic update function
self.update([])
# sign repository database if set
self.repository.sign.process_sign_repository(self.repository.repo.repo_path)
self._finalize([])
def sync(self, target: Iterable[str], built_packages: Iterable[Package]) -> None:
"""
sync to remote server
:param target: list of targets to run (e.g. s3)
:param built_packages: list of packages which has just been built
"""
targets = target or None
self.repository.process_sync(targets, built_packages)
def unknown(self) -> List[Package]:
"""
get packages which were not found in AUR
:return: unknown package list
"""
def has_aur(package_base: str, aur_url: str) -> bool:
try:
_ = Package.from_aur(package_base, aur_url)
except Exception:
return False
return True
def has_local(package_base: str) -> bool:
cache_dir = self.repository.paths.cache_for(package_base)
return cache_dir.is_dir() and not Sources.has_remotes(cache_dir)
return [
package
for package in self.repository.packages()
if not has_aur(package.base, package.aur_url) and not has_local(package.base)
]
def update(self, updates: Iterable[Package]) -> None:
"""
run package updates
:param updates: list of packages to update
"""
def process_update(paths: Iterable[Path]) -> None:
if not paths:
return # don't need to process if no update supplied
updated = [Package.load(path, self.repository.pacman, self.repository.aur_url) for path in paths]
self.repository.process_update(paths)
self._finalize(updated)
# process built packages
packages = self.repository.packages_built()
process_update(packages)
# process manual packages
tree = Tree.load(updates, self.repository.paths)
for num, level in enumerate(tree.levels()):
self.logger.info("processing level #%i %s", num, [package.base for package in level])
packages = self.repository.process_build(level)
process_update(packages)

20
src/ahriman/application/application/__init__.py Normal file
View File

@ -0,0 +1,20 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from ahriman.application.application.application import Application

51
src/ahriman/application/application/application.py Normal file
View File

@ -0,0 +1,51 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from typing import Iterable, Set
from ahriman.application.application.packages import Packages
from ahriman.application.application.repository import Repository
from ahriman.models.package import Package
class Application(Packages, Repository):
"""
base application class
"""
def _finalize(self, built_packages: Iterable[Package]) -> None:
"""
generate report and sync to remote server
"""
self.report([], built_packages)
self.sync([], built_packages)
def _known_packages(self) -> Set[str]:
"""
load packages from repository and pacman repositories
:return: list of known packages
"""
known_packages: Set[str] = set()
# local set
for base in self.repository.packages():
for package, properties in base.packages.items():
known_packages.add(package)
known_packages.update(properties.provides)
known_packages.update(self.repository.pacman.all_packages())
return known_packages

148
src/ahriman/application/application/packages.py Normal file
View File

@ -0,0 +1,148 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import requests
import shutil
from pathlib import Path
from typing import Any, Iterable, Set
from ahriman.application.application.properties import Properties
from ahriman.core.build_tools.sources import Sources
from ahriman.core.util import package_like
from ahriman.models.package import Package
from ahriman.models.package_source import PackageSource
class Packages(Properties):
"""
package control class
"""
def _finalize(self, built_packages: Iterable[Package]) -> None:
"""
generate report and sync to remote server
"""
raise NotImplementedError
def _known_packages(self) -> Set[str]:
"""
load packages from repository and pacman repositories
:return: list of known packages
"""
raise NotImplementedError
def _add_archive(self, source: str, *_: Any) -> None:
"""
add package from archive
:param source: path to package archive
"""
local_path = Path(source)
dst = self.repository.paths.packages / local_path.name
shutil.copy(local_path, dst)
def _add_aur(self, source: str, known_packages: Set[str], without_dependencies: bool) -> None:
"""
add package from AUR
:param source: package base name
:param known_packages: list of packages which are known by the service
:param without_dependencies: if set, dependency check will be disabled
"""
aur_url = self.configuration.get("alpm", "aur_url")
package = Package.load(source, self.repository.pacman, aur_url)
Sources.load(self.repository.paths.manual_for(package.base), package.git_url,
self.repository.paths.patches_for(package.base))
local_path = self.repository.paths.manual_for(package.base)
self._process_dependencies(local_path, known_packages, without_dependencies)
def _add_directory(self, source: str, *_: Any) -> None:
"""
add packages from directory
:param source: path to local directory
"""
local_path = Path(source)
for full_path in filter(package_like, local_path.iterdir()):
self._add_archive(str(full_path))
def _add_local(self, source: str, known_packages: Set[str], without_dependencies: bool) -> None:
"""
add package from local PKGBUILDs
:param source: path to directory with local source files
:param known_packages: list of packages which are known by the service
:param without_dependencies: if set, dependency check will be disabled
"""
local_path = Path(source)
aur_url = self.configuration.get("alpm", "aur_url")
package = Package.load(local_path, self.repository.pacman, aur_url)
cache_dir = self.repository.paths.cache_for(package.base)
shutil.copytree(local_path, cache_dir) # copy package to store in caches
Sources.init(cache_dir) # we need to run init command in directory where we do have permissions
dst = self.repository.paths.manual_for(package.base)
shutil.copytree(cache_dir, dst) # copy package for the build
self._process_dependencies(dst, known_packages, without_dependencies)
def _add_remote(self, source: str, *_: Any) -> None:
"""
add package from remote sources (e.g. HTTP)
:param remote_url: remote URL to the package archive
"""
dst = self.repository.paths.packages / Path(source).name # URL is path, is not it?
response = requests.get(source, stream=True)
response.raise_for_status()
with dst.open("wb") as local_file:
for chunk in response.iter_content(chunk_size=1024):
local_file.write(chunk)
def _process_dependencies(self, local_path: Path, known_packages: Set[str], without_dependencies: bool) -> None:
"""
process package dependencies
:param local_path: path to local package sources (i.e. cloned AUR repository)
:param known_packages: list of packages which are known by the service
:param without_dependencies: if set, dependency check will be disabled
"""
if without_dependencies:
return
dependencies = Package.dependencies(local_path)
self.add(dependencies.difference(known_packages), PackageSource.AUR, without_dependencies)
def add(self, names: Iterable[str], source: PackageSource, without_dependencies: bool) -> None:
"""
add packages for the next build
:param names: list of package bases to add
:param source: package source to add
:param without_dependencies: if set, dependency check will be disabled
"""
known_packages = self._known_packages() # speedup dependencies processing
for name in names:
resolved_source = source.resolve(name)
fn = getattr(self, f"_add_{resolved_source.value}")
fn(name, known_packages, without_dependencies)
def remove(self, names: Iterable[str]) -> None:
"""
remove packages from repository
:param names: list of packages (either base or name) to remove
"""
self.repository.process_remove(names)
self._finalize([])

45
src/ahriman/application/application/properties.py Normal file
View File

@ -0,0 +1,45 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
from ahriman.core.configuration import Configuration
from ahriman.core.repository import Repository
class Properties:
"""
application base properties class
:ivar architecture: repository architecture
:ivar configuration: configuration instance
:ivar logger: application logger
:ivar repository: repository instance
"""
def __init__(self, architecture: str, configuration: Configuration, no_report: bool) -> None:
"""
default constructor
:param architecture: repository architecture
:param configuration: configuration instance
:param no_report: force disable reporting
"""
self.logger = logging.getLogger("root")
self.configuration = configuration
self.architecture = architecture
self.repository = Repository(architecture, configuration, no_report)

172
src/ahriman/application/application/repository.py Normal file
View File

@ -0,0 +1,172 @@
#
# Copyright (c) 2021 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import shutil
from pathlib import Path
from typing import Callable, Iterable, List
from ahriman.application.application.properties import Properties
from ahriman.core.build_tools.sources import Sources
from ahriman.core.tree import Tree
from ahriman.models.package import Package
class Repository(Properties):
"""
repository control class
"""
def _finalize(self, built_packages: Iterable[Package]) -> None:
"""
generate report and sync to remote server
"""
raise NotImplementedError
def clean(self, build: bool, cache: bool, chroot: bool, manual: bool, packages: bool, patches: bool) -> None:
"""
run all clean methods. Warning: some functions might not be available under non-root
:param build: clear directory with package sources
:param cache: clear directory with package caches
:param chroot: clear build chroot
:param manual: clear directory with manually added packages
:param packages: clear directory with built packages
:param patches: clear directory with patches
"""
if build:
self.repository.clear_build()
if cache:
self.repository.clear_cache()
if chroot:
self.repository.clear_chroot()
if manual:
self.repository.clear_manual()
if packages:
self.repository.clear_packages()
if patches:
self.repository.clear_patches()
def report(self, target: Iterable[str], built_packages: Iterable[Package]) -> None:
"""
generate report
:param target: list of targets to run (e.g. html)
:param built_packages: list of packages which has just been built
"""
targets = target or None
self.repository.process_report(targets, built_packages)
def sign(self, packages: Iterable[str]) -> None:
"""
sign packages and repository
:param packages: only sign specified packages
"""
# copy to prebuilt directory
for package in self.repository.packages():
# no one requested this package
if packages and package.base not in packages:
continue
for archive in package.packages.values():
if archive.filepath is None:
self.logger.warning("filepath is empty for %s", package.base)
continue # avoid mypy warning
src = self.repository.paths.repository / archive.filepath
dst = self.repository.paths.packages / archive.filepath
shutil.copy(src, dst)
# run generic update function
self.update([])
# sign repository database if set
self.repository.sign.process_sign_repository(self.repository.repo.repo_path)
self._finalize([])
def sync(self, target: Iterable[str], built_packages: Iterable[Package]) -> None:
"""
sync to remote server
:param target: list of targets to run (e.g. s3)
:param built_packages: list of packages which has just been built
"""
targets = target or None
self.repository.process_sync(targets, built_packages)
def unknown(self) -> List[Package]:
"""
get packages which were not found in AUR
:return: unknown package list
"""
def has_aur(package_base: str, aur_url: str) -> bool:
try:
_ = Package.from_aur(package_base, aur_url)
except Exception:
return False
return True
def has_local(package_base: str) -> bool:
cache_dir = self.repository.paths.cache_for(package_base)
return cache_dir.is_dir() and not Sources.has_remotes(cache_dir)
return [
package
for package in self.repository.packages()
if not has_aur(package.base, package.aur_url) and not has_local(package.base)
]
def update(self, updates: Iterable[Package]) -> None:
"""
run package updates
:param updates: list of packages to update
"""
def process_update(paths: Iterable[Path]) -> None:
if not paths:
return # don't need to process if no update supplied
updated = [Package.load(path, self.repository.pacman, self.repository.aur_url) for path in paths]
self.repository.process_update(paths)
self._finalize(updated)
# process built packages
packages = self.repository.packages_built()
process_update(packages)
# process manual packages
tree = Tree.load(updates, self.repository.paths)
for num, level in enumerate(tree.levels()):
self.logger.info("processing level #%i %s", num, [package.base for package in level])
packages = self.repository.process_build(level)
process_update(packages)
def updates(self, filter_packages: Iterable[str], no_aur: bool, no_manual: bool, no_vcs: bool,
log_fn: Callable[[str], None]) -> List[Package]:
"""
get list of packages to run update process
:param filter_packages: do not check every package just specified in the list
:param no_aur: do not check for aur updates
:param no_manual: do not check for manual updates
:param no_vcs: do not check VCS packages
:param log_fn: logger function to log updates
:return: list of out-of-dated packages
"""
updates = []
if not no_aur:
updates.extend(self.repository.updates_aur(filter_packages, no_vcs))
if not no_manual:
updates.extend(self.repository.updates_manual())
for package in updates:
log_fn(f"{package.base} = {package.version}")
return updates

2
src/ahriman/application/handlers/add.py
View File

@ -46,5 +46,5 @@ class Add(Handler):
if not args.now: if not args.now:
return return
packages = application.get_updates(args.package, True, False, True, application.logger.info) packages = application.updates(args.package, True, False, True, application.logger.info)
application.update(packages) application.update(packages)

2
src/ahriman/application/handlers/clean.py
View File

@ -42,4 +42,4 @@ class Clean(Handler):
:param no_report: force disable reporting :param no_report: force disable reporting
""" """
Application(architecture, configuration, no_report).clean( Application(architecture, configuration, no_report).clean(
args.no_build, args.no_cache, args.no_chroot, args.no_manual, args.no_packages, args.no_patches) args.build, args.cache, args.chroot, args.manual, args.packages, args.patches)

4
src/ahriman/application/handlers/update.py
View File

@ -42,8 +42,8 @@ class Update(Handler):
:param no_report: force disable reporting :param no_report: force disable reporting
""" """
application = Application(architecture, configuration, no_report) application = Application(architecture, configuration, no_report)
packages = application.get_updates(args.package, args.no_aur, args.no_manual, args.no_vcs, packages = application.updates(args.package, args.no_aur, args.no_manual, args.no_vcs,
Update.log_fn(application, args.dry_run)) Update.log_fn(application, args.dry_run))
if args.dry_run: if args.dry_run:
return return

1
src/ahriman/core/repository/__init__.py
View File

@ -17,3 +17,4 @@
# You should have received a copy of the GNU General Public License # You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
# #
from ahriman.core.repository.repository import Repository

2
src/ahriman/core/status/watcher.py
View File

@ -25,7 +25,7 @@ from typing import Any, Dict, List, Optional, Tuple
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import UnknownPackage from ahriman.core.exceptions import UnknownPackage
from ahriman.core.repository.repository import Repository from ahriman.core.repository import Repository
from ahriman.models.build_status import BuildStatus, BuildStatusEnum from ahriman.models.build_status import BuildStatus, BuildStatusEnum
from ahriman.models.package import Package from ahriman.models.package import Package

44
tests/ahriman/application/application/conftest.py Normal file
View File

@ -0,0 +1,44 @@
import pytest
from pytest_mock import MockerFixture
from ahriman.application.application.packages import Packages
from ahriman.application.application.properties import Properties
from ahriman.application.application.repository import Repository
from ahriman.core.configuration import Configuration
@pytest.fixture
def application_packages(configuration: Configuration, mocker: MockerFixture) -> Packages:
"""
fixture for application with package functions
:param configuration: configuration fixture
:param mocker: mocker object
:return: application test instance
"""
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
return Packages("x86_64", configuration, no_report=True)
@pytest.fixture
def application_properties(configuration: Configuration, mocker: MockerFixture) -> Properties:
"""
fixture for application with properties only
:param configuration: configuration fixture
:param mocker: mocker object
:return: application test instance
"""
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
return Properties("x86_64", configuration, no_report=True)
@pytest.fixture
def application_repository(configuration: Configuration, mocker: MockerFixture) -> Repository:
"""
fixture for application with repository functions
:param configuration: configuration fixture
:param mocker: mocker object
:return: application test instance
"""
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
return Repository("x86_64", configuration, no_report=True)

26
tests/ahriman/application/application/test_application.py Normal file
View File

@ -0,0 +1,26 @@
from pytest_mock import MockerFixture
from ahriman.application.application import Application
from ahriman.models.package import Package
def test_finalize(application: Application, mocker: MockerFixture) -> None:
"""
must report and sync at the last
"""
report_mock = mocker.patch("ahriman.application.application.Application.report")
sync_mock = mocker.patch("ahriman.application.application.Application.sync")
application._finalize([])
report_mock.assert_called_once()
sync_mock.assert_called_once()
def test_known_packages(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return not empty list of known packages
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
packages = application._known_packages()
assert len(packages) > 1
assert package_ahriman.base in packages

207
tests/ahriman/application/application/test_application_packages.py Normal file
View File

@ -0,0 +1,207 @@
import pytest
from pathlib import Path
from pytest_mock import MockerFixture
from unittest import mock
from unittest.mock import MagicMock
from ahriman.application.application.packages import Packages
from ahriman.models.package import Package
from ahriman.models.package_description import PackageDescription
from ahriman.models.package_source import PackageSource
def test_finalize(application_packages: Packages) -> None:
"""
must raise NotImplemented for missing finalize method
"""
with pytest.raises(NotImplementedError):
application_packages._finalize([])
def test_known_packages(application_packages: Packages) -> None:
"""
must raise NotImplemented for missing finalize method
"""
with pytest.raises(NotImplementedError):
application_packages._known_packages()
def test_add_archive(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from archive
"""
copy_mock = mocker.patch("shutil.copy")
application_packages._add_archive(package_ahriman.base)
copy_mock.assert_called_once()
def test_add_aur(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from AUR
"""
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
load_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.load")
dependencies_mock = mocker.patch("ahriman.application.application.packages.Packages._process_dependencies")
application_packages._add_aur(package_ahriman.base, set(), False)
load_mock.assert_called_once()
dependencies_mock.assert_called_once()
def test_add_directory(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add packages from directory
"""
iterdir_mock = mocker.patch("pathlib.Path.iterdir",
return_value=[package.filepath for package in package_ahriman.packages.values()])
copy_mock = mocker.patch("shutil.copy")
application_packages._add_directory(package_ahriman.base)
iterdir_mock.assert_called_once()
copy_mock.assert_called_once()
def test_add_local(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from local sources
"""
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
init_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.init")
copytree_mock = mocker.patch("shutil.copytree")
dependencies_mock = mocker.patch("ahriman.application.application.packages.Packages._process_dependencies")
application_packages._add_local(package_ahriman.base, set(), False)
init_mock.assert_called_once()
copytree_mock.assert_has_calls([
mock.call(Path(package_ahriman.base), application_packages.repository.paths.cache_for(package_ahriman.base)),
mock.call(application_packages.repository.paths.cache_for(package_ahriman.base),
application_packages.repository.paths.manual_for(package_ahriman.base)),
])
dependencies_mock.assert_called_once()
def test_add_remote(application_packages: Packages, package_description_ahriman: PackageDescription,
mocker: MockerFixture) -> None:
"""
must add package from remote source
"""
response_mock = MagicMock()
response_mock.iter_content.return_value = ["chunk"]
open_mock = mocker.patch("pathlib.Path.open")
request_mock = mocker.patch("requests.get", return_value=response_mock)
url = f"https://host/{package_description_ahriman.filename}"
application_packages._add_remote(url)
open_mock.assert_called_once_with("wb")
request_mock.assert_called_once_with(url, stream=True)
response_mock.raise_for_status.assert_called_once()
def test_process_dependencies(application_packages: Packages, mocker: MockerFixture) -> None:
"""
must process dependencies addition
"""
missing = {"python"}
path = Path("local")
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies", return_value=missing)
add_mock = mocker.patch("ahriman.application.application.packages.Packages.add")
application_packages._process_dependencies(path, set(), False)
dependencies_mock.assert_called_once_with(path)
add_mock.assert_called_once_with(missing, PackageSource.AUR, False)
def test_process_dependencies_missing(application_packages: Packages, mocker: MockerFixture) -> None:
"""
must process dependencies addition only for missing packages
"""
path = Path("local")
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies",
return_value={"python", "python-aiohttp"})
add_mock = mocker.patch("ahriman.application.application.packages.Packages.add")
application_packages._process_dependencies(path, {"python"}, False)
dependencies_mock.assert_called_once_with(path)
add_mock.assert_called_once_with({"python-aiohttp"}, PackageSource.AUR, False)
def test_process_dependencies_skip(application_packages: Packages, mocker: MockerFixture) -> None:
"""
must skip dependencies processing
"""
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies")
add_mock = mocker.patch("ahriman.application.application.packages.Packages.add")
application_packages._process_dependencies(Path("local"), set(), True)
dependencies_mock.assert_not_called()
add_mock.assert_not_called()
def test_add_add_archive(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from archive via add function
"""
mocker.patch("ahriman.application.application.packages.Packages._known_packages", return_value=set())
add_mock = mocker.patch("ahriman.application.application.packages.Packages._add_archive")
application_packages.add([package_ahriman.base], PackageSource.Archive, False)
add_mock.assert_called_once()
def test_add_add_aur(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from AUR via add function
"""
mocker.patch("ahriman.application.application.packages.Packages._known_packages", return_value=set())
add_mock = mocker.patch("ahriman.application.application.packages.Packages._add_aur")
application_packages.add([package_ahriman.base], PackageSource.AUR, True)
add_mock.assert_called_once()
def test_add_add_directory(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add packages from directory via add function
"""
mocker.patch("ahriman.application.application.packages.Packages._known_packages", return_value=set())
add_mock = mocker.patch("ahriman.application.application.packages.Packages._add_directory")
application_packages.add([package_ahriman.base], PackageSource.Directory, False)
add_mock.assert_called_once()
def test_add_add_local(application_packages: Packages, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from local sources via add function
"""
mocker.patch("ahriman.application.application.packages.Packages._known_packages", return_value=set())
add_mock = mocker.patch("ahriman.application.application.packages.Packages._add_local")
application_packages.add([package_ahriman.base], PackageSource.Local, False)
add_mock.assert_called_once()
def test_add_add_remote(application_packages: Packages, package_description_ahriman: PackageDescription,
mocker: MockerFixture) -> None:
"""
must add package from remote source via add function
"""
mocker.patch("ahriman.application.application.packages.Packages._known_packages", return_value=set())
add_mock = mocker.patch("ahriman.application.application.packages.Packages._add_remote")
url = f"https://host/{package_description_ahriman.filename}"
application_packages.add([url], PackageSource.Remote, False)
add_mock.assert_called_once()
def test_remove(application_packages: Packages, mocker: MockerFixture) -> None:
"""
must remove package
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_remove")
finalize_mock = mocker.patch("ahriman.application.application.packages.Packages._finalize")
application_packages.remove([])
executor_mock.assert_called_once()
finalize_mock.assert_called_once()

8
tests/ahriman/application/application/test_application_properties.py Normal file
View File

@ -0,0 +1,8 @@
from ahriman.application.application.properties import Properties
def test_create_tree(application_properties: Properties) -> None:
"""
must have repository attribute
"""
assert application_properties.repository

270
tests/ahriman/application/application/test_application_repository.py Normal file
View File

@ -0,0 +1,270 @@
import pytest
from pytest_mock import MockerFixture
from unittest import mock
from ahriman.application.application.repository import Repository
from ahriman.core.tree import Leaf, Tree
from ahriman.models.package import Package
def test_finalize(application_repository: Repository) -> None:
"""
must raise NotImplemented for missing finalize method
"""
with pytest.raises(NotImplementedError):
application_repository._finalize([])
def test_clean_build(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean build directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_build")
application_repository.clean(True, False, False, False, False, False)
clear_mock.assert_called_once()
def test_clean_cache(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean cache directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_cache")
application_repository.clean(False, True, False, False, False, False)
clear_mock.assert_called_once()
def test_clean_chroot(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean chroot directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_chroot")
application_repository.clean(False, False, True, False, False, False)
clear_mock.assert_called_once()
def test_clean_manual(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean manual directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_manual")
application_repository.clean(False, False, False, True, False, False)
clear_mock.assert_called_once()
def test_clean_packages(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean packages directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_packages")
application_repository.clean(False, False, False, False, True, False)
clear_mock.assert_called_once()
def test_clean_patches(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must clean packages directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_patches")
application_repository.clean(False, False, False, False, False, True)
clear_mock.assert_called_once()
def test_report(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must generate report
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_report")
application_repository.report([], [])
executor_mock.assert_called_once()
def test_sign(application_repository: Repository, package_ahriman: Package, package_python_schedule: Package,
mocker: MockerFixture) -> None:
"""
must sign world
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages",
return_value=[package_ahriman, package_python_schedule])
copy_mock = mocker.patch("shutil.copy")
update_mock = mocker.patch("ahriman.application.application.repository.Repository.update")
sign_repository_mock = mocker.patch("ahriman.core.sign.gpg.GPG.process_sign_repository")
finalize_mock = mocker.patch("ahriman.application.application.repository.Repository._finalize")
application_repository.sign([])
copy_mock.assert_has_calls([
mock.call(pytest.helpers.anyvar(str), pytest.helpers.anyvar(str)),
mock.call(pytest.helpers.anyvar(str), pytest.helpers.anyvar(str))
])
update_mock.assert_called_once_with([])
sign_repository_mock.assert_called_once()
finalize_mock.assert_called_once()
def test_sign_skip(application_repository: Repository, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must skip sign packages with empty filename
"""
package_ahriman.packages[package_ahriman.base].filename = None
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.application.application.repository.Repository.update")
mocker.patch("ahriman.application.application.repository.Repository._finalize")
application_repository.sign([])
def test_sign_specific(application_repository: Repository, package_ahriman: Package, package_python_schedule: Package,
mocker: MockerFixture) -> None:
"""
must sign only specified packages
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages",
return_value=[package_ahriman, package_python_schedule])
copy_mock = mocker.patch("shutil.copy")
update_mock = mocker.patch("ahriman.application.application.repository.Repository.update")
sign_repository_mock = mocker.patch("ahriman.core.sign.gpg.GPG.process_sign_repository")
finalize_mock = mocker.patch("ahriman.application.application.repository.Repository._finalize")
application_repository.sign([package_ahriman.base])
copy_mock.assert_called_once()
update_mock.assert_called_once_with([])
sign_repository_mock.assert_called_once()
finalize_mock.assert_called_once()
def test_sync(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must sync to remote
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_sync")
application_repository.sync([], [])
executor_mock.assert_called_once()
def test_unknown_no_aur(application_repository: Repository, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return empty list in case if there is locally stored PKGBUILD
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur", side_effect=Exception())
mocker.patch("pathlib.Path.is_dir", return_value=True)
mocker.patch("ahriman.core.build_tools.sources.Sources.has_remotes", return_value=False)
assert not application_repository.unknown()
def test_unknown_no_aur_no_local(application_repository: Repository, package_ahriman: Package,
mocker: MockerFixture) -> None:
"""
must return list of packages missing in aur and in local storage
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur", side_effect=Exception())
mocker.patch("pathlib.Path.is_dir", return_value=False)
packages = application_repository.unknown()
assert packages == [package_ahriman]
def test_unknown_no_local(application_repository: Repository, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return empty list in case if there is package in AUR
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur")
mocker.patch("pathlib.Path.is_dir", return_value=False)
assert not application_repository.unknown()
def test_update(application_repository: Repository, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package updates
"""
paths = [package.filepath for package in package_ahriman.packages.values()]
tree = Tree([Leaf(package_ahriman, set())])
mocker.patch("ahriman.core.tree.Tree.load", return_value=tree)
mocker.patch("ahriman.core.repository.repository.Repository.packages_built", return_value=[])
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
build_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_build", return_value=paths)
update_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_update")
finalize_mock = mocker.patch("ahriman.application.application.repository.Repository._finalize")
application_repository.update([package_ahriman])
build_mock.assert_called_once()
update_mock.assert_called_once_with(paths)
finalize_mock.assert_called_once_with([package_ahriman])
def test_updates_all(application_repository: Repository, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must get updates for all
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur",
return_value=[package_ahriman])
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates([], no_aur=False, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with([], False)
updates_manual_mock.assert_called_once()
def test_updates_disabled(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must get updates without anything
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates([], no_aur=True, no_manual=True, no_vcs=False, log_fn=print)
updates_aur_mock.assert_not_called()
updates_manual_mock.assert_not_called()
def test_updates_no_aur(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must get updates without aur
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates([], no_aur=True, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_not_called()
updates_manual_mock.assert_called_once()
def test_updates_no_manual(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must get updates without manual
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates([], no_aur=False, no_manual=True, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with([], False)
updates_manual_mock.assert_not_called()
def test_updates_no_vcs(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must get updates without VCS
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates([], no_aur=False, no_manual=False, no_vcs=True, log_fn=print)
updates_aur_mock.assert_called_once_with([], True)
updates_manual_mock.assert_called_once()
def test_updates_with_filter(application_repository: Repository, mocker: MockerFixture) -> None:
"""
must get updates without VCS
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application_repository.updates(["filter"], no_aur=False, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with(["filter"], False)
updates_manual_mock.assert_called_once()

2
tests/ahriman/application/handlers/test_handler_add.py
View File

@ -41,7 +41,7 @@ def test_run_with_updates(args: argparse.Namespace, configuration: Configuration
mocker.patch("ahriman.application.application.Application.add") mocker.patch("ahriman.application.application.Application.add")
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create") mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
application_mock = mocker.patch("ahriman.application.application.Application.update") application_mock = mocker.patch("ahriman.application.application.Application.update")
updates_mock = mocker.patch("ahriman.application.application.Application.get_updates") updates_mock = mocker.patch("ahriman.application.application.Application.updates")
Add.run(args, "x86_64", configuration, True) Add.run(args, "x86_64", configuration, True)
application_mock.assert_called_once() application_mock.assert_called_once()

12
tests/ahriman/application/handlers/test_handler_clean.py
View File

@ -12,12 +12,12 @@ def _default_args(args: argparse.Namespace) -> argparse.Namespace:
:param args: command line arguments fixture :param args: command line arguments fixture
:return: generated arguments for these test cases :return: generated arguments for these test cases
""" """
args.no_build = False args.build = False
args.no_cache = False args.cache = False
args.no_chroot = False args.chroot = False
args.no_manual = False args.manual = False
args.no_packages = False args.packages = False
args.no_patches = False args.patches = False
return args return args

4
tests/ahriman/application/handlers/test_handler_update.py
View File

@ -28,7 +28,7 @@ def test_run(args: argparse.Namespace, configuration: Configuration, mocker: Moc
args = _default_args(args) args = _default_args(args)
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create") mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
application_mock = mocker.patch("ahriman.application.application.Application.update") application_mock = mocker.patch("ahriman.application.application.Application.update")
updates_mock = mocker.patch("ahriman.application.application.Application.get_updates") updates_mock = mocker.patch("ahriman.application.application.Application.updates")
Update.run(args, "x86_64", configuration, True) Update.run(args, "x86_64", configuration, True)
application_mock.assert_called_once() application_mock.assert_called_once()
@ -42,7 +42,7 @@ def test_run_dry_run(args: argparse.Namespace, configuration: Configuration, moc
args = _default_args(args) args = _default_args(args)
args.dry_run = True args.dry_run = True
mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create") mocker.patch("ahriman.models.repository_paths.RepositoryPaths.tree_create")
updates_mock = mocker.patch("ahriman.application.application.Application.get_updates") updates_mock = mocker.patch("ahriman.application.application.Application.updates")
Update.run(args, "x86_64", configuration, True) Update.run(args, "x86_64", configuration, True)
updates_mock.assert_called_once() updates_mock.assert_called_once()

398
tests/ahriman/application/test_application.py
View File

@ -1,398 +0,0 @@
import pytest
from pathlib import Path
from pytest_mock import MockerFixture
from unittest import mock
from unittest.mock import MagicMock
from ahriman.application.application import Application
from ahriman.core.tree import Leaf, Tree
from ahriman.models.package import Package
from ahriman.models.package_description import PackageDescription
from ahriman.models.package_source import PackageSource
def test_finalize(application: Application, mocker: MockerFixture) -> None:
"""
must report and sync at the last
"""
report_mock = mocker.patch("ahriman.application.application.Application.report")
sync_mock = mocker.patch("ahriman.application.application.Application.sync")
application._finalize([])
report_mock.assert_called_once()
sync_mock.assert_called_once()
def test_known_packages(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return not empty list of known packages
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
packages = application._known_packages()
assert len(packages) > 1
assert package_ahriman.base in packages
def test_get_updates_all(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must get updates for all
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur",
return_value=[package_ahriman])
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates([], no_aur=False, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with([], False)
updates_manual_mock.assert_called_once()
def test_get_updates_disabled(application: Application, mocker: MockerFixture) -> None:
"""
must get updates without anything
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates([], no_aur=True, no_manual=True, no_vcs=False, log_fn=print)
updates_aur_mock.assert_not_called()
updates_manual_mock.assert_not_called()
def test_get_updates_no_aur(application: Application, mocker: MockerFixture) -> None:
"""
must get updates without aur
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates([], no_aur=True, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_not_called()
updates_manual_mock.assert_called_once()
def test_get_updates_no_manual(application: Application, mocker: MockerFixture) -> None:
"""
must get updates without manual
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates([], no_aur=False, no_manual=True, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with([], False)
updates_manual_mock.assert_not_called()
def test_get_updates_no_vcs(application: Application, mocker: MockerFixture) -> None:
"""
must get updates without VCS
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates([], no_aur=False, no_manual=False, no_vcs=True, log_fn=print)
updates_aur_mock.assert_called_once_with([], True)
updates_manual_mock.assert_called_once()
def test_get_updates_with_filter(application: Application, mocker: MockerFixture) -> None:
"""
must get updates without VCS
"""
updates_aur_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_aur")
updates_manual_mock = mocker.patch("ahriman.core.repository.update_handler.UpdateHandler.updates_manual")
application.get_updates(["filter"], no_aur=False, no_manual=False, no_vcs=False, log_fn=print)
updates_aur_mock.assert_called_once_with(["filter"], False)
updates_manual_mock.assert_called_once()
def test_add_archive(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from archive
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
copy_mock = mocker.patch("shutil.copy")
application.add([package_ahriman.base], PackageSource.Archive, False)
copy_mock.assert_called_once()
def test_add_aur(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from AUR
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
load_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.load")
application.add([package_ahriman.base], PackageSource.AUR, True)
load_mock.assert_called_once()
def test_add_aur_with_dependencies(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from AUR with dependencies
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
mocker.patch("ahriman.core.build_tools.sources.Sources.load")
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies")
application.add([package_ahriman.base], PackageSource.AUR, False)
dependencies_mock.assert_called_once()
def test_add_directory(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add packages from directory
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
iterdir_mock = mocker.patch("pathlib.Path.iterdir",
return_value=[package.filepath for package in package_ahriman.packages.values()])
copy_mock = mocker.patch("shutil.copy")
application.add([package_ahriman.base], PackageSource.Directory, False)
iterdir_mock.assert_called_once()
copy_mock.assert_called_once()
def test_add_local(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from local sources
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
init_mock = mocker.patch("ahriman.core.build_tools.sources.Sources.init")
copytree_mock = mocker.patch("shutil.copytree")
application.add([package_ahriman.base], PackageSource.Local, True)
init_mock.assert_called_once()
copytree_mock.assert_has_calls([
mock.call(Path(package_ahriman.base), application.repository.paths.cache_for(package_ahriman.base)),
mock.call(application.repository.paths.cache_for(package_ahriman.base),
application.repository.paths.manual_for(package_ahriman.base)),
])
def test_add_local_with_dependencies(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must add package from local sources with dependencies
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
mocker.patch("ahriman.core.build_tools.sources.Sources.init")
mocker.patch("shutil.copytree")
dependencies_mock = mocker.patch("ahriman.models.package.Package.dependencies")
application.add([package_ahriman.base], PackageSource.Local, False)
dependencies_mock.assert_called_once()
def test_add_remote(application: Application, package_description_ahriman: PackageDescription,
mocker: MockerFixture) -> None:
"""
must add package from remote source
"""
mocker.patch("ahriman.application.application.Application._known_packages", return_value=set())
response_mock = MagicMock()
response_mock.iter_content.return_value = ["chunk"]
open_mock = mocker.patch("pathlib.Path.open")
request_mock = mocker.patch("requests.get", return_value=response_mock)
url = f"https://host/{package_description_ahriman.filename}"
application.add([url], PackageSource.Remote, False)
open_mock.assert_called_once_with("wb")
request_mock.assert_called_once_with(url, stream=True)
response_mock.raise_for_status.assert_called_once()
def test_clean_build(application: Application, mocker: MockerFixture) -> None:
"""
must clean build directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_build")
application.clean(False, True, True, True, True, True)
clear_mock.assert_called_once()
def test_clean_cache(application: Application, mocker: MockerFixture) -> None:
"""
must clean cache directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_cache")
application.clean(True, False, True, True, True, True)
clear_mock.assert_called_once()
def test_clean_chroot(application: Application, mocker: MockerFixture) -> None:
"""
must clean chroot directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_chroot")
application.clean(True, True, False, True, True, True)
clear_mock.assert_called_once()
def test_clean_manual(application: Application, mocker: MockerFixture) -> None:
"""
must clean manual directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_manual")
application.clean(True, True, True, False, True, True)
clear_mock.assert_called_once()
def test_clean_packages(application: Application, mocker: MockerFixture) -> None:
"""
must clean packages directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_packages")
application.clean(True, True, True, True, False, True)
clear_mock.assert_called_once()
def test_clean_patches(application: Application, mocker: MockerFixture) -> None:
"""
must clean packages directory
"""
clear_mock = mocker.patch("ahriman.core.repository.cleaner.Cleaner.clear_patches")
application.clean(True, True, True, True, True, False)
clear_mock.assert_called_once()
def test_remove(application: Application, mocker: MockerFixture) -> None:
"""
must remove package
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_remove")
finalize_mock = mocker.patch("ahriman.application.application.Application._finalize")
application.remove([])
executor_mock.assert_called_once()
finalize_mock.assert_called_once()
def test_report(application: Application, mocker: MockerFixture) -> None:
"""
must generate report
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_report")
application.report([], [])
executor_mock.assert_called_once()
def test_sign(application: Application, package_ahriman: Package, package_python_schedule: Package,
mocker: MockerFixture) -> None:
"""
must sign world
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages",
return_value=[package_ahriman, package_python_schedule])
copy_mock = mocker.patch("shutil.copy")
update_mock = mocker.patch("ahriman.application.application.Application.update")
sign_repository_mock = mocker.patch("ahriman.core.sign.gpg.GPG.process_sign_repository")
finalize_mock = mocker.patch("ahriman.application.application.Application._finalize")
application.sign([])
copy_mock.assert_has_calls([
mock.call(pytest.helpers.anyvar(str), pytest.helpers.anyvar(str)),
mock.call(pytest.helpers.anyvar(str), pytest.helpers.anyvar(str))
])
update_mock.assert_called_once_with([])
sign_repository_mock.assert_called_once()
finalize_mock.assert_called_once()
def test_sign_skip(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must skip sign packages with empty filename
"""
package_ahriman.packages[package_ahriman.base].filename = None
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.application.application.Application.update")
application.sign([])
def test_sign_specific(application: Application, package_ahriman: Package, package_python_schedule: Package,
mocker: MockerFixture) -> None:
"""
must sign only specified packages
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages",
return_value=[package_ahriman, package_python_schedule])
copy_mock = mocker.patch("shutil.copy")
update_mock = mocker.patch("ahriman.application.application.Application.update")
sign_repository_mock = mocker.patch("ahriman.core.sign.gpg.GPG.process_sign_repository")
finalize_mock = mocker.patch("ahriman.application.application.Application._finalize")
application.sign([package_ahriman.base])
copy_mock.assert_called_once()
update_mock.assert_called_once_with([])
sign_repository_mock.assert_called_once()
finalize_mock.assert_called_once()
def test_sync(application: Application, mocker: MockerFixture) -> None:
"""
must sync to remote
"""
executor_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_sync")
application.sync([], [])
executor_mock.assert_called_once()
def test_unknown_no_aur(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return empty list in case if there is locally stored PKGBUILD
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur", side_effect=Exception())
mocker.patch("pathlib.Path.is_dir", return_value=True)
mocker.patch("ahriman.core.build_tools.sources.Sources.has_remotes", return_value=False)
assert not application.unknown()
def test_unknown_no_aur_no_local(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return list of packages missing in aur and in local storage
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur", side_effect=Exception())
mocker.patch("pathlib.Path.is_dir", return_value=False)
packages = application.unknown()
assert packages == [package_ahriman]
def test_unknown_no_local(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must return empty list in case if there is package in AUR
"""
mocker.patch("ahriman.core.repository.repository.Repository.packages", return_value=[package_ahriman])
mocker.patch("ahriman.models.package.Package.from_aur")
mocker.patch("pathlib.Path.is_dir", return_value=False)
assert not application.unknown()
def test_update(application: Application, package_ahriman: Package, mocker: MockerFixture) -> None:
"""
must process package updates
"""
paths = [package.filepath for package in package_ahriman.packages.values()]
tree = Tree([Leaf(package_ahriman, set())])
mocker.patch("ahriman.core.tree.Tree.load", return_value=tree)
mocker.patch("ahriman.core.repository.repository.Repository.packages_built", return_value=[])
mocker.patch("ahriman.models.package.Package.load", return_value=package_ahriman)
build_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_build", return_value=paths)
update_mock = mocker.patch("ahriman.core.repository.executor.Executor.process_update")
finalize_mock = mocker.patch("ahriman.application.application.Application._finalize")
application.update([package_ahriman])
build_mock.assert_called_once()
update_mock.assert_called_once_with(paths)
finalize_mock.assert_called_once_with([package_ahriman])

2
tests/ahriman/core/repository/conftest.py
View File

@ -3,10 +3,10 @@ import pytest
from pytest_mock import MockerFixture from pytest_mock import MockerFixture
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.repository import Repository
from ahriman.core.repository.cleaner import Cleaner from ahriman.core.repository.cleaner import Cleaner
from ahriman.core.repository.executor import Executor from ahriman.core.repository.executor import Executor
from ahriman.core.repository.properties import Properties from ahriman.core.repository.properties import Properties
from ahriman.core.repository.repository import Repository
from ahriman.core.repository.update_handler import UpdateHandler from ahriman.core.repository.update_handler import UpdateHandler

2
tests/ahriman/core/repository/test_repository.py
View File

@ -1,7 +1,7 @@
from pathlib import Path from pathlib import Path
from pytest_mock import MockerFixture from pytest_mock import MockerFixture
from ahriman.core.repository.repository import Repository from ahriman.core.repository import Repository
from ahriman.models.package import Package from ahriman.models.package import Package