snmp-subagent changes for bookworm upgrade (#313)

* remove loop argument from asyncio APIs

* Add test for code coverage

Taken from "commit 2dbbb5c"

---------

Co-authored-by: Qi Luo <qiluo-msft@users.noreply.github.com>

src/ax_interface/agent.py

@@ -15,9 +15,9 @@ def __init__(self, mib_cls, update_frequency, loop):
         self.loop = loop
 
         # synchronization events
-        self.run_enabled = asyncio.Event(loop=loop)
-        self.oid_updaters_enabled = asyncio.Event(loop=loop)
-        self.stopped = asyncio.Event(loop=loop)
+        self.run_enabled = asyncio.Event()
+        self.oid_updaters_enabled = asyncio.Event()
+        self.stopped = asyncio.Event()
 
         # Initialize our MIB
         self.mib_table = MIBTable(mib_cls, update_frequency)
@@ -46,7 +46,7 @@ async def run_in_event_loop(self):
             # signal background tasks to halt
             self.oid_updaters_enabled.clear()
             # wait for handlers to come back
-            await asyncio.wait_for(background_task, BACKGROUND_WAIT_TIMEOUT, loop=self.loop)
+            await asyncio.wait_for(background_task, BACKGROUND_WAIT_TIMEOUT)
 
         # signal that we're done!
         self.stopped.set()

src/ax_interface/mib.py

@@ -300,7 +300,7 @@ def start_background_tasks(self, event):
             fut = asyncio.ensure_future(updater.start())
             fut.add_done_callback(MIBTable._done_background_task_callback)
             tasks.append(fut)
-        return asyncio.gather(*tasks, loop=event._loop)
+        return asyncio.gather(*tasks)
 
     def _find_parent_prefix(self, item):
         oids = sorted(self.prefixes)

src/ax_interface/protocol.py

@@ -17,7 +17,7 @@ def __init__(self, mib_table, loop):
         self.loop = loop
         self.session_id = -1
         self.mib_table = mib_table
-        self.closed = asyncio.Event(loop=loop)
+        self.closed = asyncio.Event()
         self.counter = 0
 
     def send_pdu(self, pdu):

tests/test_agent.py

@@ -0,0 +1,27 @@
+import asyncio
+import os
+import sys
+import time
+from unittest import TestCase
+
+modules_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(modules_path, 'src'))
+
+import ax_interface
+
+class SonicMIB(metaclass=ax_interface.mib.MIBMeta):
+    """
+    Test
+    """
+
+class TestAgentLoop(TestCase):
+
+    async def delayed_shutdown(self, agent):
+        await asyncio.sleep(5)
+        await agent.shutdown()
+
+    def test_agent_loop(self):
+        event_loop = asyncio.get_event_loop()
+        agent = ax_interface.Agent(SonicMIB, 5, event_loop)
+        event_loop.create_task(self.delayed_shutdown(agent))
+        event_loop.run_until_complete(agent.run_in_event_loop())