Simplify check_port by using asyncio.get_running_loop()

Author: agners

supervisor/addons/addon.py

@@ -539,7 +539,7 @@ async def watchdog_application(self) -> bool:
 
         # TCP monitoring
         if s_prefix == "tcp":
-            return await check_port(self.coresys.loop, self.ip_address, port)
+            return await check_port(self.ip_address, port)
 
         # lookup the correct protocol from config
         if t_proto:

supervisor/homeassistant/api.py

@@ -141,7 +141,6 @@ async def get_api_state(self) -> str | None:
 
         # Check if port is up
         if not await check_port(
-            self.coresys.loop,
             self.sys_homeassistant.ip_address,
             self.sys_homeassistant.api_port,
         ):

supervisor/ingress.py

@@ -163,9 +163,7 @@ async def get_dynamic_port(self, addon_slug: str) -> int:
         while (
             port is None
             or port in self.ports.values()
-            or await check_port(
-                self.coresys.loop, self.sys_docker.network.gateway, port
-            )
+            or await check_port(self.sys_docker.network.gateway, port)
         ):
             port = random.randint(62000, 65500)
 

supervisor/utils/__init__.py

@@ -39,15 +39,13 @@ async def wrap_api(api, *args, **kwargs):
     return wrap_api
 
 
-async def check_port(
-    loop: asyncio.AbstractEventLoop, address: IPv4Address, port: int
-) -> bool:
+async def check_port(address: IPv4Address, port: int) -> bool:
     """Check if port is mapped."""
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     sock.setblocking(False)
     try:
         async with asyncio.timeout(0.5):
-            await loop.sock_connect(sock, (str(address), port))
+            await asyncio.get_running_loop().sock_connect(sock, (str(address), port))
     except (OSError, TimeoutError):
         return False
     finally:

tests/utils/test_check_port.py

@@ -7,9 +7,9 @@
 
 async def test_exists_open_port(coresys: CoreSys):
     """Test a exists network port."""
-    assert await check_port(coresys.loop, ip_address("8.8.8.8"), 53)
+    assert await check_port(ip_address("8.8.8.8"), 53)
 
 
 async def test_not_exists_port(coresys: CoreSys):
     """Test a not exists network service."""
-    assert not await check_port(coresys.loop, ip_address("192.0.2.1"), 53)
+    assert not await check_port(ip_address("192.0.2.1"), 53)