186 lines
6.6 KiB
Python
186 lines
6.6 KiB
Python
import io
|
|
import unittest
|
|
|
|
from munch import Munch
|
|
from minecraft.networking.modern import ClientTickEnd, ContainerOpenPacket
|
|
from minecraft.networking.packets import PacketBuffer
|
|
|
|
from mosfet.game import (
|
|
ContainerClickDropPacket, ContainerClickQuickMovePacket, Game,
|
|
)
|
|
from mosfet.protocol.types import Slot
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self):
|
|
self.packets = []
|
|
|
|
def write_packet(self, packet, force=False):
|
|
self.packets.append((packet, force))
|
|
|
|
|
|
class TestInventoryDrain(unittest.TestCase):
|
|
def test_drop_packet_encodes_throw_whole_stack(self):
|
|
packet = ContainerClickDropPacket(
|
|
container_id=0, state_id=7, slot=45)
|
|
payload = PacketBuffer()
|
|
packet.write_fields(payload)
|
|
|
|
self.assertEqual(packet.id, 0x12)
|
|
self.assertEqual(payload.get_writable(), bytes.fromhex(
|
|
'0007002d01040000'))
|
|
|
|
def test_server_held_slot_updates_drop_target(self):
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={38: Slot(True, 864, 1)}, holding_slot=0,
|
|
holding_hand=1, holding=355,
|
|
)
|
|
|
|
game.handle_set_held_slot(Munch(slot=2))
|
|
|
|
self.assertEqual(game.g.holding_slot, 2)
|
|
self.assertEqual(game.g.holding_hand, 0)
|
|
self.assertEqual(game.g.holding, 864)
|
|
|
|
def test_drop_packet_encodes_throw_one_item(self):
|
|
packet = ContainerClickDropPacket(
|
|
container_id=0, state_id=7, slot=45, button=0)
|
|
payload = PacketBuffer()
|
|
packet.write_fields(payload)
|
|
|
|
self.assertEqual(payload.get_writable(), bytes.fromhex(
|
|
'0007002d00040000'))
|
|
|
|
def test_quick_move_packet_encodes_shift_click(self):
|
|
packet = ContainerClickQuickMovePacket(
|
|
container_id=3, state_id=7, slot=42)
|
|
payload = PacketBuffer()
|
|
packet.write_fields(payload)
|
|
|
|
self.assertEqual(packet.id, 0x12)
|
|
self.assertEqual(payload.get_writable(), bytes.fromhex(
|
|
'0307002a00010000'))
|
|
|
|
def test_open_window_packet_decodes_container_metadata(self):
|
|
# window id, menu type, empty anonymous-NBT title
|
|
packet = ContainerOpenPacket()
|
|
packet.read(io.BytesIO(bytes.fromhex('010200')))
|
|
self.assertEqual(packet.window_id, 1)
|
|
self.assertEqual(packet.inventory_type, 2)
|
|
self.assertIsNone(packet.title)
|
|
|
|
def test_drop_stack_targets_currently_held_slot(self):
|
|
connection = FakeConnection()
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={38: Slot(True, 62, 3)},
|
|
inventory_state_ids={0: 7},
|
|
holding_hand=0, holding_slot=2,
|
|
connection=connection, draining=False, dumping=None,
|
|
drop_waiting=None,
|
|
)
|
|
|
|
self.assertTrue(game.drop_stack())
|
|
packet = connection.packets[0][0]
|
|
self.assertIsInstance(packet, ContainerClickDropPacket)
|
|
self.assertEqual((packet.container_id, packet.state_id, packet.slot),
|
|
(0, 7, 38))
|
|
self.assertIsInstance(connection.packets[1][0], ClientTickEnd)
|
|
self.assertEqual(game.g.drop_waiting['slot'], 38)
|
|
|
|
def test_drop_stack_uses_offhand_when_that_is_the_held_hand(self):
|
|
connection = FakeConnection()
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={45: Slot(True, 355, 1)},
|
|
inventory_state_ids={0: 2},
|
|
holding_hand=1, holding_slot=0,
|
|
connection=connection, draining=False, dumping=None,
|
|
drop_waiting=None,
|
|
)
|
|
|
|
self.assertTrue(game.drop_stack())
|
|
self.assertEqual(connection.packets[0][0].slot, 45)
|
|
|
|
def test_dump_drops_every_matching_stack_sequentially(self):
|
|
connection = FakeConnection()
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={9: Slot(True, 62, 5), 36: Slot(True, 62, 2),
|
|
37: Slot(True, 1, 1)},
|
|
inventory_state_ids={0: 3}, connection=connection,
|
|
draining=False, dumping=None, drop_waiting=None,
|
|
)
|
|
|
|
self.assertEqual(game.start_dump(62), 2)
|
|
game._tick_dump()
|
|
self.assertEqual(connection.packets[0][0].slot, 9)
|
|
|
|
game.g.inventory_state_ids[0] = 4
|
|
game.g.inv[9] = Slot(False)
|
|
game._tick_dump()
|
|
self.assertEqual(connection.packets[2][0].slot, 36)
|
|
self.assertEqual(game.g.dump_dropped, 1)
|
|
|
|
game.g.inventory_state_ids[0] = 5
|
|
game.g.inv[36] = Slot(False)
|
|
game._tick_dump()
|
|
self.assertIsNone(game.g.dumping)
|
|
self.assertEqual(game.g.dump_dropped, 2)
|
|
|
|
def test_drop_one_waits_for_decreased_stack_count(self):
|
|
connection = FakeConnection()
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={38: Slot(True, 62, 3)}, inventory_state_ids={0: 4},
|
|
holding_hand=0, holding_slot=2, connection=connection,
|
|
draining=False, dumping=None, drop_waiting=None, window=None,
|
|
)
|
|
|
|
self.assertTrue(game.drop_inventory_slot(38, whole_stack=False))
|
|
packet = connection.packets[0][0]
|
|
self.assertEqual((packet.slot, packet.button), (38, 0))
|
|
waiting = game.g.drop_waiting
|
|
self.assertEqual(waiting['expected_count'], 2)
|
|
game.g.inv[38] = Slot(True, 62, 2)
|
|
game._tick_single_drop()
|
|
self.assertIsNone(game.g.drop_waiting)
|
|
|
|
def test_dump_refuses_to_overlap_inventory_drain(self):
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(inv={9: Slot(True, 62, 1)}, draining=True,
|
|
dumping=None, drop_waiting=None)
|
|
self.assertIsNone(game.start_dump(62))
|
|
|
|
def test_drain_waits_for_authoritative_slot_update(self):
|
|
connection = FakeConnection()
|
|
game = Game.__new__(Game)
|
|
game.g = Munch(
|
|
inv={36: Slot(True, 62, 3)},
|
|
inventory_state_ids={0: 5},
|
|
connection=connection,
|
|
draining=False,
|
|
)
|
|
|
|
self.assertEqual(game.start_drain(), 1)
|
|
game._tick_drain()
|
|
self.assertTrue(game.g.draining)
|
|
self.assertEqual(game.g.drain_waiting['slot'], 36)
|
|
self.assertIsInstance(connection.packets[0][0],
|
|
ContainerClickDropPacket)
|
|
self.assertEqual(connection.packets[0][0].state_id, 5)
|
|
self.assertIsInstance(connection.packets[1][0], ClientTickEnd)
|
|
|
|
# InventoryManager updates the state id and slot from server packets.
|
|
game.g.inventory_state_ids[0] = 6
|
|
game.g.inv[36] = Slot(False)
|
|
game._tick_drain()
|
|
self.assertFalse(game.g.draining)
|
|
self.assertEqual(game.g.drain_dropped, 1)
|
|
self.assertIsNone(game.g.drain_error)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|