sync local changes

Author: VigneshVSV

hololinked/core/thing.py

@@ -359,10 +359,9 @@ def exit(self) -> None:
             self.logger.debug("exit() called on a object that is not exposed yet.")
             return 
         if self._owners:
-            raise BreakInnerLoop # stops the inner loop of the object
-        else:
             raise NotImplementedError("call exit on the top-level object, composed objects cannot exit the loop. "+
                                 f"This object belongs to {self._owners.__class__.__name__} with ID {self._owners.id}.")
+        self.rpc_server.stop()
 
 
     @action()

hololinked/server/http/__init__.py

@@ -13,7 +13,7 @@
 from ...param import Parameterized
 from ...param.parameters import Integer, IPAddress, ClassSelector, Selector, TypedList, String
 from ...constants import HTTP_METHODS, ZMQ_TRANSPORTS, HTTPServerTypes, Operations
-from ...utils import complete_pending_tasks_in_current_loop, complete_pending_tasks_in_current_loop_async, forkable, get_IP_from_interface, get_current_async_loop, issubklass, pep8_to_dashed_name, get_default_logger, run_callable_somehow
+from ...utils import complete_pending_tasks_in_current_loop, complete_pending_tasks_in_current_loop_async, forkable, get_IP_from_interface, get_current_async_loop, issubklass, pep8_to_dashed_name, get_default_logger, print_pending_tasks_in_current_loop, run_callable_somehow
 from ...serializers.serializers import JSONSerializer
 from ...schema_validators import BaseSchemaValidator, JSONSchemaValidator
 from ...core.property import Property
@@ -217,7 +217,8 @@ def listen(self, forked: bool = False) -> None:
         self.tornado_instance.listen(port=self.port, address=self.address)    
         self.logger.info(f'started webserver at {self._IP}, ready to receive requests.')
         self.tornado_event_loop.start()
-        complete_pending_tasks_in_current_loop() # will reach here only when the server is stopped, so complete pending tasks
+        if forked:
+            complete_pending_tasks_in_current_loop() # will reach here only when the server is stopped, so complete pending tasks
 
 
     def stop(self, attempt_async_stop: bool = True) -> None:
@@ -236,7 +237,8 @@ def stop(self, attempt_async_stop: bool = True) -> None:
         run_callable_somehow(self.tornado_instance.close_all_connections())
         if self.tornado_event_loop is not None:
             self.tornado_event_loop.stop()
-        # complete_pending_tasks_in_current_loop()
+        complete_pending_tasks_in_current_loop()
+        print_pending_tasks_in_current_loop()
 
 
     async def async_stop(self) -> None:
@@ -251,6 +253,7 @@ async def async_stop(self) -> None:
         await self.tornado_instance.close_all_connections()
         if self.tornado_event_loop is not None:
             self.tornado_event_loop.stop()
+        print_pending_tasks_in_current_loop()
         # await complete_pending_tasks_in_current_loop_async()
 
 

hololinked/utils.py

@@ -155,6 +155,17 @@ async def complete_pending_tasks_in_current_loop_async():
     """
     await asyncio.gather(*asyncio.all_tasks(get_current_async_loop()))
 
+def print_pending_tasks_in_current_loop():
+    """
+    Print all pending tasks in the current asyncio event loop.
+    """
+    tasks = asyncio.all_tasks(get_current_async_loop())
+    if not tasks:
+        print("No pending tasks in the current event loop.")
+        return
+    for task in tasks:
+        print(f"Task: {task}, Status: {task._state}")
+
 
 def get_signature(callable : typing.Callable) -> typing.Tuple[typing.List[str], typing.List[type]]: 
     """

tests/test_13_protocols_http.py

@@ -4,7 +4,7 @@
 from hololinked.core.meta import ThingMeta
 from hololinked.server.http import HTTPServer
 from hololinked.core.zmq.rpc_server import RPCServer # sets loop policy, TODO: move somewhere else
-from hololinked.utils import get_current_async_loop, issubklass
+from hololinked.utils import get_current_async_loop, issubklass, print_pending_tasks_in_current_loop
 
 try:
     from .things import OceanOpticsSpectrometer, TestThing
@@ -18,26 +18,28 @@
 class TestHTTPServer(TestCase):
 
 
-    def notest_1_init_run_and_stop(self):
+    def test_1_init_run_and_stop(self):
         """Test basic init, run and stop of the HTTP server."""
 
         # init, run and stop synchronously
         server = HTTPServer(log_level=logging.ERROR+10)
-        # self.assertTrue(server.all_ok)
-        # server.listen(forked=True)
-        # time.sleep(3)
-        # server.stop()
+        self.assertTrue(server.all_ok)
+        server.listen(forked=True)
+        time.sleep(3)
+        server.stop()
+        time.sleep(2)
 
         # init, run and stop asynchronously
         server.listen(forked=True)
         time.sleep(3)
         get_current_async_loop().run_until_complete(server.async_stop())
+        time.sleep(2)
 
         server.listen(forked=True)
         time.sleep(3)
         response = requests.post('http://localhost:8080/stop')
         self.assertIn(response.status_code, [200, 201, 202, 204])
-        
+        time.sleep(2)
 
     def notest_2_add_interaction_affordance(self):
         """Test adding an interaction affordance to the HTTP server."""
@@ -168,8 +170,9 @@ def test_5_http_handler_functionalities(self):
                     8085
                 ]):
             thing.run_with_http_server(forked=True, port=port)
-            time.sleep(3)
+            time.sleep(3) # TODO: add a way to check if the server is running
 
+        session = requests.Session()
         for (method, path, body) in [
                     ('get',  '/test-spectrometer/max-intensity', 16384),
                     ('get',  '/test-spectrometer/serial-number', 'simulation'),
@@ -178,12 +181,22 @@ def test_5_http_handler_functionalities(self):
                     ('post', '/test-spectrometer/connect', None)
                 ]:
             if method == 'get':
-                response = requests.get(f'http://localhost:{port}{path}')
+                response = session.get(f'http://localhost:{port}{path}')
             elif method == 'post':
-                response = requests.post(f'http://localhost:{port}{path}')
+                response = session.post(f'http://localhost:{port}{path}')
             self.assertTrue(response.status_code in [200, 201, 202, 204])
             if body:
                 self.assertTrue(response.json() == body)
+
+
+        for (method, path, body) in [
+                ('post', '/test-spectrometer/exit', None),
+                ('post', '/stop', None)
+            ]:
+            if method == 'get':
+                response = session.get(f'http://localhost:{port}{path}')
+            elif method == 'post':
+                response = session.post(f'http://localhost:{port}{path}')
 
 
     def notest_5_run_thing_with_http_server(self):