From e167fb33754b9fc48c674d5bb460eff1bd62a6a3 Mon Sep 17 00:00:00 2001
From: Sven van Ashbrook <TheSven73@googlemail.com>
Date: Wed, 17 Jul 2024 18:31:39 +0000
Subject: [PATCH] Add tests to verify autoproxy server object lifetime

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.
---
 tests/test_server.py | 98 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 98 insertions(+)

diff --git a/tests/test_server.py b/tests/test_server.py
index da79784..7759f42 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -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
@@ -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"""