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()