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
|
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from pykka import Actor
if TYPE_CHECKING:
from collections.abc import Iterator
from pykka import ActorProxy, Future
from tests.types import Runtime
class DynamicMethodActor(Actor):
def add_method(self, name: str) -> None:
setattr(self, name, lambda: "returned by " + name)
def use_foo_through_self_proxy(self) -> Future[str]:
return self.actor_ref.proxy().foo() # type: ignore[no-any-return]
@pytest.fixture(scope="module")
def actor_class(runtime: Runtime) -> type[DynamicMethodActor]:
class DynamicMethodActorImpl(DynamicMethodActor, runtime.actor_class): # type: ignore[name-defined]
pass
return DynamicMethodActorImpl
@pytest.fixture
def proxy(
actor_class: type[DynamicMethodActor],
) -> Iterator[ActorProxy[DynamicMethodActor]]:
proxy = actor_class.start().proxy()
yield proxy
proxy.stop()
def test_can_call_method_that_was_added_at_runtime(
proxy: ActorProxy[DynamicMethodActor],
) -> None:
# We need to .get() after .add_method() to be sure that the method has
# been added before we try to use it through the proxy.
proxy.add_method("foo").get()
assert proxy.foo().get() == "returned by foo"
def test_can_proxy_itself_and_use_attrs_added_at_runtime(
proxy: ActorProxy[DynamicMethodActor],
) -> None:
# We don't need to .get() after .add_method() here, because the actor
# will process the .add_method() call before processing the
# .use_foo_through_self_proxy() call, which again will use the new
# method, .foo().
proxy.add_method("foo")
outer_future = proxy.use_foo_through_self_proxy()
inner_future = outer_future.get(timeout=1)
result = inner_future.get(timeout=1)
assert result == "returned by foo"
|
