mirror of
https://github.com/Dinnerbone/mcstatus.git
synced 2026-04-05 19:41:24 +08:00
Fix compatibility with python3.7 and a few linter things
This commit is contained in:
@@ -1,5 +1,13 @@
|
||||
from abc import abstractmethod, ABC
|
||||
from typing import SupportsBytes, Iterable, SupportsIndex, Union
|
||||
|
||||
try:
|
||||
from typing import SupportsBytes, Iterable, SupportsIndex, Tuple, Union
|
||||
|
||||
BytesConvertable = Union[SupportsIndex, Iterable[SupportsIndex]]
|
||||
except ImportError:
|
||||
from typing import SupportsBytes, Iterable, Tuple, Union
|
||||
|
||||
BytesConvertable = Union[int, Iterable[int]]
|
||||
import socket
|
||||
import struct
|
||||
import asyncio
|
||||
@@ -10,8 +18,6 @@ from ctypes import c_int32 as signed_int32
|
||||
|
||||
from ..scripts.address_tools import ip_type
|
||||
|
||||
BytesConvertable = Union[SupportsIndex, Iterable[SupportsIndex]]
|
||||
|
||||
|
||||
class Connection:
|
||||
def __init__(self):
|
||||
@@ -185,7 +191,7 @@ class AsyncReadConnection(Connection, ABC):
|
||||
|
||||
|
||||
class TCPSocketConnection(Connection):
|
||||
def __init__(self, addr: tuple[str, int], timeout: float = 3):
|
||||
def __init__(self, addr: Tuple[str, int], timeout: float = 3):
|
||||
Connection.__init__(self)
|
||||
self.socket = socket.create_connection(addr, timeout=timeout)
|
||||
self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
||||
@@ -219,7 +225,7 @@ class TCPSocketConnection(Connection):
|
||||
|
||||
|
||||
class UDPSocketConnection(Connection):
|
||||
def __init__(self, addr: tuple[str, int], timeout: float = 3):
|
||||
def __init__(self, addr: Tuple[str, int], timeout: float = 3):
|
||||
Connection.__init__(self)
|
||||
self.addr = addr
|
||||
self.socket = socket.socket(
|
||||
@@ -264,7 +270,7 @@ class TCPAsyncSocketConnection(AsyncReadConnection):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
async def connect(self, addr: tuple[str, int], timeout: float = 3):
|
||||
async def connect(self, addr: Tuple[str, int], timeout: float = 3):
|
||||
self.timeout = timeout
|
||||
conn = asyncio.open_connection(addr[0], addr[1])
|
||||
self.reader, self.writer = await asyncio.wait_for(conn, timeout=self.timeout)
|
||||
@@ -296,7 +302,7 @@ class UDPAsyncSocketConnection(AsyncReadConnection):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
async def connect(self, addr: tuple[str, int], timeout: float = 3):
|
||||
async def connect(self, addr: Tuple[str, int], timeout: float = 3):
|
||||
self.timeout = timeout
|
||||
conn = asyncio_dgram.connect((addr[0], addr[1]))
|
||||
self.stream = await asyncio.wait_for(conn, timeout=self.timeout)
|
||||
|
||||
@@ -276,5 +276,5 @@ class UDPSocketConnectionTest:
|
||||
|
||||
self.connection.socket.sendto.assert_called_once_with( # type: ignore[attr-defined]
|
||||
bytearray.fromhex("7FAA"),
|
||||
("localhost", 1234)
|
||||
("localhost", 1234),
|
||||
)
|
||||
|
||||
@@ -11,7 +11,7 @@ def test_is_completely_asynchronous():
|
||||
assertions = 0
|
||||
for attribute in dir(conn):
|
||||
if attribute.startswith("read_"):
|
||||
assert iscoroutinefunction(conn.__getattribute__(attribute))
|
||||
assert iscoroutinefunction(getattr(conn, attribute))
|
||||
assertions += 1
|
||||
assert assertions > 0, "None of the read_* attributes were async"
|
||||
|
||||
@@ -21,6 +21,6 @@ def test_query_is_completely_asynchronous():
|
||||
assertions = 0
|
||||
for attribute in dir(conn):
|
||||
if attribute.startswith("read_"):
|
||||
assert iscoroutinefunction(conn.__getattribute__(attribute))
|
||||
assert iscoroutinefunction(getattr(conn, attribute))
|
||||
assertions += 1
|
||||
assert assertions > 0, "None of the read_* attributes were async"
|
||||
|
||||
Reference in New Issue
Block a user