Commit 243de78e authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

Add Assembly logic and testing

parent ff608796
Pipeline #26573 passed with stages
in 3 minutes and 1 second
......@@ -17,16 +17,22 @@
# along with this program. If not, see
import logging
from abc import ABC, abstractmethod
from rabotnik import Rule
logger = logging.getLogger(__name__)
class Assembly(ABC):
"""Basic rule assembly class"""
class Assembly:
"""Rule assembly.
An `Assembly` groups `Rules`. The `Rules` are passed as a list which will be
evaluated in list order when `` is called.
def __init__(self, rules: list[Rule]):
self.rules = rules
def run(self):
async def run(self, *args, **kwargs):
"""Main function to run the rules defined in an assembly"""
for rule in self.rules:
await rule.evaluate(*args, **kwargs)
......@@ -20,6 +20,7 @@ import os
import asyncio
import pytest
from rabotnik import Rule
DBHOST = os.environ.get("RABOTNIK_TEST_HOST", "localhost")
......@@ -72,6 +73,18 @@ class CallbackCounter:
return f"""{self.__class__} {hex(id(self))} called {self.call_count} times"""
class DemoRule(Rule, CallbackCounter):
"""A `Rule` that logs how often it has been invoked for testing."""
async def evaluate(self, *args, **kwargs):
self(*args, **kwargs)
def demo_rule():
yield DemoRule()
def counting_callback():
"""Function scoped fixture yielding a `CallbackCounter`."""
#!/usr/bin/env python3
# Copyright (C) 2020-2021:
# Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero 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
# General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see
import pytest
from rabotnik import Assembly
async def test_assembly(demo_rule):
assembly = Assembly(rules=[demo_rule])
payload = {"id": "a"}
assert assembly.rules[0].call_arguments == [payload]
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment