import types
from ms_service_metric.core.handler import MetricHandler, HandlerType
from ms_service_metric.core.symbol import Symbol
from ms_service_metric.metrics.metrics_manager import MetricType
class DummyWatcher:
def watch_module(self, _module_name, _callback):
return
def unwatch_module(self, _module_name, _callback):
return
class DummyManager:
def is_updating(self):
return False
class DummyTarget:
def foo(self, x):
return x * 2
def test_given_wrap_handler_when_module_loaded_then_wrapped_result_unhook_restores():
module_path = __name__
symbol_path = f"{module_path}:DummyTarget.foo"
watcher = DummyWatcher()
manager = DummyManager()
symbol = Symbol(symbol_path, watcher=watcher, manager=manager)
def wrap_handler(ori, *args, **kwargs):
return ori(*args, **kwargs) + 1
handler = MetricHandler(
name="h1",
symbol_info={"symbol_path": symbol_path},
hook_func=wrap_handler,
min_version=None,
max_version=None,
metrics_config=[],
)
hook_type, _ = handler.get_hook_func(DummyTarget.foo)
assert hook_type == HandlerType.WRAP
symbol.add_handler(handler)
event = types.SimpleNamespace(module_name=module_path)
symbol._on_module_loaded(event)
assert DummyTarget().foo(5) == 11
symbol.unhook()
assert DummyTarget().foo(5) == 10