464 lines
16 KiB
Python
464 lines
16 KiB
Python
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_sleep_attempt_aborts_if_night_was_skipped_while_approaching(self):
|
|
bed_id = next(iter(blocks.BED_IDS))
|
|
world = FakeWorld(block=bed_id)
|
|
state = SleepWithBedStates(Munch(
|
|
time=13000, dimension='overworld', world=world,
|
|
game=FakeGame(), path=[], look_at=None))
|
|
state.init()
|
|
state.area = (0, 64, 0)
|
|
state.state = state.use_bed
|
|
state.g.time = 25
|
|
|
|
state.use_bed()
|
|
|
|
self.assertEqual(state.state, state.cleanup)
|
|
|
|
def test_occupied_bed_is_abandoned_before_waiting_for_night(self):
|
|
bed_id = next(iter(blocks.BED_IDS))
|
|
|
|
class OccupiedWorld(FakeWorld):
|
|
@staticmethod
|
|
def check_bed_occupied(_position):
|
|
return True
|
|
|
|
state = SleepWithBedStates(Munch(
|
|
time=13000, dimension='overworld',
|
|
world=OccupiedWorld(block=bed_id), game=FakeGame(),
|
|
path=[], look_at=None))
|
|
state.init()
|
|
state.area = (0, 64, 0)
|
|
state.beds = [(2, 64, 0)]
|
|
state.state = state.use_bed
|
|
|
|
state.use_bed()
|
|
|
|
self.assertEqual(state.state, state.choose_bed)
|
|
|
|
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()
|