Source code for netdef.testutils

import datetime
import inspect
import os
from unittest.mock import Mock

from netdef.Engines.expression.Expression import Expression
from netdef.Rules.utils import get_module_from_string
from netdef.Shared import SharedConfig
from netdef.Shared.Shared import Shared
from netdef.Sources.BaseSource import BaseSource, StatusCode

[docs]class MockShared(Shared): def __init__(self, config_string=""): identifier = "" install_path = None proj_path = None read_from_files = False if "[general]" not in config_string: config_string = "[general]\nversion=1\n{}".format(config_string) default_config_string = config_string self.config = SharedConfig.Config( identifier, install_path, proj_path, default_config_string, read_from_files ) self.sources = Mock() self.expressions = Mock() self.restart_on_exit = False
[docs]class MockSource: def __init__(self, expression, source): self.expression = expression self.source = source
[docs] def update_value( self, val, stime=None, stat_none=False, stat_init=False, stat_good=False, stat_invalid=False, run_expression=True, ): "A Helper function to update values in expression" src = self.source src.get = val if stat_good: src.status_code = StatusCode.GOOD elif stat_init: src.status_code = StatusCode.INITIAL elif stat_invalid: src.status_code = StatusCode.INVALID elif stat_none: src.status_code = StatusCode.NONE else: src.status_code = StatusCode.NONE if isinstance(stime, datetime.datetime): src.source_time = stime else: src.source_time = datetime.datetime.utcnow() if run_expression: args = self.expression.get_args(self.source) kwargs = self.expression.get_kwargs() return self.expression.execute(args, kwargs) return None
[docs] def assert_value(self, value): "A helper function to assert value and timestamp" assert isinstance(self.source.set_source_time, datetime.datetime) assert self.source.set_value == value
[docs] def assert_called(self): self.source.set_callback.assert_called()
[docs] def assert_called_once(self): self.source.set_callback.assert_called_once()
[docs] def assert_called_with(self, value): self.source.set_callback.assert_called_with( self.source, value, self.source.set_source_time )
[docs] def assert_called_once_with(self, value): self.source.set_callback.assert_called_once_with( self.source, value, self.source.set_source_time )
[docs] def assert_any_call(self, value): self.source.set_callback.assert_any_call( self.source, value, self.source.set_source_time )
[docs] def assert_not_called(self): self.source.set_callback.assert_not_called()
@property def call_count(self): return self.source.set_callback.call_count @property def call_args(self): args, kwargs = self.source.set_callback.call_args src, val, stime = args return val @property def call_args_list(self): call_args_list = self.source.set_callback.call_args_list return [args[1] for args, kwargs in call_args_list]
[docs]class MockExpression: """ Example:: from netdef.testutils import MockExpression def test_hello(): mock = MockExpression( module="config/", intern=InternalSource("generic"), cmd=CmdSource("echo hello") ) mock.intern.update_value(None, stat_init=True) mock.cmd.assert_called_once_with("world") mock.intern.assert_not_called() """ def __init__(self, **kwargs): self._kwargs = kwargs self._expr = None self._pymod = None for k, v in self._kwargs.items(): if isinstance(v, Expression): self._expr = v break elif isinstance(v, type(os)): self._pymod = v elif isinstance(v, str) and k == "module": self._pymod = get_module_from_string( v, __package__, os.getcwd(), "testutils", "mockexpression" ) if self._pymod: func = self._kwargs.get("expression", "expression") self._expr = Expression( getattr(self._pymod, func), self._pymod.__file__ ) break assert not self._expr is None for k, v in self._kwargs.items(): if isinstance(v, BaseSource): setattr(self, k, MockSource(self._expr, v)) v.register_set_callback(Mock()) for arg in inspect.signature(self._expr.expression).parameters.keys(): self._expr.add_arg(self._kwargs[arg])
[docs] def get_module(self): "Returns the expression module" return self._pymod
[docs] def set_init_values(self, **kwargs): for k, v in kwargs.items(): attr = getattr(self, k) attr.update_value(val=v, stat_init=True, run_expression=False)
[docs] def set_none_values(self, **kwargs): for k, v in kwargs.items(): attr = getattr(self, k) attr.update_value(val=v, stat_none=True, run_expression=False)
[docs] def get_callbacks(self): return ((arg, arg.set_callback) for arg in self._expr.args)
def __getattr__(self, name): # this is only to please pylint raise AttributeError("MockExpression has no attribute '{}'".format(name))