Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions src/_balder/collector.py
Original file line number Diff line number Diff line change
Expand Up @@ -675,14 +675,19 @@ def _exchange_strings_with_objects(self):
This method exchanges all strings (that can be used in decorators) are exchanged with their real objects. It
secures this for all :class:`Device` and :class:`VDevice` references inside the session.
"""
all_scenarios = [scenario for scenario in self.all_collected_scenarios_with_mro
if issubclass(scenario, Scenario) and scenario != Scenario]
all_setups = [setup for setup in self.all_collected_setups_with_mro
if issubclass(setup, Setup) and setup != Setup]

# resolve connection Device strings (for all devices that are directly defined inside the scenario/setup)
for cur_scenario_or_setup in self.all_scenarios_and_setups:
for cur_scenario_or_setup in all_scenarios + all_setups:
cur_scenario_or_setup_controller = NormalScenarioSetupController.get_for(cur_scenario_or_setup)
for cur_device in cur_scenario_or_setup_controller.get_all_inner_device_classes():
DeviceController.get_for(cur_device).resolve_connection_device_strings()

# resolve connection VDevice strings (for all absolute devices of this scenario/setup)
for cur_scenario_or_setup in self.all_scenarios_and_setups:
for cur_scenario_or_setup in all_scenarios + all_setups:
cur_scenario_or_setup_controller = NormalScenarioSetupController.get_for(cur_scenario_or_setup)
for cur_device in cur_scenario_or_setup_controller.get_all_abs_inner_device_classes():
DeviceController.get_for(cur_device).resolve_mapped_vdevice_strings()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
from typing import Literal, Union, List
from datetime import datetime

import pathlib
import balder
import argparse
import logging
from multiprocessing import Queue
from .lib.utils import FixtureReturn

logger = logging.getLogger(__file__)


class MyTestException(Exception):
pass


class RuntimeObserver:
"""This is a helper object, that will be used from this test environment to observe the execution order"""
queue: Union[Queue, None] = None

error_throwing = {}

@staticmethod
def add_entry(file: Union[str, pathlib.Path], cls: object, meth: callable, msg: str,
category: Literal["fixture", "testcase", "feature"] = None,
part: Literal["construction", "teardown"] = None):
"""
adds a new entry and sends it over the queue

:param file: the full filepath where the log will be generated

:param cls: the class object, the entry is generated in

:param meth: the method name, the entry is generated in

:param msg: the message that should be inserted into the entry

:param category: optional string of the category the entry is from

:param part: optional string of the sub part the entry is from
"""
if hasattr(meth, 'fn'):
meth = meth.fn
new_dataset = {
"timestamp": datetime.now(), "file": file, "cls": "" if cls is None else cls.__name__,
"meth": meth.__name__, "msg": msg, "category": category, "part": part
}
logger.info("{:22} | {:20} | {:30} | {:12} | {:15} | {}".format(
pathlib.Path(file).parts[-1], "" if cls is None else cls.__name__, "" if meth is None else meth.__name__,
"" if category is None else category, "" if part is None else part, "" if msg is None else msg))

RuntimeObserver.queue.put(new_dataset)
# check if we have to throw the error
error_throwing_required = len(RuntimeObserver.error_throwing) > 0
for cur_key in RuntimeObserver.error_throwing.keys():
new_dataset_val = new_dataset[cur_key]
if callable(new_dataset_val):
new_dataset_val = new_dataset_val.__name__
if new_dataset_val != RuntimeObserver.error_throwing[cur_key]:
error_throwing_required = False
break
if error_throwing_required:
raise MyTestException(f'raise test triggered exception for `{str(RuntimeObserver.error_throwing)}`')


class MyErrorThrowingPlugin(balder.BalderPlugin):
"""
This is a plugin that reads the values from console arguments and sets these values into the
:class:`RuntimeObserver`. The static method `RuntimeObserver.add_entry` will automatically throw an exception on the
given position.
"""

def addoption(self, argument_parser: argparse.ArgumentParser):
argument_parser.add_argument('--test-error-file', help='the file id, the error should be thrown in')
argument_parser.add_argument('--test-error-cls', help='the class id, the error should be thrown in')
argument_parser.add_argument('--test-error-meth', help='the meth id, the error should be thrown in')
argument_parser.add_argument('--test-error-part', help='the part (`construct` or `teardown`), the error should '
'be thrown in - only for fixtures')

def modify_collected_pyfiles(self, pyfiles: List[pathlib.Path]) -> List[pathlib.Path]:
# use this method to set the values
RuntimeObserver.error_throwing = {}
if self.balder_session.parsed_args.test_error_file:
path = pathlib.Path(self.balder_session.parsed_args.test_error_file)
if not path.is_absolute():
path = str(self.balder_session.working_dir.joinpath(path))
RuntimeObserver.error_throwing['file'] = path
if self.balder_session.parsed_args.test_error_cls:
RuntimeObserver.error_throwing['cls'] = self.balder_session.parsed_args.test_error_cls
if self.balder_session.parsed_args.test_error_meth:
RuntimeObserver.error_throwing['meth'] = self.balder_session.parsed_args.test_error_meth
if self.balder_session.parsed_args.test_error_part:
RuntimeObserver.error_throwing['part'] = self.balder_session.parsed_args.test_error_part
return pyfiles


@balder.fixture(level="session")
def balderglob_fixture_session():
RuntimeObserver.add_entry(__file__, None, balderglob_fixture_session, "begin execution CONSTRUCTION of fixture",
category="fixture", part="construction")

yield FixtureReturn.BALDERGLOB_SESSION

RuntimeObserver.add_entry(__file__, None, balderglob_fixture_session, "begin execution TEARDOWN of fixture",
category="fixture", part="teardown")


@balder.fixture(level="setup")
def balderglob_fixture_setup():
RuntimeObserver.add_entry(__file__, None, balderglob_fixture_setup, "begin execution CONSTRUCTION of fixture",
category="fixture", part="construction")

yield FixtureReturn.BALDERGLOB_SETUP

RuntimeObserver.add_entry(__file__, None, balderglob_fixture_setup, "begin execution TEARDOWN of fixture",
category="fixture", part="teardown")


@balder.fixture(level="scenario")
def balderglob_fixture_scenario():
RuntimeObserver.add_entry(__file__, None, balderglob_fixture_scenario, "begin execution CONSTRUCTION of fixture",
category="fixture", part="construction")

yield FixtureReturn.BALDERGLOB_SCENARIO

RuntimeObserver.add_entry(__file__, None, balderglob_fixture_scenario, "begin execution TEARDOWN of fixture",
category="fixture", part="teardown")


@balder.fixture(level="variation")
def balderglob_fixture_variation():
RuntimeObserver.add_entry(__file__, None, balderglob_fixture_variation, "begin execution CONSTRUCTION of fixture",
category="fixture", part="construction")

yield FixtureReturn.BALDERGLOB_VARIATION

RuntimeObserver.add_entry(__file__, None, balderglob_fixture_variation, "begin execution TEARDOWN of fixture",
category="fixture", part="teardown")


@balder.fixture(level="testcase")
def balderglob_fixture_testcase():
RuntimeObserver.add_entry(__file__, None, balderglob_fixture_testcase, "begin execution CONSTRUCTION of fixture",
category="fixture", part="construction")

yield FixtureReturn.BALDERGLOB_TESTCASE

RuntimeObserver.add_entry(__file__, None, balderglob_fixture_testcase, "begin execution TEARDOWN of fixture",
category="fixture", part="teardown")
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
import balder


@balder.insert_into_tree()
class AConnection(balder.Connection):
pass


@balder.insert_into_tree()
class BConnection(balder.Connection):
pass
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
import balder
from ..balderglob import RuntimeObserver


class FeatureI(balder.Feature):

def do_something(self):
RuntimeObserver.add_entry(
__file__, FeatureI, FeatureI.do_something, "enter `FeatureI.do_something`", category="feature")


class FeatureIOverwritten(FeatureI):

def do_something(self):
RuntimeObserver.add_entry(
__file__, FeatureIOverwritten, FeatureIOverwritten.do_something, "enter `FeatureIChild.do_something`",
category="feature")


class FeatureII(balder.Feature):

def do_something(self):
RuntimeObserver.add_entry(
__file__, FeatureII, FeatureII.do_something, "enter `FeatureII.do_something`", category="feature")


class NewlyDefinedFeature(balder.Feature):
"""this feature will be instantiated in `ScenarioAChild` (and not in `ScenarioAParent`)"""
def do_something(self):
RuntimeObserver.add_entry(
__file__, NewlyDefinedFeature, NewlyDefinedFeature.do_something, "enter `NewlyDefinedFeature.do_something`",
category="feature")


class FeatureIII(balder.Feature):

def do_something(self):
RuntimeObserver.add_entry(
__file__, FeatureIII, FeatureIII.do_something, "enter `FeatureIII.do_something`", category="feature")


class FeatureIV(balder.Feature):

def do_something(self):
RuntimeObserver.add_entry(
__file__, FeatureIV, FeatureIV.do_something, "enter `FeatureIV.do_something`", category="feature")
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@


class FixtureReturn:
"""helper const class for return values"""
BALDERGLOB_SESSION = "balderglob_session_fixt"
BALDERGLOB_SETUP = "balderglob_setup_fixt"
BALDERGLOB_SCENARIO = "balderglob_scenario_fixt"
BALDERGLOB_VARIATION = "balderglob_variation_fixt"
BALDERGLOB_TESTCASE = "balderglob_testcase_fixt"

SETUP_SESSION = "setup_session_fixt"
SETUP_SETUP = "setup_setup_fixt"
SETUP_SCENARIO = "setup_scenario_fixt"
SETUP_VARIATION = "setup_variation_fixt"
SETUP_TESTCASE = "setup_testcase_fixt"

SCENARIO_SESSION = "scenario_session_fixt"
SCENARIO_SETUP = "scenario_setup_fixt"
SCENARIO_SCENARIO = "scenario_scenario_fixt"
SCENARIO_VARIATION = "scenario_variation_fixt"
SCENARIO_TESTCASE = "scenario_testcase_fixt"

Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
import balder
import logging
from ..lib.features import FeatureIOverwritten, FeatureII, NewlyDefinedFeature
from ..lib.connections import AConnection
from ..balderglob import RuntimeObserver
from .scenario_a_parent import ScenarioAParent

logger = logging.getLogger(__file__)


class ScenarioAChild(ScenarioAParent):
"""This is the CHILD scenario of category A"""

class ScenarioDevice1(ScenarioAParent.ScenarioDevice1):
i = FeatureIOverwritten()

@balder.connect(ScenarioDevice1, over_connection=AConnection)
class ScenarioDevice2(ScenarioAParent.ScenarioDevice2):
ii = FeatureII()
new = NewlyDefinedFeature()

def test_a_1(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.test_a_1, category="testcase",
msg=f"execute Test `{ScenarioAChild.test_a_1.__qualname__}`")
self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

def test_a_2(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.test_a_2, category="testcase",
msg=f"execute Test `{ScenarioAChild.test_a_2.__qualname__}`")
self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

@balder.fixture(level="session")
def fixture_session(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_session, category="fixture",
part="construction", msg="begin execution CONSTRUCTION of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

yield

RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_session, category="fixture",
part="teardown", msg="begin execution TEARDOWN of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

@balder.fixture(level="setup")
def fixture_setup(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_setup, category="fixture",
part="construction", msg="begin execution CONSTRUCTION of fixture")
self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

yield

RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_setup, category="fixture",
part="teardown", msg="begin execution TEARDOWN of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

@balder.fixture(level="scenario")
def fixture_scenario(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_scenario, category="fixture",
part="construction", msg="begin execution CONSTRUCTION of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

yield

RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_scenario, category="fixture",
part="teardown", msg="begin execution TEARDOWN of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

@balder.fixture(level="variation")
def fixture_variation(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_variation, category="fixture",
part="construction", msg="begin execution CONSTRUCTION of fixture")
self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

yield

RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_variation, category="fixture",
part="teardown", msg="begin execution TEARDOWN of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

@balder.fixture(level="testcase")
def fixture_testcase(self):
RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_testcase, category="fixture",
part="construction", msg="begin execution CONSTRUCTION of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()

yield

RuntimeObserver.add_entry(__file__, ScenarioAChild, ScenarioAChild.fixture_testcase, category="fixture",
part="teardown", msg="begin execution TEARDOWN of fixture")

self.ScenarioDevice1.i.do_something()
self.ScenarioDevice2.ii.do_something()
self.ScenarioDevice2.new.do_something()
Loading
Loading