test_search_hooks.py
2.32 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from __future__ import annotations
from types import SimpleNamespace
from apps.web_api.bootstrap.search_hooks import (
build_search_hook_bindings,
)
def test_build_search_hook_bindings_delegates_to_hook_source_callables():
calls: list[tuple[str, dict[str, object]]] = []
def resolve_search_query(**kwargs):
calls.append(("resolve", kwargs))
return "task-1", "museum research"
def dispatch_search_request(**kwargs):
calls.append(("dispatch", kwargs))
return {"success": True, "accepted": True}
hook_source = SimpleNamespace(
resolve_search_query=resolve_search_query,
dispatch_search_request=dispatch_search_request,
)
bindings = build_search_hook_bindings(hook_source)
assert bindings.resolve_search_query(payload={"q": "museum"}, research_task_service="svc") == (
"task-1",
"museum research",
)
assert bindings.dispatch_search_request(query="museum research", process_registry="registry") == {
"success": True,
"accepted": True,
}
assert calls == [
("resolve", {"payload": {"q": "museum"}, "research_task_service": "svc"}),
("dispatch", {"query": "museum research", "process_registry": "registry"}),
]
def test_build_search_hook_bindings_reads_updated_hook_source_callables():
def initial_resolve_search_query(**kwargs):
return "old-task", str(kwargs["payload"])
def replacement_resolve_search_query(**kwargs):
return "new-task", str(kwargs["payload"]).upper()
def initial_dispatch_search_request(**kwargs):
return {"message": f"old:{kwargs['query']}"}
def replacement_dispatch_search_request(**kwargs):
return {"message": f"new:{kwargs['query']}"}
hook_source = SimpleNamespace(
resolve_search_query=initial_resolve_search_query,
dispatch_search_request=initial_dispatch_search_request,
)
bindings = build_search_hook_bindings(hook_source)
hook_source.resolve_search_query = replacement_resolve_search_query
hook_source.dispatch_search_request = replacement_dispatch_search_request
assert bindings.resolve_search_query(payload={"q": "museum"}, research_task_service=None) == (
"new-task",
"{'Q': 'MUSEUM'}",
)
assert bindings.dispatch_search_request(query="museum") == {"message": "new:museum"}