Port loiter and harden 26.3 bed updates

This commit is contained in:
Tanner
2026-11-05 01:05:52 +00:00
parent b9d43c7c37
commit efa2140a91
12 changed files with 664 additions and 65 deletions
+221 -6
View File
@@ -1,5 +1,6 @@
import io
import unittest
from contextlib import redirect_stdout
from munch import Munch
from minecraft.networking.packets import PacketBuffer
@@ -9,8 +10,9 @@ 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
from minecraft.networking.modern import ClientTickEnd, UseItemOnBlockPacket
class FakeConnection:
@@ -21,6 +23,14 @@ class FakeConnection:
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
@@ -28,16 +38,46 @@ class FakeWorld:
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):
self.placements.append((location, face, hand))
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_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)
@@ -78,6 +118,20 @@ class TestSleepWithBed(unittest.TestCase):
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)
@@ -91,6 +145,147 @@ class TestSleepWithBed(unittest.TestCase):
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_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()
@@ -106,20 +301,40 @@ class TestSleepWithBed(unittest.TestCase):
state.place_bed()
self.assertEqual(len(game.placements), 1)
location, face, hand = game.placements[0]
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_aborts_if_server_rejects_bed_placement(self):
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.wait_time = 0
state.bed_place_deadline = 0
state.wait_use()