Skip to content

Commit

Permalink
Add tests to verify autoproxy server object lifetime
Browse files Browse the repository at this point in the history
An autoproxy consists of a server object, and a client
proxy. When the client proxy is released, the server
object should be as well.

Add a few simple tests to verify superficially
that this does indeed happen.

Note: these tests currently fail.
  • Loading branch information
TheSven73 committed Jul 17, 2024
1 parent 57187c3 commit e167fb3
Showing 1 changed file with 98 additions and 0 deletions.
98 changes: 98 additions & 0 deletions tests/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
Pyro - Python Remote Objects. Copyright by Irmen de Jong (irmen@razorvine.net).
"""

from functools import lru_cache
import time
import threading
import serpent
Expand Down Expand Up @@ -185,6 +186,103 @@ def testDaemonConnectFail(self):
assert "rigged connection failure" in message


@lru_cache(maxsize=None)
class LiveObjects:
def __init__(self):
self.__objs: dict[int, str] = {}

def add(self, obj):
self.__objs[id(obj)] = str(obj)

def remove(self, obj):
self.__objs.pop(id(obj), None)

def clear(self):
self.__objs.clear()

def get_alive_names(self):
return self.__objs.values()


@Pyro5.server.expose
class Child:
def __init__(self, name: str):
self.__name = name
LiveObjects().add(self)

def __del__(self):
LiveObjects().remove(self)

def __repr__(self):
return self.__name

def ping(self):
return f"I am {self.__name}"


@Pyro5.server.expose
class Parent:
def __init__(self):
self.__name = "outer"
LiveObjects().add(self)

def __del__(self):
LiveObjects().remove(self)

def __repr__(self):
return self.__name

def ping(self):
return f"I am {self.__name}"

def createAutoproxyInstance(self):
i = Child(name="single instance")
self._pyroDaemon.register(i)
return i

def createAutoproxyGenerator(self):
ii = (Child(name=f"iterator instance {i}") for i in range(4))
for i in ii:
self._pyroDaemon.register(i)
yield i


class TestServerAutoproxyLifetime:
"""tests to verify that autoproxy server objects are released after use"""

def setup_method(self):
LiveObjects().clear()
self.daemon = Pyro5.server.Daemon(port=0)
uri = self.daemon.register(Parent, force=True)
self.objectUri = uri
self.daemonthread = DaemonLoopThread(self.daemon)
self.daemonthread.start()
self.daemonthread.running.wait()
time.sleep(0.05)

def teardown_method(self):
time.sleep(0.05)
if self.daemon is not None:
self.daemon.shutdown()
self.daemonthread.join()
assert not LiveObjects().get_alive_names(), "server autoproxy object(s) still alive"

def testSession(self):
with Pyro5.client.Proxy(self.objectUri) as p1, Pyro5.client.Proxy(self.objectUri) as p2:
assert "outer" in p1.ping()
assert "outer" in p2.ping()

def testAutoproxyInstance(self):
with Pyro5.client.Proxy(self.objectUri) as p:
inst = p.createAutoproxyInstance()
assert "single instance" in inst.ping()

def testAutoproxyGenerator(self):
with Pyro5.client.Proxy(self.objectUri) as p:
for inner in p.createAutoproxyGenerator():
assert "iterator instance" in inner.ping()


class TestServerOnce:
"""tests that are fine to run with just a single server type"""

Expand Down

0 comments on commit e167fb3

Please sign in to comment.