Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement reading and writing boolean values #555

Merged
merged 9 commits into from
Jun 25, 2023
18 changes: 17 additions & 1 deletion mcstatus/protocol/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from ctypes import c_uint32 as unsigned_int32
from ctypes import c_uint64 as unsigned_int64
from ipaddress import ip_address
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, cast

import asyncio_dgram

Expand Down Expand Up @@ -122,6 +122,10 @@ def write_ulong(self, value: int) -> None:
"""Write 8 bytes for value ``0 - 18446744073709551613 (2 ** 64 - 1)``."""
self.write(self._pack("Q", value))

def write_bool(self, value: bool) -> None:
"""Write 1 byte for boolean `True` or `False`"""
self.write(self._pack("?", value))

def write_buffer(self, buffer: "Connection") -> None:
"""Flush buffer, then write a varint of the length of the buffer's data, then write buffer data."""
data = buffer.flush()
Expand Down Expand Up @@ -214,6 +218,10 @@ async def write_ulong(self, value: int) -> None:
"""Write 8 bytes for value ``0 - 18446744073709551613 (2 ** 64 - 1)``."""
await self.write(self._pack("Q", value))

async def write_bool(self, value: bool) -> None:
"""Write 1 byte for boolean `True` or `False`"""
await self.write(self._pack("?", value))

async def write_buffer(self, buffer: "Connection") -> None:
"""Flush buffer, then write a varint of the length of the buffer's data, then write buffer data."""
data = buffer.flush()
Expand Down Expand Up @@ -302,6 +310,10 @@ def read_ulong(self) -> int:
"""Return ``0 - 18446744073709551613 (2 ** 64 - 1)``. Read 8 bytes."""
return self._unpack("Q", self.read(8))

def read_bool(self) -> bool:
"""Return `True` or `False`. Read 1 byte."""
return cast(bool, self._unpack("?", self.read(1)))

def read_buffer(self) -> "Connection":
"""Read a varint for length, then return a new connection from length read bytes."""
length = self.read_varint()
Expand Down Expand Up @@ -391,6 +403,10 @@ async def read_ulong(self) -> int:
"""Return ``0 - 18446744073709551613 (2 ** 64 - 1)``. Read 8 bytes."""
return self._unpack("Q", await self.read(8))

async def read_bool(self) -> bool:
"""Return `True` or `False`. Read 1 byte."""
return cast(bool, self._unpack("?", await self.read(1)))

async def read_buffer(self) -> Connection:
"""Read a varint for length, then return a new connection from length read bytes."""
length = await self.read_varint()
Expand Down
12 changes: 12 additions & 0 deletions tests/protocol/test_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,18 @@ def test_write_ulong_positive(self):

assert self.connection.flush() == bytearray.fromhex("8000000000000000")

@pytest.mark.parametrize("as_bytes,as_bool", [("01", True), ("00", False)])
def test_read_bool(self, as_bytes: str, as_bool: bool) -> None:
self.connection.receive(bytearray.fromhex(as_bytes))

assert self.connection.read_bool() is as_bool

@pytest.mark.parametrize("as_bytes,as_bool", [("01", True), ("00", False)])
def test_write_bool(self, as_bytes: str, as_bool: bool) -> None:
self.connection.write_bool(as_bool)

assert self.connection.flush() == bytearray.fromhex(as_bytes)

def test_read_buffer(self):
self.connection.receive(bytearray.fromhex("027FAA"))
buffer = self.connection.read_buffer()
Expand Down