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

expose trigger configuration schema

Browse Source 
Note that this commit contains the following breaking changes:

* remote pull and remote push triggers are now enabled by default (with
  empty target list)
* remote pull and remote push triggers now require target option to be
  set (old behaviour had fallback on `gitremote`)
* validation is now considered to be stable, so it is enabled by default
  in docker image (can be disabled however)
This commit is contained in:
Evgenii Alekseev 2023-01-10 03:14:21 +02:00
parent b09aea13af
commit 5a05c8ce91
27 changed files with 717 additions and 448 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
CONTRIBUTING.md
View File

@ -31,13 +31,15 @@ Again, the most checks can be performed by `make check` command, though some add
```python ```python
def foo(argument: str, *, flag: bool = False) -> int: def foo(argument: str, *, flag: bool = False) -> int:
""" """
do foo do foo. With very very very long
docstring
Note: Note:
Very important note about this function Very important note about this function
Args: Args:
argument(str): an argument argument(str): an argument. This argument has
long description also
flag(bool, optional): a flag (Default value = False) flag(bool, optional): a flag (Default value = False)
Returns: Returns:

2
Dockerfile
View File

@ -12,7 +12,7 @@ ENV AHRIMAN_REPOSITORY="aur-clone"
ENV AHRIMAN_REPOSITORY_ROOT="/var/lib/ahriman/ahriman" ENV AHRIMAN_REPOSITORY_ROOT="/var/lib/ahriman/ahriman"
ENV AHRIMAN_UNIX_SOCKET="" ENV AHRIMAN_UNIX_SOCKET=""
ENV AHRIMAN_USER="ahriman" ENV AHRIMAN_USER="ahriman"
ENV AHRIMAN_VALIDATE_CONFIGURATION="" ENV AHRIMAN_VALIDATE_CONFIGURATION="yes"
# install environment # install environment
## update pacman.conf with multilib ## update pacman.conf with multilib

46
docs/faq.rst
View File

@ -144,52 +144,30 @@ Before using this command you will need to create local directory, put ``PKGBUIL
How to fetch PKGBUILDs from remote repository How to fetch PKGBUILDs from remote repository
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
For that purpose you could use ``RemotePullTrigger`` trigger. To do so you will need: For that purpose you could use ``RemotePullTrigger`` trigger. To do so you will need to configure trigger as following:
#. .. code-block:: ini
Append ``triggers`` option in ``build`` section with the following line:
.. code-block:: ini [remote-pull]
target = gitremote
[build] [gitremote]
triggers = ahriman.core.gitremote.RemotePullTrigger pull_url = https://github.com/username/repository
#.
Configure trigger as following:
.. code-block:: ini
[remote-pull]
target = gitremote
[gitremote]
pull_url = https://github.com/username/repository
During the next application run it will fetch repository from the specified url and will try to find packages there which can be used as local sources. During the next application run it will fetch repository from the specified url and will try to find packages there which can be used as local sources.
How to push updated PKGBUILDs to remote repository How to push updated PKGBUILDs to remote repository
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
For that purpose you'd need to use another trigger called ``RemotePushTrigger``. Configure it as following: For that purpose you'd need to use another trigger called ``RemotePushTrigger``. Configure trigger as following:
#. .. code-block:: ini
Append ``triggers`` option in ``build`` section with the trigger name:
.. code-block:: ini [remote-push]
target = gitremote
[build] [gitremote]
triggers = ahriman.core.gitremote.RemotePushTrigger push_url = https://github.com/username/repository
#.
Configure trigger as following:
.. code-block:: ini
[remote-push]
target = gitremote
[gitremote]
push_url = https://github.com/username/repository
Unlike ``RemotePullTrigger`` trigger, the ``RemotePushTrigger`` more likely will require authorization. It is highly recommended to use application tokens for that instead of using your password (e.g. for Github you can generate tokens `here <https://github.com/settings/tokens>`_ with scope ``public_repo``). Authorization can be supplied by using authorization part of the url, e.g. ``https://key:token@github.com/username/repository``. Unlike ``RemotePullTrigger`` trigger, the ``RemotePushTrigger`` more likely will require authorization. It is highly recommended to use application tokens for that instead of using your password (e.g. for Github you can generate tokens `here <https://github.com/settings/tokens>`_ with scope ``public_repo``). Authorization can be supplied by using authorization part of the url, e.g. ``https://key:token@github.com/username/repository``.

5
docs/triggers.rst
View File

@ -125,3 +125,8 @@ Setup the trigger
First, put the trigger in any path it can be exported, e.g. by packing the resource into python package (which will lead to import path as ``package.slack_reporter.SlackReporter``) or just put file somewhere it can be accessed by application (e.g. ``/usr/local/lib/slack_reporter.py.SlackReporter``). First, put the trigger in any path it can be exported, e.g. by packing the resource into python package (which will lead to import path as ``package.slack_reporter.SlackReporter``) or just put file somewhere it can be accessed by application (e.g. ``/usr/local/lib/slack_reporter.py.SlackReporter``).
After that run application as usual and receive notification in your slack channel. After that run application as usual and receive notification in your slack channel.
Trigger configuration schema
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Triggers can expose their configuration schema. It can be achieved by implementing ``CONFIGURATION_SCHEMA`` class variable according to `cerberus <https://docs.python-cerberus.org/>`_ documentation. For more details and examples, please refer to built-in triggers implementations.

8
package/share/ahriman/settings/ahriman.ini
View File

@ -23,7 +23,7 @@ build_command = extra-x86_64-build
ignore_packages = ignore_packages =
makechrootpkg_flags = makechrootpkg_flags =
makepkg_flags = --nocolor --ignorearch makepkg_flags = --nocolor --ignorearch
triggers = ahriman.core.report.ReportTrigger ahriman.core.upload.UploadTrigger triggers = ahriman.core.gitremote.RemotePullTrigger ahriman.core.report.ReportTrigger ahriman.core.upload.UploadTrigger ahriman.core.gitremote.RemotePushTrigger
vcs_allowed_age = 604800 vcs_allowed_age = 604800
[repository] [repository]
@ -33,6 +33,12 @@ root = /var/lib/ahriman
[sign] [sign]
target = target =
[remote-pull]
target =
[remote-push]
target =
[report] [report]
target = console target = console

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

@ -48,6 +48,6 @@ class Triggers(Handler):
application = Application(architecture, configuration, report=report, unsafe=unsafe) application = Application(architecture, configuration, report=report, unsafe=unsafe)
if args.trigger: if args.trigger:
loader = application.repository.triggers loader = application.repository.triggers
loader.triggers = [loader.load_trigger(trigger) for trigger in args.trigger] loader.triggers = [loader.load_trigger(trigger, architecture, configuration) for trigger in args.trigger]
application.on_start() application.on_start()
application.on_result(Result()) application.on_result(Result())

89
src/ahriman/application/handlers/validate.py
View File

@ -20,16 +20,15 @@
import argparse import argparse
import copy import copy
from typing import Any, Callable, Dict, Optional, Type from typing import Any, Dict, Type
from ahriman.application.handlers import Handler from ahriman.application.handlers import Handler
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.configuration.schema import CONFIGURATION_SCHEMA, \ from ahriman.core.configuration.schema import CONFIGURATION_SCHEMA, ConfigurationSchema
GITREMOTE_REMOTE_PULL_SCHEMA, GITREMOTE_REMOTE_PUSH_SCHEMA, \
REPORT_CONSOLE_SCHEMA, REPORT_EMAIL_SCHEMA, REPORT_HTML_SCHEMA, REPORT_TELEGRAM_SCHEMA,\
UPLOAD_GITHUB_SCHEMA, UPLOAD_RSYNC_SCHEMA, UPLOAD_S3_SCHEMA
from ahriman.core.configuration.validator import Validator from ahriman.core.configuration.validator import Validator
from ahriman.core.exceptions import ExtensionError
from ahriman.core.formatters import ValidationPrinter from ahriman.core.formatters import ValidationPrinter
from ahriman.core.triggers import TriggerLoader
class Validate(Handler): class Validate(Handler):
@ -64,7 +63,7 @@ class Validate(Handler):
Validate.check_if_empty(args.exit_code, True) Validate.check_if_empty(args.exit_code, True)
@staticmethod @staticmethod
def schema(architecture: str, configuration: Configuration) -> Dict[str, Any]: def schema(architecture: str, configuration: Configuration) -> ConfigurationSchema:
""" """
get schema with triggers get schema with triggers
@ -73,45 +72,39 @@ class Validate(Handler):
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
Returns: Returns:
Dict[str, Any]: configuration validation schema ConfigurationSchema: configuration validation schema
""" """
root = copy.deepcopy(CONFIGURATION_SCHEMA) root = copy.deepcopy(CONFIGURATION_SCHEMA)
# that's actually bad design, but in order to validate built-in triggers we need to know which are set # create trigger loader instance
Validate.schema_insert(architecture, configuration, root, "remote-pull", lambda _: GITREMOTE_REMOTE_PULL_SCHEMA) loader = TriggerLoader()
Validate.schema_insert(architecture, configuration, root, "remote-push", lambda _: GITREMOTE_REMOTE_PUSH_SCHEMA) for trigger in loader.selected_triggers(configuration):
try:
trigger_class = loader.load_trigger_class(trigger)
except ExtensionError:
continue
report_schemas = { # default settings if any
"console": REPORT_CONSOLE_SCHEMA, for schema_name, schema in trigger_class.configuration_schema(architecture, None).items():
"email": REPORT_EMAIL_SCHEMA, erased = Validate.schema_erase_required(copy.deepcopy(schema))
"html": REPORT_HTML_SCHEMA, root[schema_name] = Validate.schema_merge(root.get(schema_name, {}), erased)
"telegram": REPORT_TELEGRAM_SCHEMA,
}
for schema_name, schema in report_schemas.items():
root[schema_name] = Validate.schema_erase_required(copy.deepcopy(schema))
Validate.schema_insert(architecture, configuration, root, "report", report_schemas.get)
upload_schemas = { # settings according to enabled triggers
"github": UPLOAD_GITHUB_SCHEMA, for schema_name, schema in trigger_class.configuration_schema(architecture, configuration).items():
"rsync": UPLOAD_RSYNC_SCHEMA, root[schema_name] = Validate.schema_merge(root.get(schema_name, {}), copy.deepcopy(schema))
"s3": UPLOAD_S3_SCHEMA,
}
for schema_name, schema in upload_schemas.items():
root[schema_name] = Validate.schema_erase_required(copy.deepcopy(schema))
Validate.schema_insert(architecture, configuration, root, "upload", upload_schemas.get)
return root return root
@staticmethod @staticmethod
def schema_erase_required(schema: Dict[str, Any]) -> Dict[str, Any]: def schema_erase_required(schema: ConfigurationSchema) -> ConfigurationSchema:
""" """
recursively remove required field from supplied cerberus schema recursively remove required field from supplied cerberus schema
Args: Args:
schema(Dict[str, Any]): source schema from which required field must be removed schema(ConfigurationSchema): source schema from which required field must be removed
Returns: Returns:
Dict[str, Any]: schema without required fields ConfigurationSchema: schema without required fields. Note, that source schema will be modified in-place
""" """
schema.pop("required", None) schema.pop("required", None)
for value in filter(lambda v: isinstance(v, dict), schema.values()): for value in filter(lambda v: isinstance(v, dict), schema.values()):
@ -119,32 +112,24 @@ class Validate(Handler):
return schema return schema
@staticmethod @staticmethod
def schema_insert(architecture: str, configuration: Configuration, root: Dict[str, Any], root_section: str, def schema_merge(source: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]:
schema_mapping: Callable[[str], Optional[Dict[str, Any]]]) -> Dict[str, Any]:
""" """
insert child schema into the root schema based on mapping rules merge child schema into source. In case if source already contains values, new keys will be added
(possibly with overrides - in case if such key already set also)
Notes:
Actually it is a bad design, because we are reading triggers configuration from parsers which (basically)
don't know anything about triggers. But in order to validate built-in triggers we need to know which are set
Args: Args:
architecture(str): repository architecture source(Dict[str, Any]): source (current) schema into which will be merged
configuration(Configuration): configuration instance schema(Dict[str, Any]): new schema to be merged
root(Dict[str, Any]): root schema in which child schema will be inserted
root_section(str): section name in root schema
schema_mapping(Callable[[str], Optional[Dict[str, Any]]]): extractor for child schema based on trigger type
Returns: Returns:
Dict[str, Any]: modified root schema. Note, however, that schema will be modified in place Dict[str, Any]: schema with added elements from source schema if they were set before and not presented
in the new one. Note, that schema will be modified in-place
""" """
if not configuration.has_section(root_section): for key, value in source.items():
return root if key not in schema:
schema[key] = value # new key found, just add it as is
elif isinstance(value, dict):
# value is dictionary, so we need to go deeper
Validate.schema_merge(value, schema[key])
targets = configuration.getlist(root_section, "target", fallback=[]) return schema
for target in targets:
section, schema_name = configuration.gettype(target, architecture)
if (schema := schema_mapping(schema_name)) is not None:
root[section] = copy.deepcopy(schema)
return root

13
src/ahriman/core/configuration/configuration.py
View File

@ -189,7 +189,7 @@ class Configuration(configparser.RawConfigParser):
def getpath(self, *args: Any, **kwargs: Any) -> Path: ... # type: ignore def getpath(self, *args: Any, **kwargs: Any) -> Path: ... # type: ignore
def gettype(self, section: str, architecture: str) -> Tuple[str, str]: def gettype(self, section: str, architecture: str, *, fallback: Optional[str] = None) -> Tuple[str, str]:
""" """
get type variable with fallback to old logic. Despite the fact that it has same semantics as other get* methods, get type variable with fallback to old logic. Despite the fact that it has same semantics as other get* methods,
but it has different argument list but it has different argument list
@ -197,6 +197,8 @@ class Configuration(configparser.RawConfigParser):
Args: Args:
section(str): section name section(str): section name
architecture(str): repository architecture architecture(str): repository architecture
fallback(Optional[str], optional): optional fallback type if any. If set, second element of the tuple will
be always set to this value (Default value = None)
Returns: Returns:
Tuple[str, str]: section name and found type name Tuple[str, str]: section name and found type name
@ -204,9 +206,8 @@ class Configuration(configparser.RawConfigParser):
Raises: Raises:
configparser.NoSectionError: in case if no section found configparser.NoSectionError: in case if no section found
""" """
group_type = self.get(section, "type", fallback=None) # new-style logic if (group_type := self.get(section, "type", fallback=fallback)) is not None:
if group_type is not None: return section, group_type # new-style logic
return section, group_type
# okay lets check for the section with architecture name # okay lets check for the section with architecture name
full_section = self.section_name(section, architecture) full_section = self.section_name(section, architecture)
if self.has_section(full_section): if self.has_section(full_section):
@ -274,14 +275,14 @@ class Configuration(configparser.RawConfigParser):
self.load(path) self.load(path)
self.merge_sections(architecture) self.merge_sections(architecture)
def set_option(self, section: str, option: str, value: Optional[str]) -> None: def set_option(self, section: str, option: str, value: str) -> None:
""" """
set option. Unlike default ``configparser.RawConfigParser.set`` it also creates section if it does not exist set option. Unlike default ``configparser.RawConfigParser.set`` it also creates section if it does not exist
Args: Args:
section(str): section name section(str): section name
option(str): option name option(str): option name
value(Optional[str]): option value as string in parsable format value(str): option value as string in parsable format
""" """
if not self.has_section(section): if not self.has_section(section):
self.add_section(section) self.add_section(section)

276
src/ahriman/core/configuration/schema.py
View File

@ -17,16 +17,16 @@
# 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/>.
# #
# pylint: disable=too-many-lines from typing import Any, Dict
__all__ = [
"CONFIGURATION_SCHEMA",
"GITREMOTE_REMOTE_PULL_SCHEMA", "GITREMOTE_REMOTE_PUSH_SCHEMA",
"REPORT_CONSOLE_SCHEMA", "REPORT_EMAIL_SCHEMA", "REPORT_HTML_SCHEMA", "REPORT_TELEGRAM_SCHEMA",
"UPLOAD_GITHUB_SCHEMA", "UPLOAD_RSYNC_SCHEMA", "UPLOAD_S3_SCHEMA",
]
CONFIGURATION_SCHEMA = { __all__ = ["CONFIGURATION_SCHEMA", "ConfigurationSchema"]
ConfigurationSchema = Dict[str, Dict[str, Any]]
CONFIGURATION_SCHEMA: ConfigurationSchema = {
"settings": { "settings": {
"type": "dict", "type": "dict",
"schema": { "schema": {
@ -292,263 +292,3 @@ CONFIGURATION_SCHEMA = {
}, },
}, },
} }
GITREMOTE_REMOTE_PULL_SCHEMA = {
"type": "dict",
"schema": {
"pull_url": {
"type": "string",
"required": True,
},
"pull_branch": {
"type": "string",
},
},
}
GITREMOTE_REMOTE_PUSH_SCHEMA = {
"type": "dict",
"schema": {
"commit_author": {
"type": "string",
},
"push_url": {
"type": "string",
"required": True,
},
"push_branch": {
"type": "string",
},
},
}
REPORT_CONSOLE_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["console"],
},
"use_utf": {
"type": "boolean",
"coerce": "boolean",
},
},
}
REPORT_EMAIL_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["email"],
},
"full_template_path": {
"type": "path",
"coerce": "absolute_path",
"path_exists": True,
},
"homepage": {
"type": "string",
},
"host": {
"type": "string",
"required": True,
},
"link_path": {
"type": "string",
"required": True,
},
"no_empty_report": {
"type": "boolean",
"coerce": "boolean",
},
"password": {
"type": "string",
},
"port": {
"type": "integer",
"coerce": "integer",
"required": True,
},
"receivers": {
"type": "list",
"coerce": "list",
"schema": {"type": "string"},
"required": True,
"empty": False,
},
"sender": {
"type": "string",
"required": True,
},
"ssl": {
"type": "string",
"allowed": ["ssl", "starttls", "disabled"],
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
"user": {
"type": "string",
},
},
}
REPORT_HTML_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["html"],
},
"homepage": {
"type": "string",
},
"link_path": {
"type": "string",
"required": True,
},
"path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
},
}
REPORT_TELEGRAM_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["telegram"],
},
"api_key": {
"type": "string",
"required": True,
},
"chat_id": {
"type": "string",
"required": True,
},
"homepage": {
"type": "string",
},
"link_path": {
"type": "string",
"required": True,
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
"template_type": {
"type": "string",
"allowed": ["MarkdownV2", "HTML", "Markdown"],
},
"timeout": {
"type": "integer",
"coerce": "integer",
},
},
}
UPLOAD_GITHUB_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["github"],
},
"owner": {
"type": "string",
"required": True,
},
"password": {
"type": "string",
"required": True,
},
"repository": {
"type": "string",
"required": True,
},
"timeout": {
"type": "integer",
"coerce": "integer",
},
"username": {
"type": "string",
},
},
}
UPLOAD_RSYNC_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["rsync"],
},
"command": {
"type": "list",
"coerce": "list",
"schema": {"type": "string"},
"required": True,
"empty": False,
},
"remote": {
"type": "string",
"required": True,
},
},
}
UPLOAD_S3_SCHEMA = {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["s3"],
},
"access_key": {
"type": "string",
"required": True,
},
"bucket": {
"type": "string",
"required": True,
},
"chunk_size": {
"type": "integer",
"coerce": "integer",
},
"region": {
"type": "string",
"required": True,
},
"secret_key": {
"type": "string",
"required": True,
},
},
}

1
src/ahriman/core/configuration/validator.py
View File

@ -84,6 +84,7 @@ class Validator(RootValidator): # type: ignore
Returns: Returns:
int: value converted to int according to configuration rules int: value converted to int according to configuration rules
""" """
del self
return int(value) return int(value)
def _normalize_coerce_list(self, value: str) -> List[str]: def _normalize_coerce_list(self, value: str) -> List[str]:

36
src/ahriman/core/gitremote/remote_pull_trigger.py
View File

@ -17,6 +17,8 @@
# 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 typing import List, Type
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.gitremote.remote_pull import RemotePull from ahriman.core.gitremote.remote_pull import RemotePull
from ahriman.core.triggers import Trigger from ahriman.core.triggers import Trigger
@ -30,6 +32,22 @@ class RemotePullTrigger(Trigger):
targets(List[str]): git remote target list targets(List[str]): git remote target list
""" """
CONFIGURATION_SCHEMA = {
"gitremote": {
"type": "dict",
"schema": {
"pull_url": {
"type": "string",
"required": True,
},
"pull_branch": {
"type": "string",
},
},
},
}
CONFIGURATION_SCHEMA_FALLBACK = "gitremote"
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self, architecture: str, configuration: Configuration) -> None:
""" """
default constructor default constructor
@ -39,13 +57,27 @@ class RemotePullTrigger(Trigger):
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
""" """
Trigger.__init__(self, architecture, configuration) Trigger.__init__(self, architecture, configuration)
self.targets = configuration.getlist("remote-pull", "target", fallback=["gitremote"]) self.targets = self.configuration_sections(configuration)
@classmethod
def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
"""
extract configuration sections from configuration
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: read configuration sections belong to this trigger
"""
return configuration.getlist("remote-pull", "target", fallback=[])
def on_start(self) -> None: def on_start(self) -> None:
""" """
trigger action which will be called at the start of the application trigger action which will be called at the start of the application
""" """
for target in self.targets: for target in self.targets:
section, _ = self.configuration.gettype(target, self.architecture) section, _ = self.configuration.gettype(
target, self.architecture, fallback=self.CONFIGURATION_SCHEMA_FALLBACK)
runner = RemotePull(self.configuration, section) runner = RemotePull(self.configuration, section)
runner.run() runner.run()

39
src/ahriman/core/gitremote/remote_push_trigger.py
View File

@ -17,7 +17,7 @@
# 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 typing import Iterable from typing import Iterable, List, Type
from ahriman.core import context from ahriman.core import context
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
@ -37,6 +37,25 @@ class RemotePushTrigger(Trigger):
targets(List[str]): git remote target list targets(List[str]): git remote target list
""" """
CONFIGURATION_SCHEMA = {
"gitremote": {
"type": "dict",
"schema": {
"commit_author": {
"type": "string",
},
"push_url": {
"type": "string",
"required": True,
},
"push_branch": {
"type": "string",
},
},
},
}
CONFIGURATION_SCHEMA_FALLBACK = "gitremote"
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self, architecture: str, configuration: Configuration) -> None:
""" """
default constructor default constructor
@ -46,7 +65,20 @@ class RemotePushTrigger(Trigger):
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
""" """
Trigger.__init__(self, architecture, configuration) Trigger.__init__(self, architecture, configuration)
self.targets = configuration.getlist("remote-push", "target", fallback=["gitremote"]) self.targets = self.configuration_sections(configuration)
@classmethod
def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
"""
extract configuration sections from configuration
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: read configuration sections belong to this trigger
"""
return configuration.getlist("remote-push", "target", fallback=[])
def on_result(self, result: Result, packages: Iterable[Package]) -> None: def on_result(self, result: Result, packages: Iterable[Package]) -> None:
""" """
@ -63,6 +95,7 @@ class RemotePushTrigger(Trigger):
database = ctx.get(ContextKey("database", SQLite)) database = ctx.get(ContextKey("database", SQLite))
for target in self.targets: for target in self.targets:
section, _ = self.configuration.gettype(target, self.architecture) section, _ = self.configuration.gettype(
target, self.architecture, fallback=self.CONFIGURATION_SCHEMA_FALLBACK)
runner = RemotePush(self.configuration, database, section) runner = RemotePush(self.configuration, database, section)
runner.run(result) runner.run(result)

159
src/ahriman/core/report/report_trigger.py
View File

@ -17,7 +17,7 @@
# 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 typing import Iterable from typing import Iterable, List, Type
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.triggers import Trigger from ahriman.core.triggers import Trigger
@ -34,6 +34,148 @@ class ReportTrigger(Trigger):
targets(List[str]): report target list targets(List[str]): report target list
""" """
CONFIGURATION_SCHEMA = {
"console": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["console"],
},
"use_utf": {
"type": "boolean",
"coerce": "boolean",
},
},
},
"email": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["email"],
},
"full_template_path": {
"type": "path",
"coerce": "absolute_path",
"path_exists": True,
},
"homepage": {
"type": "string",
},
"host": {
"type": "string",
"required": True,
},
"link_path": {
"type": "string",
"required": True,
},
"no_empty_report": {
"type": "boolean",
"coerce": "boolean",
},
"password": {
"type": "string",
},
"port": {
"type": "integer",
"coerce": "integer",
"required": True,
},
"receivers": {
"type": "list",
"coerce": "list",
"schema": {"type": "string"},
"required": True,
"empty": False,
},
"sender": {
"type": "string",
"required": True,
},
"ssl": {
"type": "string",
"allowed": ["ssl", "starttls", "disabled"],
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
"user": {
"type": "string",
},
},
},
"html": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["html"],
},
"homepage": {
"type": "string",
},
"link_path": {
"type": "string",
"required": True,
},
"path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
},
},
"telegram": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["telegram"],
},
"api_key": {
"type": "string",
"required": True,
},
"chat_id": {
"type": "string",
"required": True,
},
"homepage": {
"type": "string",
},
"link_path": {
"type": "string",
"required": True,
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
"path_exists": True,
},
"template_type": {
"type": "string",
"allowed": ["MarkdownV2", "HTML", "Markdown"],
},
"timeout": {
"type": "integer",
"coerce": "integer",
},
},
},
}
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self, architecture: str, configuration: Configuration) -> None:
""" """
default constructor default constructor
@ -43,7 +185,20 @@ class ReportTrigger(Trigger):
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
""" """
Trigger.__init__(self, architecture, configuration) Trigger.__init__(self, architecture, configuration)
self.targets = configuration.getlist("report", "target", fallback=[]) self.targets = self.configuration_sections(configuration)
@classmethod
def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
"""
extract configuration sections from configuration
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: read configuration sections belong to this trigger
"""
return configuration.getlist("report", "target", fallback=[])
def on_result(self, result: Result, packages: Iterable[Package]) -> None: def on_result(self, result: Result, packages: Iterable[Package]) -> None:
""" """

2
src/ahriman/core/repository/repository_properties.py
View File

@ -80,4 +80,4 @@ class RepositoryProperties(LazyLogging):
self.sign = GPG(architecture, configuration) self.sign = GPG(architecture, configuration)
self.repo = Repo(self.name, self.paths, self.sign.repository_sign_args) self.repo = Repo(self.name, self.paths, self.sign.repository_sign_args)
self.reporter = Client.load(configuration, report=report) self.reporter = Client.load(configuration, report=report)
self.triggers = TriggerLoader(architecture, configuration) self.triggers = TriggerLoader.load(architecture, configuration)

67
src/ahriman/core/triggers/trigger.py
View File

@ -17,9 +17,12 @@
# 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 typing import Iterable from __future__ import annotations
from typing import Iterable, List, Optional, Type
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.configuration.schema import ConfigurationSchema
from ahriman.core.log import LazyLogging from ahriman.core.log import LazyLogging
from ahriman.models.package import Package from ahriman.models.package import Package
from ahriman.models.result import Result from ahriman.models.result import Result
@ -30,6 +33,9 @@ class Trigger(LazyLogging):
trigger base class trigger base class
Attributes: Attributes:
CONFIGURATION_SCHEMA(ConfigurationSchema): (class attribute) configuration schema template
CONFIGURATION_SCHEMA_FALLBACK(Optional[str]): (class attribute) optional fallback option for defining
configuration schema type used
architecture(str): repository architecture architecture(str): repository architecture
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
@ -47,10 +53,13 @@ class Trigger(LazyLogging):
>>> configuration = Configuration() >>> configuration = Configuration()
>>> configuration.set_option("build", "triggers", "my.awesome.package.CustomTrigger") >>> configuration.set_option("build", "triggers", "my.awesome.package.CustomTrigger")
>>> >>>
>>> loader = TriggerLoader("x86_64", configuration) >>> loader = TriggerLoader.load("x86_64", configuration)
>>> loader.on_result(Result(), []) >>> loader.on_result(Result(), [])
""" """
CONFIGURATION_SCHEMA: ConfigurationSchema = {}
CONFIGURATION_SCHEMA_FALLBACK: Optional[str] = None
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self, architecture: str, configuration: Configuration) -> None:
""" """
default constructor default constructor
@ -62,6 +71,60 @@ class Trigger(LazyLogging):
self.architecture = architecture self.architecture = architecture
self.configuration = configuration self.configuration = configuration
@classmethod
def configuration_schema(cls: Type[Trigger], architecture: str,
configuration: Optional[Configuration]) -> ConfigurationSchema:
"""
configuration schema based on supplied service configuration
Notes:
Schema must be in cerberus format, for details and examples you can check built-in triggers.
Args:
architecture(str): repository architecture
configuration(Optional[Configuration]): configuration instance. If set to None, the default schema
should be returned
Returns:
ConfigurationSchema: configuration schema in cerberus format
"""
if configuration is None:
return cls.CONFIGURATION_SCHEMA
result: ConfigurationSchema = {}
for target in cls.configuration_sections(configuration):
if not configuration.has_section(target):
continue
section, schema_name = configuration.gettype(
target, architecture, fallback=cls.CONFIGURATION_SCHEMA_FALLBACK)
if schema_name not in cls.CONFIGURATION_SCHEMA:
continue
result[section] = cls.CONFIGURATION_SCHEMA[schema_name]
return result
@classmethod
def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
"""
extract configuration sections from configuration
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: read configuration sections belong to this trigger
Examples:
This method can be used in order to extract specific configuration sections which are set by user, e.g.
from sources::
>>> @staticmethod
>>> def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
>>> return configuration.getlist("report", "target", fallback=[])
"""
del configuration
return []
def on_result(self, result: Result, packages: Iterable[Package]) -> None: def on_result(self, result: Result, packages: Iterable[Package]) -> None:
""" """
trigger action which will be called after build process with process result trigger action which will be called after build process with process result

85
src/ahriman/core/triggers/trigger_loader.py
View File

@ -17,13 +17,15 @@
# 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 __future__ import annotations
import contextlib import contextlib
import importlib import importlib
import os import os
from pathlib import Path from pathlib import Path
from types import ModuleType from types import ModuleType
from typing import Generator, Iterable from typing import Generator, Iterable, List, Type
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import ExtensionError from ahriman.core.exceptions import ExtensionError
@ -38,8 +40,6 @@ class TriggerLoader(LazyLogging):
trigger loader class trigger loader class
Attributes: Attributes:
architecture(str): repository architecture
configuration(Configuration): configuration instance
triggers(List[Trigger]): list of loaded triggers according to the configuration triggers(List[Trigger]): list of loaded triggers according to the configuration
Examples: Examples:
@ -50,7 +50,7 @@ class TriggerLoader(LazyLogging):
Having such configuration you can create instance of the loader:: Having such configuration you can create instance of the loader::
>>> loader = TriggerLoader("x86_64", configuration) >>> loader = TriggerLoader.load("x86_64", configuration)
>>> print(loader.triggers) >>> print(loader.triggers)
After that you are free to run triggers:: After that you are free to run triggers::
@ -58,23 +58,46 @@ class TriggerLoader(LazyLogging):
>>> loader.on_result(Result(), []) >>> loader.on_result(Result(), [])
""" """
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self) -> None:
""" """
default constructor default constructor
"""
self._on_stop_requested = False
self.triggers: List[Trigger] = []
@classmethod
def load(cls: Type[TriggerLoader], architecture: str, configuration: Configuration) -> TriggerLoader:
"""
create instance from configuration
Args: Args:
architecture(str): repository architecture architecture(str): repository architecture
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
"""
self.architecture = architecture
self.configuration = configuration
self._on_stop_requested = False Returns:
self.triggers = [ TriggerLoader: fully loaded trigger instance
self.load_trigger(trigger) """
for trigger in configuration.getlist("build", "triggers") instance = cls()
instance.triggers = [
instance.load_trigger(trigger, architecture, configuration)
for trigger in instance.selected_triggers(configuration)
] ]
return instance
@staticmethod
def selected_triggers(configuration: Configuration) -> List[str]:
"""
read configuration and return triggers which are set by settings
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: list of triggers according to configuration
"""
return configuration.getlist("build", "triggers", fallback=[])
@contextlib.contextmanager @contextlib.contextmanager
def __execute_trigger(self, trigger: Trigger) -> Generator[None, None, None]: def __execute_trigger(self, trigger: Trigger) -> Generator[None, None, None]:
""" """
@ -130,16 +153,39 @@ class TriggerLoader(LazyLogging):
except ModuleNotFoundError: except ModuleNotFoundError:
raise ExtensionError(f"Module {package} not found") raise ExtensionError(f"Module {package} not found")
def load_trigger(self, module_path: str) -> Trigger: def load_trigger(self, module_path: str, architecture: str, configuration: Configuration) -> Trigger:
""" """
load trigger by module path load trigger by module path
Args: Args:
module_path(str): module import path to load module_path(str): module import path to load
architecture(str): repository architecture
configuration(Configuration): configuration instance
Returns: Returns:
Trigger: loaded trigger based on settings Trigger: loaded trigger based on settings
Raises:
InvalidExtension: in case if trigger could not be instantiated
"""
trigger_type = self.load_trigger_class(module_path)
try:
trigger = trigger_type(architecture, configuration)
except Exception:
raise ExtensionError(f"Could not load instance of trigger from {trigger_type} loaded from {module_path}")
return trigger
def load_trigger_class(self, module_path: str) -> Type[Trigger]:
"""
load trigger class by module path
Args:
module_path(str): module import path to load
Returns:
Type[Trigger]: loaded trigger type by module path
Raises: Raises:
InvalidExtension: in case if module cannot be loaded from the specified module path or is not a trigger InvalidExtension: in case if module cannot be loaded from the specified module path or is not a trigger
""" """
@ -156,16 +202,11 @@ class TriggerLoader(LazyLogging):
trigger_type = getattr(module, class_name, None) trigger_type = getattr(module, class_name, None)
if not isinstance(trigger_type, type): if not isinstance(trigger_type, type):
raise ExtensionError(f"{class_name} of {package_or_path} is not a type") raise ExtensionError(f"{class_name} of {package_or_path} is not a type")
if not issubclass(trigger_type, Trigger):
raise ExtensionError(f"Class {class_name} of {package_or_path} is not a Trigger subclass")
self.logger.info("loaded type %s of package %s", class_name, package_or_path) self.logger.info("loaded type %s of package %s", class_name, package_or_path)
return trigger_type
try:
trigger = trigger_type(self.architecture, self.configuration)
except Exception:
raise ExtensionError(f"Could not load instance of trigger from {class_name} of {package_or_path}")
if not isinstance(trigger, Trigger):
raise ExtensionError(f"Class {class_name} of {package_or_path} is not a Trigger")
return trigger
def on_result(self, result: Result, packages: Iterable[Package]) -> None: def on_result(self, result: Result, packages: Iterable[Package]) -> None:
""" """

97
src/ahriman/core/upload/upload_trigger.py
View File

@ -17,7 +17,7 @@
# 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 typing import Iterable from typing import Iterable, List, Type
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.triggers import Trigger from ahriman.core.triggers import Trigger
@ -34,6 +34,86 @@ class UploadTrigger(Trigger):
targets(List[str]): upload target list targets(List[str]): upload target list
""" """
CONFIGURATION_SCHEMA = {
"github": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["github"],
},
"owner": {
"type": "string",
"required": True,
},
"password": {
"type": "string",
"required": True,
},
"repository": {
"type": "string",
"required": True,
},
"timeout": {
"type": "integer",
"coerce": "integer",
},
"username": {
"type": "string",
},
},
},
"rsync": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["rsync"],
},
"command": {
"type": "list",
"coerce": "list",
"schema": {"type": "string"},
"required": True,
"empty": False,
},
"remote": {
"type": "string",
"required": True,
},
},
},
"s3": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["s3"],
},
"access_key": {
"type": "string",
"required": True,
},
"bucket": {
"type": "string",
"required": True,
},
"chunk_size": {
"type": "integer",
"coerce": "integer",
},
"region": {
"type": "string",
"required": True,
},
"secret_key": {
"type": "string",
"required": True,
},
},
},
}
def __init__(self, architecture: str, configuration: Configuration) -> None: def __init__(self, architecture: str, configuration: Configuration) -> None:
""" """
default constructor default constructor
@ -43,7 +123,20 @@ class UploadTrigger(Trigger):
configuration(Configuration): configuration instance configuration(Configuration): configuration instance
""" """
Trigger.__init__(self, architecture, configuration) Trigger.__init__(self, architecture, configuration)
self.targets = configuration.getlist("upload", "target", fallback=[]) self.targets = self.configuration_sections(configuration)
@classmethod
def configuration_sections(cls: Type[Trigger], configuration: Configuration) -> List[str]:
"""
extract configuration sections from configuration
Args:
configuration(Configuration): configuration instance
Returns:
List[str]: read configuration sections belong to this trigger
"""
return configuration.getlist("upload", "target", fallback=[])
def on_result(self, result: Result, packages: Iterable[Package]) -> None: def on_result(self, result: Result, packages: Iterable[Package]) -> None:
""" """

36
tests/ahriman/application/handlers/test_handler_validate.py
View File

@ -5,8 +5,9 @@ from pytest_mock import MockerFixture
from ahriman.application.handlers import Validate from ahriman.application.handlers import Validate
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.configuration.schema import CONFIGURATION_SCHEMA, GITREMOTE_REMOTE_PULL_SCHEMA from ahriman.core.configuration.schema import CONFIGURATION_SCHEMA
from ahriman.core.configuration.validator import Validator from ahriman.core.configuration.validator import Validator
from ahriman.core.gitremote import RemotePullTrigger, RemotePushTrigger
def _default_args(args: argparse.Namespace) -> argparse.Namespace: def _default_args(args: argparse.Namespace) -> argparse.Namespace:
@ -60,7 +61,6 @@ def test_schema(configuration: Configuration) -> None:
assert schema.pop("console") assert schema.pop("console")
assert schema.pop("email") assert schema.pop("email")
assert schema.pop("github") assert schema.pop("github")
assert schema.pop("gitremote")
assert schema.pop("html") assert schema.pop("html")
assert schema.pop("rsync") assert schema.pop("rsync")
assert schema.pop("s3") assert schema.pop("s3")
@ -69,6 +69,14 @@ def test_schema(configuration: Configuration) -> None:
assert schema == CONFIGURATION_SCHEMA assert schema == CONFIGURATION_SCHEMA
def test_schema_invalid_trigger(configuration: Configuration) -> None:
"""
must skip trigger if it caused exception on load
"""
configuration.set_option("build", "triggers", "some.invalid.trigger.path.Trigger")
assert Validate.schema("x86_64", configuration) == CONFIGURATION_SCHEMA
def test_schema_erase_required() -> None: def test_schema_erase_required() -> None:
""" """
must remove required field from dictionaries recursively must remove required field from dictionaries recursively
@ -77,24 +85,18 @@ def test_schema_erase_required() -> None:
assert "required" not in json.dumps(Validate.schema_erase_required(CONFIGURATION_SCHEMA)) assert "required" not in json.dumps(Validate.schema_erase_required(CONFIGURATION_SCHEMA))
def test_schema_insert(configuration: Configuration) -> None: def test_schema_merge() -> None:
""" """
must insert child schema to root must merge schemas correctly
""" """
result = Validate.schema_insert("x86_64", configuration, CONFIGURATION_SCHEMA, "remote-pull", erased = Validate.schema_erase_required(CONFIGURATION_SCHEMA)
lambda _: GITREMOTE_REMOTE_PULL_SCHEMA) assert Validate.schema_merge(erased, CONFIGURATION_SCHEMA) == CONFIGURATION_SCHEMA
assert result["gitremote"] == GITREMOTE_REMOTE_PULL_SCHEMA
merged = Validate.schema_merge(RemotePullTrigger.CONFIGURATION_SCHEMA, RemotePushTrigger.CONFIGURATION_SCHEMA)
def test_schema_insert_skip(configuration: Configuration) -> None: for key in RemotePullTrigger.CONFIGURATION_SCHEMA["gitremote"]["schema"]:
""" assert key in merged["gitremote"]["schema"]
must do nothing in case if there is no such section or option for key in RemotePushTrigger.CONFIGURATION_SCHEMA["gitremote"]["schema"]:
""" assert key in merged["gitremote"]["schema"]
configuration.remove_section("remote-pull")
result = Validate.schema_insert("x86_64", configuration, CONFIGURATION_SCHEMA, "remote-pull",
lambda _: GITREMOTE_REMOTE_PULL_SCHEMA)
assert result == CONFIGURATION_SCHEMA
def test_disallow_auto_architecture_run() -> None: def test_disallow_auto_architecture_run() -> None:

9
tests/ahriman/core/configuration/test_configuration.py
View File

@ -207,6 +207,15 @@ def test_gettype(configuration: Configuration) -> None:
assert provider == "s3" assert provider == "s3"
def test_gettype_with_fallback(configuration: Configuration) -> None:
"""
must return same provider name as in fallback
"""
section, provider = configuration.gettype("rsync", "x86_64", fallback="abracadabra")
assert section == "rsync"
assert provider == "abracadabra"
def test_gettype_from_section(configuration: Configuration) -> None: def test_gettype_from_section(configuration: Configuration) -> None:
""" """
must extract type from section name must extract type from section name

11
tests/ahriman/core/gitremote/test_remote_pull_trigger.py
View File

@ -4,6 +4,17 @@ from ahriman.core.configuration import Configuration
from ahriman.core.gitremote import RemotePullTrigger from ahriman.core.gitremote import RemotePullTrigger
def test_configuration_sections(configuration: Configuration) -> None:
"""
must correctly parse target list
"""
configuration.set_option("remote-pull", "target", "a b c")
assert RemotePullTrigger.configuration_sections(configuration) == ["a", "b", "c"]
configuration.remove_option("remote-pull", "target")
assert RemotePullTrigger.configuration_sections(configuration) == []
def test_on_start(configuration: Configuration, mocker: MockerFixture) -> None: def test_on_start(configuration: Configuration, mocker: MockerFixture) -> None:
""" """
must clone repo on start must clone repo on start

11
tests/ahriman/core/gitremote/test_remote_push_trigger.py
View File

@ -8,6 +8,17 @@ from ahriman.models.package import Package
from ahriman.models.result import Result from ahriman.models.result import Result
def test_configuration_sections(configuration: Configuration) -> None:
"""
must correctly parse target list
"""
configuration.set_option("remote-push", "target", "a b c")
assert RemotePushTrigger.configuration_sections(configuration) == ["a", "b", "c"]
configuration.remove_option("remote-push", "target")
assert RemotePushTrigger.configuration_sections(configuration) == []
def test_on_result(configuration: Configuration, result: Result, package_ahriman: Package, def test_on_result(configuration: Configuration, result: Result, package_ahriman: Package,
database: SQLite, mocker: MockerFixture) -> None: database: SQLite, mocker: MockerFixture) -> None:
""" """

11
tests/ahriman/core/report/test_report_trigger.py
View File

@ -5,6 +5,17 @@ from ahriman.core.report import ReportTrigger
from ahriman.models.result import Result from ahriman.models.result import Result
def test_configuration_sections(configuration: Configuration) -> None:
"""
must correctly parse target list
"""
configuration.set_option("report", "target", "a b c")
assert ReportTrigger.configuration_sections(configuration) == ["a", "b", "c"]
configuration.remove_option("report", "target")
assert ReportTrigger.configuration_sections(configuration) == []
def test_on_result(configuration: Configuration, mocker: MockerFixture) -> None: def test_on_result(configuration: Configuration, mocker: MockerFixture) -> None:
""" """
must run report for specified targets must run report for specified targets

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

@ -28,4 +28,4 @@ def trigger_loader(configuration: Configuration) -> TriggerLoader:
Returns: Returns:
TriggerLoader: trigger loader test instance TriggerLoader: trigger loader test instance
""" """
return TriggerLoader("x86_64", configuration) return TriggerLoader.load("x86_64", configuration)

55
tests/ahriman/core/triggers/test_trigger.py
View File

@ -1,9 +1,64 @@
from unittest.mock import MagicMock from unittest.mock import MagicMock
from ahriman.core.configuration import Configuration
from ahriman.core.report import ReportTrigger
from ahriman.core.triggers import Trigger from ahriman.core.triggers import Trigger
from ahriman.models.result import Result from ahriman.models.result import Result
def test_configuration_schema(configuration: Configuration) -> None:
"""
must return used configuration schema
"""
section = "console"
configuration.set_option("report", "target", section)
expected = {section: ReportTrigger.CONFIGURATION_SCHEMA[section]}
assert ReportTrigger.configuration_schema("x86_64", configuration) == expected
def test_configuration_schema_no_section(configuration: Configuration) -> None:
"""
must return nothing in case if section doesn't exist
"""
section = "abracadabra"
configuration.set_option("report", "target", section)
assert ReportTrigger.configuration_schema("x86_64", configuration) == {}
def test_configuration_schema_no_schema(configuration: Configuration) -> None:
"""
must return nothing in case if schema doesn't exist
"""
section = "abracadabra"
configuration.set_option("report", "target", section)
configuration.set_option(section, "key", "value")
assert ReportTrigger.configuration_schema("x86_64", configuration) == {}
def test_configuration_schema_empty() -> None:
"""
must return default schema if no configuration set
"""
assert ReportTrigger.configuration_schema("x86_64", None) == ReportTrigger.CONFIGURATION_SCHEMA
def test_configuration_schema_variables(configuration: Configuration) -> None:
"""
must return empty schema
"""
assert Trigger.CONFIGURATION_SCHEMA == {}
assert Trigger.CONFIGURATION_SCHEMA_FALLBACK is None
def test_configuration_sections(configuration: Configuration) -> None:
"""
must return empty section list
"""
assert Trigger.configuration_sections(configuration) == []
def test_on_result(trigger: Trigger) -> None: def test_on_result(trigger: Trigger) -> None:
""" """
must pass execution nto run method must pass execution nto run method

84
tests/ahriman/core/triggers/test_trigger_loader.py
View File

@ -5,75 +5,97 @@ from pytest_mock import MockerFixture
from ahriman.core.configuration import Configuration from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import ExtensionError from ahriman.core.exceptions import ExtensionError
from ahriman.core.report import ReportTrigger
from ahriman.core.triggers import TriggerLoader from ahriman.core.triggers import TriggerLoader
from ahriman.models.package import Package from ahriman.models.package import Package
from ahriman.models.result import Result from ahriman.models.result import Result
def test_load_trigger_package(trigger_loader: TriggerLoader) -> None: def test_selected_triggers(configuration: Configuration) -> None:
""" """
must load trigger from package must return used triggers
""" """
assert trigger_loader.load_trigger("ahriman.core.report.ReportTrigger") configuration.set_option("build", "triggers", "a b c")
assert TriggerLoader.selected_triggers(configuration) == ["a", "b", "c"]
configuration.remove_option("build", "triggers")
assert TriggerLoader.selected_triggers(configuration) == []
def test_load_trigger_package_invalid_import(trigger_loader: TriggerLoader, mocker: MockerFixture) -> None: def test_load_trigger(trigger_loader: TriggerLoader, configuration: Configuration) -> None:
""" """
must raise InvalidExtension on invalid import must load trigger
""" """
mocker.patch("ahriman.core.triggers.trigger_loader.importlib.import_module", side_effect=ModuleNotFoundError()) loaded = trigger_loader.load_trigger("ahriman.core.report.ReportTrigger", "x86_64", configuration)
with pytest.raises(ExtensionError): assert loaded
trigger_loader.load_trigger("random.module") assert isinstance(loaded, ReportTrigger)
def test_load_trigger_package_not_trigger(trigger_loader: TriggerLoader) -> None: def test_load_trigger_package_error_on_creation(trigger_loader: TriggerLoader, configuration: Configuration,
""" mocker: MockerFixture) -> None:
must raise InvalidExtension if imported module is not a type
"""
with pytest.raises(ExtensionError):
trigger_loader.load_trigger("ahriman.core.util.check_output")
def test_load_trigger_package_error_on_creation(trigger_loader: TriggerLoader, mocker: MockerFixture) -> None:
""" """
must raise InvalidException on trigger initialization if any exception is thrown must raise InvalidException on trigger initialization if any exception is thrown
""" """
mocker.patch("ahriman.core.triggers.trigger.Trigger.__init__", side_effect=Exception()) mocker.patch("ahriman.core.triggers.trigger.Trigger.__init__", side_effect=Exception())
with pytest.raises(ExtensionError): with pytest.raises(ExtensionError):
trigger_loader.load_trigger("ahriman.core.report.ReportTrigger") trigger_loader.load_trigger("ahriman.core.report.ReportTrigger", "x86_64", configuration)
def test_load_trigger_package_is_not_trigger(trigger_loader: TriggerLoader) -> None: def test_load_trigger_class_package(trigger_loader: TriggerLoader) -> None:
"""
must load trigger class from package
"""
assert trigger_loader.load_trigger_class("ahriman.core.report.ReportTrigger") == ReportTrigger
def test_load_trigger_class_package_invalid_import(trigger_loader: TriggerLoader, mocker: MockerFixture) -> None:
"""
must raise InvalidExtension on invalid import
"""
mocker.patch("ahriman.core.triggers.trigger_loader.importlib.import_module", side_effect=ModuleNotFoundError())
with pytest.raises(ExtensionError):
trigger_loader.load_trigger_class("random.module")
def test_load_trigger_class_package_not_trigger(trigger_loader: TriggerLoader) -> None:
"""
must raise InvalidExtension if imported module is not a type
"""
with pytest.raises(ExtensionError):
trigger_loader.load_trigger_class("ahriman.core.util.check_output")
def test_load_trigger_class_package_is_not_trigger(trigger_loader: TriggerLoader) -> None:
""" """
must raise InvalidExtension if loaded class is not a trigger must raise InvalidExtension if loaded class is not a trigger
""" """
with pytest.raises(ExtensionError): with pytest.raises(ExtensionError):
trigger_loader.load_trigger("ahriman.core.sign.gpg.GPG") trigger_loader.load_trigger_class("ahriman.core.sign.gpg.GPG")
def test_load_trigger_path(trigger_loader: TriggerLoader, resource_path_root: Path) -> None: def test_load_trigger_class_path(trigger_loader: TriggerLoader, resource_path_root: Path) -> None:
""" """
must load trigger from path must load trigger class from path
""" """
path = resource_path_root.parent.parent / "src" / "ahriman" / "core" / "report" / "report_trigger.py" path = resource_path_root.parent.parent / "src" / "ahriman" / "core" / "report" / "__init__.py"
assert trigger_loader.load_trigger(f"{path}.ReportTrigger") assert trigger_loader.load_trigger_class(f"{path}.ReportTrigger") == ReportTrigger
def test_load_trigger_path_directory(trigger_loader: TriggerLoader, resource_path_root: Path) -> None: def test_load_trigger_class_path_directory(trigger_loader: TriggerLoader, resource_path_root: Path) -> None:
""" """
must raise InvalidExtension if provided import path is directory must raise InvalidExtension if provided import path is directory
""" """
path = resource_path_root.parent.parent / "src" / "ahriman" / "core" / "report" path = resource_path_root.parent.parent / "src" / "ahriman" / "core" / "report"
with pytest.raises(ExtensionError): with pytest.raises(ExtensionError):
trigger_loader.load_trigger(f"{path}.ReportTrigger") trigger_loader.load_trigger_class(f"{path}.ReportTrigger")
def test_load_trigger_path_not_found(trigger_loader: TriggerLoader) -> None: def test_load_trigger_class_path_not_found(trigger_loader: TriggerLoader) -> None:
""" """
must raise InvalidExtension if file cannot be found must raise InvalidExtension if file cannot be found
""" """
with pytest.raises(ExtensionError): with pytest.raises(ExtensionError):
trigger_loader.load_trigger("/some/random/path.py.SomeRandomModule") trigger_loader.load_trigger_class("/some/random/path.py.SomeRandomModule")
def test_on_result(trigger_loader: TriggerLoader, package_ahriman: Package, mocker: MockerFixture) -> None: def test_on_result(trigger_loader: TriggerLoader, package_ahriman: Package, mocker: MockerFixture) -> None:
@ -119,9 +141,11 @@ def test_on_stop_with_on_start(configuration: Configuration, mocker: MockerFixtu
""" """
must call on_stop on exit if on_start was called must call on_stop on exit if on_start was called
""" """
mocker.patch("ahriman.core.upload.UploadTrigger.on_start")
mocker.patch("ahriman.core.report.ReportTrigger.on_start")
on_stop_mock = mocker.patch("ahriman.core.triggers.trigger_loader.TriggerLoader.on_stop") on_stop_mock = mocker.patch("ahriman.core.triggers.trigger_loader.TriggerLoader.on_stop")
trigger_loader = TriggerLoader("x86_64", configuration) trigger_loader = TriggerLoader.load("x86_64", configuration)
trigger_loader.on_start() trigger_loader.on_start()
del trigger_loader del trigger_loader
on_stop_mock.assert_called_once_with() on_stop_mock.assert_called_once_with()
@ -133,7 +157,7 @@ def test_on_stop_without_on_start(configuration: Configuration, mocker: MockerFi
""" """
on_stop_mock = mocker.patch("ahriman.core.triggers.trigger_loader.TriggerLoader.on_stop") on_stop_mock = mocker.patch("ahriman.core.triggers.trigger_loader.TriggerLoader.on_stop")
trigger_loader = TriggerLoader("x86_64", configuration) trigger_loader = TriggerLoader.load("x86_64", configuration)
del trigger_loader del trigger_loader
on_stop_mock.assert_not_called() on_stop_mock.assert_not_called()

11
tests/ahriman/core/upload/test_upload_trigger.py
View File

@ -5,6 +5,17 @@ from ahriman.core.upload import UploadTrigger
from ahriman.models.result import Result from ahriman.models.result import Result
def test_configuration_sections(configuration: Configuration) -> None:
"""
must correctly parse target list
"""
configuration.set_option("upload", "target", "a b c")
assert UploadTrigger.configuration_sections(configuration) == ["a", "b", "c"]
configuration.remove_option("upload", "target")
assert UploadTrigger.configuration_sections(configuration) == []
def test_on_result(configuration: Configuration, mocker: MockerFixture) -> None: def test_on_result(configuration: Configuration, mocker: MockerFixture) -> None:
""" """
must run report for specified targets must run report for specified targets

2
tests/testresources/core/ahriman.ini
View File

@ -45,7 +45,7 @@ push_url = https://github.com/arcan1s/repository.git
pull_url = https://github.com/arcan1s/repository.git pull_url = https://github.com/arcan1s/repository.git
[report] [report]
target = target = console
[email] [email]
host = 127.0.0.1 host = 127.0.0.1