import io import unittest from contextlib import redirect_stdout from unittest.mock import patch from munch import Munch from minecraft.networking.packets import PacketBuffer from minecraft.networking.types import Float, Long, VarInt, VarLong from mosfet.game import Game from mosfet.info import blocks, items from mosfet.protocol.types import Slot from mosfet.jobs.sleep_with_bed import SleepWithBedStates from mosfet.world import World from mosfet.protocol.modern import PlayerCommandPacket, UpdateTimePacket from minecraft.networking.modern import ClientTickEnd, UseItemOnBlockPacket class FakeConnection: def __init__(self): self.packets = [] def write_packet(self, packet, force=False): self.packets.append((packet, force)) class FakeChunks: def __init__(self, block): self.block = block def get_block_at(self, *position): return self.block class FakeWorld: def __init__(self, block=blocks.AIR): self.block = block def block_at(self, *position): return self.block def check_bed_occupied(self, position): return False class FakeGame: def __init__(self): self.placements = [] def place_block(self, location, face, hand=0, cursor_y=None): self.placements.append((location, face, hand, cursor_y)) class TestSleepWithBed(unittest.TestCase): def test_confirmed_sleep_records_bed_for_following_cache_pass(self): bed = (1, 64, 2) g = Munch(world=FakeWorld(), path=[], look_at=None) state = SleepWithBedStates(g) state.area = bed state.sleep_confirmed = True state.cleanup() self.assertEqual(g.post_sleep_cache_bed, bed) def test_unconfirmed_sleep_does_not_trigger_post_sleep_cache(self): g = Munch(world=FakeWorld(), path=[], look_at=None) state = SleepWithBedStates(g) state.area = (1, 64, 2) state.sleep_confirmed = False state.cleanup() self.assertNotIn('post_sleep_cache_bed', g) def test_bed_occupancy_only_logs_first_observation_and_changes(self): bed_states = [ state for state in blocks.BED_IDS if blocks.BLOCKS[state] == 'white_bed' and blocks.PROPS[state].get('part') == 'foot' and blocks.PROPS[state].get('facing') == 'north' ] free = next(state for state in bed_states if blocks.PROPS[state].get('occupied') == 'false') occupied = next(state for state in bed_states if blocks.PROPS[state].get('occupied') == 'true') chunks = FakeChunks(free) world = World(Munch(chunks=chunks)) output = io.StringIO() with redirect_stdout(output): self.assertFalse(world.check_bed_occupied((1, 64, 2))) self.assertFalse(world.check_bed_occupied((1, 64, 2))) chunks.block = occupied self.assertTrue(world.check_bed_occupied((1, 64, 2))) self.assertTrue(world.check_bed_occupied((1, 64, 2))) lines = output.getvalue().splitlines() self.assertEqual(len(lines), 2) self.assertIn('-> free', lines[0]) self.assertIn('-> occupied', lines[1]) def test_update_time_packet_decodes_clock_updates(self): payload = PacketBuffer() Long.send(123456, payload) VarInt.send(1, payload) VarInt.send(0, payload) VarLong.send(13000, payload) Float.send(0.5, payload) Float.send(1.0, payload) packet = UpdateTimePacket() packet.read(io.BytesIO(payload.get_writable())) self.assertEqual(packet.id, 0x73) self.assertEqual(packet.age, 123456) self.assertEqual(packet.clock_updates, [{ 'clock_id': 0, 'total_ticks': 13000, 'partial_tick': 0.5, 'rate': 1.0, }]) def test_game_advances_world_clock_between_server_updates(self): game = Game.__new__(Game) game.g = Munch( time_base=None, time_update_received_at=None, clock_rate=1.0, time=0.0, connection=Munch(registry_ids={'minecraft:world_clock': { 'minecraft:the_end': 1, 'minecraft:overworld': 0, }}), dimension='overworld') packet = Munch(age=900000, clock_updates=[ {'clock_id': 1, 'total_ticks': 6000, 'partial_tick': 0.0, 'rate': 1.0}, {'clock_id': 0, 'total_ticks': 13000, 'partial_tick': 0.5, 'rate': 1.0}, ]) with patch('mosfet.game.time.monotonic', return_value=100.0): game.handle_time_update(packet) self.assertEqual(game.g.time, 13000.5) self.assertEqual(game.g.clock_name, 'minecraft:overworld') with patch('mosfet.game.time.monotonic', return_value=100.5): game._advance_world_time() self.assertEqual(game.g.time, 13010.5) def test_player_command_encodes_stop_sleeping(self): packet = PlayerCommandPacket( entity_id=7, action_id=0, jump_boost=0) payload = PacketBuffer() packet.write_fields(payload) self.assertEqual(packet.id, 0x2A) self.assertEqual(payload.get_writable(), b'\x07\x00\x00') def test_game_leave_bed_sends_protocol_777_stop_sleeping(self): connection = FakeConnection() game = Game.__new__(Game) game.g = Munch(connection=connection, eid=42) game.leave_bed() packet, force = connection.packets[0] self.assertIsInstance(packet, PlayerCommandPacket) self.assertEqual((packet.entity_id, packet.action_id, packet.jump_boost), (42, 0, 0)) self.assertTrue(force) self.assertIsInstance(connection.packets[1][0], ClientTickEnd) def test_existing_bed_use_aims_inside_the_short_bed_collision_shape(self): connection = FakeConnection() game = Game.__new__(Game) game.g = Munch(connection=connection) game.sequence = 0 game.place_block((-175, 70, -312), 1, cursor_y=0.5) packet = connection.packets[0][0] self.assertIsInstance(packet, UseItemOnBlockPacket) self.assertEqual(packet.location, (-175, 70, -312)) self.assertEqual(packet.cursor_y, 0.5) self.assertIsInstance(connection.packets[1][0], ClientTickEnd) def test_bed_in_offhand_is_selectable_without_swapping_inventory(self): connection = FakeConnection() game = Game.__new__(Game) bed_id = next(iter(items.BED_IDS)) game.g = Munch( connection=connection, inv={45: Slot(True, bed_id, 1)}, holding_hand=0) self.assertTrue(game.select_item(items.BED_IDS)) self.assertEqual(game.g.holding, bed_id) self.assertEqual(game.g.holding_hand, 1) self.assertIsInstance(connection.packets[0][0], ClientTickEnd) def test_bed_head_position_is_normalized_to_foot(self): bed_name = 'white_bed' foot_state = next( state for state in blocks.BED_IDS if blocks.BLOCKS[state] == bed_name and blocks.PROPS[state].get('part') == 'foot' and blocks.PROPS[state].get('facing') == 'west') head_state = next( state for state in blocks.BED_IDS if blocks.BLOCKS[state] == bed_name and blocks.PROPS[state].get('part') == 'head' and blocks.PROPS[state].get('facing') == 'west') foot, head = (-175, 70, -312), (-176, 70, -312) class BedWorld: def block_at(self, *position): return {foot: foot_state, head: head_state}.get(tuple(position), blocks.AIR) state = SleepWithBedStates(Munch(world=BedWorld())) self.assertEqual(state._bed_foot(head), foot) self.assertEqual(state._bed_foot(foot), foot) def test_bed_search_falls_back_to_loaded_blocks_if_index_is_empty(self): bed = (1, 64, 0) bed_state = next( state for state in blocks.BED_IDS if blocks.PROPS[state].get('part') == 'foot') class IndexedWorld: def find_blocks_indexed(self, center, block_ids, distance=0): return [] def find_blocks_3d(self, center, block_ids, distance, y_limit): self.assert_search = (center, block_ids, distance, y_limit) return iter([bed]) def block_at(self, *position): return bed_state if tuple(position) == bed else blocks.AIR def check_bed_occupied(self, position): return False world = IndexedWorld() state = SleepWithBedStates(Munch( world=world, pos=(0, 64, 0), time=13000, dimension='overworld')) state.find_beds() self.assertEqual(state.beds, [bed]) self.assertEqual(state.state, state.choose_bed) def test_force_new_bed_skips_existing_bed_scan(self): class IndexedWorld: def find_blocks_indexed(self, *args, **kwargs): raise AssertionError('forced placement must skip bed search') def find_blocks_3d(self, *args, **kwargs): raise AssertionError('forced placement must skip local scan') state = SleepWithBedStates(Munch( world=IndexedWorld(), pos=(0, 64, 0), time=13000, dimension='overworld')) state.force_new_bed = True state.find_beds() self.assertEqual(state.beds, []) self.assertEqual(state.state, state.choose_bed) def test_bed_approach_follows_path_before_aiming_at_bed(self): bed = (4, 64, 2) opening = (3, 64, 0) state = SleepWithBedStates(Munch( pos=(0, 64, 0), path=[opening], look_at=bed, game=FakeGame(), world=FakeWorld())) state.area = bed state.opening = opening state.opening_deadline = 9999999999 state.going_to_bed() self.assertIsNone(state.g.look_at) state.g.pos = opening state.going_to_bed() self.assertEqual(state.g.look_at, bed) self.assertEqual(state.state, state.turn_to_bed) def test_bed_placement_approach_follows_path(self): opening = (3, 64, 0) state = SleepWithBedStates(Munch( pos=(0, 64, 0), path=[opening], look_at=(4, 64, 2), game=FakeGame(), world=FakeWorld())) state.area = (4, 64, 2) state.opening = opening state.opening_deadline = 9999999999 state.going_to_area() self.assertIsNone(state.g.look_at) def test_arrival_accepts_completed_path_with_small_position_error(self): state = SleepWithBedStates(Munch( pos=(0, 64, 0), path=[(1, 64, 0)], look_at=None, game=FakeGame(), world=FakeWorld())) state.opening = (1, 64, 0) state.opening_deadline = 9999999999 state.going_to_bed() self.assertEqual(state.state, state.turn_to_bed) self.assertEqual(state.g.path, []) def test_unconfirmed_sleep_retries_using_head_block(self): foot = (0, 64, 0) head = (0, 64, -1) bed_id = next( state for state in blocks.BED_IDS if blocks.PROPS[state].get('part') == 'foot') class ThreatWorld(FakeWorld): def find_threats(self, center, distance): return [] game = FakeGame() state = SleepWithBedStates(Munch( time=13000, pos=(1, 64, 0), world=ThreatWorld(block=bed_id), game=game, path=[], look_at=None, correction_count=0)) state.area = foot state.bed_head = head state.sleep_use_attempts = 1 state.sleep_retry_at = 0 state.sleep_deadline = 9999999999 state.sleep_bed() self.assertEqual(game.placements, [(head, 1, 0, 0.5)]) self.assertEqual(state.sleep_use_attempts, 2) def test_failed_existing_bed_use_is_blacklisted(self): bed_id = next(iter(blocks.BED_IDS)) class ThreatWorld(FakeWorld): def find_threats(self, center, distance): return [] state = SleepWithBedStates(Munch( time=13000, pos=(1, 64, 0), world=ThreatWorld(block=bed_id), game=FakeGame(), path=[], look_at=None, correction_count=0)) state.area = (0, 64, 0) state.sleep_use_attempts = 2 state.sleep_retry_at = 9999999999 state.sleep_deadline = 0 state.sleep_bed() self.assertIn(state.area, state.bad_beds) self.assertEqual(state.state, state.cleanup) def test_bed_waits_with_status_until_sleep_time(self): bed_id = next(iter(blocks.BED_IDS)) state = SleepWithBedStates(Munch( time=12400, world=FakeWorld(block=bed_id), game=FakeGame(), path=[], look_at=None)) state.area = (0, 64, 0) state.state = state.use_bed state.use_bed() self.assertEqual(state.state, state.use_bed) self.assertTrue(state.waiting_for_night) def test_daytime_sleep_aborts_cleanly(self): state = SleepWithBedStates(Munch(time=1000, dimension='overworld')) state.init() self.assertEqual(state.state, state.cleanup) def test_placing_bed_uses_support_block_face_top(self): game = FakeGame() state = SleepWithBedStates(Munch( game=game, world=FakeWorld(), path=[], look_at=None, holding_hand=1)) state.area = (4, 70, 5) state.place_bed() self.assertEqual(len(game.placements), 1) location, face, hand, cursor_y = game.placements[0] self.assertEqual(location, (4, 69, 5)) self.assertEqual(int(face), 1) self.assertEqual(hand, 1) self.assertIsNone(cursor_y) self.assertTrue(state.my_bed) self.assertEqual(state.state, state.wait_use) def test_wait_use_waits_for_delayed_chunk_update(self): bed_id = next(iter(blocks.BED_IDS)) world = FakeWorld(block=blocks.AIR) state = SleepWithBedStates(Munch( game=FakeGame(), world=world, path=[], look_at=None)) state.area = (4, 70, 5) state.my_bed = True state.bed_place_deadline = 9999999999 state.state = state.wait_use state.wait_use() self.assertEqual(state.state, state.wait_use) self.assertTrue(state.my_bed) world.block = bed_id state.wait_use() self.assertEqual(state.state, state.use_bed) self.assertIsNotNone(state.bed_head) def test_wait_use_aborts_after_update_timeout(self): state = SleepWithBedStates(Munch( game=FakeGame(), world=FakeWorld(block=blocks.AIR), path=[], look_at=None)) state.area = (4, 70, 5) state.my_bed = True state.bed_place_deadline = 0 state.wait_use() self.assertIn(state.area, state.bad_areas) self.assertFalse(state.my_bed) self.assertEqual(state.state, state.cleanup) if __name__ == '__main__': unittest.main()