Source code for cfme.fixtures.networks

import re
from argparse import Namespace
from contextlib import contextmanager

import pytest

from cfme.exceptions import NeedleNotFoundInLog
from cfme.utils.log import logger
from cfme.utils.wait import TimedOutError
from cfme.utils.wait import wait_for

[docs]@pytest.fixture def targeted_refresh(merkyl_setup, merkyl_inspector): """ This fixture tests whether targeted refresh was triggered for given targets. It basically tails evm.log to see whether a line like this has occured: ``` [Collection of targets with id: [{:ems_ref=>"b35d3afe-6f19-4da8-b1ee-b79de433cace"}, ... ]] ``` for each target and raises an error if not. Usage: def test_something(targeted_refresh) trigger_targeted_refresh() <- some function that will trigger targeted refresh with targeted_refresh.target_timeout(): targeted_refresh.register_target('ref-123456', 'Subnet named TEST') targeted_refresh.register_target('ref-aaaabb', 'Router named TEST') """ evm_log = '/var/www/miq/vmdb/log/evm.log' needle_template = r'^.*Collection of targets with id.*:ems_ref=>"{}".*$' merkyl_inspector.add_log(evm_log) merkyl_inspector.reset_log(evm_log) targets = set() # { (ems_ref, comment), (ems_ref, comment), ... } def check_log():'Looking for %s needles in evm.log: %s', len(targets), [t[1] for t in targets]) content = merkyl_inspector.get_log(evm_log) for target in set(targets): if target[0].search(content):'Found needle %s', target[1]) targets.remove(target) return len(targets) == 0 @contextmanager def timeout(): yield try: wait_for(check_log, delay=5, num_sec=60) except TimedOutError: raise NeedleNotFoundInLog('Targeted refresh did not trigger for:\n{}'.format( ',\n'.join(['- ' + t[1] for t in targets]))) def register_target(ems_ref, comment): targets.add((re.compile(needle_template.format(ems_ref), re.MULTILINE), comment)) yield Namespace(register_target=register_target, timeout=timeout)