Add inventory controls and enhance dashboard

tests/test_inventory_commands.py

@@ -0,0 +1,51 @@
+import unittest
+
+from munch import Munch
+
+from mosfet.commands import Commands
+
+
+class FakeChat:
+    def __init__(self):
+        self.sent = []
+
+    def set_handler(self, handler):
+        self.handler = handler
+
+    def send(self, text):
+        self.sent.append(text)
+
+
+class FakeGame:
+    def __init__(self):
+        self.dump_result = 2
+        self.drop_result = True
+        self.dump_item = None
+
+    def drop_stack(self):
+        return self.drop_result
+
+    def start_dump(self, item_id):
+        self.dump_item = item_id
+        return self.dump_result
+
+
+class TestInventoryCommands(unittest.TestCase):
+    def setUp(self):
+        self.chat = FakeChat()
+        self.game = FakeGame()
+        self.commands = Commands(Munch(
+            name='mosfet1', chat=self.chat, game=self.game))
+
+    def test_drop_command_is_enabled_for_bot(self):
+        self.commands.handle_chat(('CHAT', 'tester', 'mosfet1: 1drop'))
+        self.assertEqual(self.chat.sent, ['ok'])
+
+    def test_dump_command_starts_dump_for_item_id(self):
+        self.commands.handle_chat(('CHAT', 'tester', 'mosfet1: 1dump 62'))
+        self.assertEqual(self.game.dump_item, 62)
+        self.assertEqual(self.chat.sent, ['ok'])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_inventory_drain.py

@@ -30,6 +30,28 @@ class TestInventoryDrain(unittest.TestCase):
         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)
@@ -48,6 +70,89 @@ class TestInventoryDrain(unittest.TestCase):
         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)

tests/test_web_inventory_api.py

@@ -0,0 +1,79 @@
+import importlib.util
+from pathlib import Path
+import threading
+import unittest
+
+from mosfet.protocol.types import Slot
+
+
+ENTRY = Path(__file__).resolve().parents[1] / 'mosfet.py'
+SPEC = importlib.util.spec_from_file_location('mosfet_web_entry', ENTRY)
+mosfet_web = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(mosfet_web)
+
+
+class FakeGame:
+    def __init__(self):
+        self.drop_calls = []
+        self.dump_item = None
+        self.dump_result = 2
+
+    def drop_inventory_slot(self, slot, whole_stack=True):
+        self.drop_calls.append((slot, whole_stack))
+        return True
+
+    def start_dump(self, item_id):
+        self.dump_item = item_id
+        return self.dump_result
+
+
+class TestWebInventoryAPI(unittest.TestCase):
+    def setUp(self):
+        self.g = mosfet_web.g
+        self.original = {
+            key: self.g.get(key) for key in ('connected', 'game', 'inv')
+        }
+        self.game = FakeGame()
+        self.g.connected = True
+        self.g.game = self.game
+        self.g.inv = {
+            9: Slot(True, 1145, 3,
+                    components={'custom_name': {'text': 'Test Paper'}})
+        }
+        self.g.reload_lock = threading.RLock()
+        self.client = mosfet_web.app.test_client()
+
+    def tearDown(self):
+        for key, value in self.original.items():
+            self.g[key] = value
+
+    def test_inventory_api_exposes_slot_components(self):
+        response = self.client.get('/api/global')
+        self.assertEqual(response.status_code, 200)
+        item = response.get_json()['inv']['9']
+        self.assertEqual(item['components']['custom_name']['text'], 'Test Paper')
+
+    def test_drop_endpoint_supports_one_stack_and_matching_actions(self):
+        one = self.client.post('/api/inventory/drop', json={
+            'slot': 9, 'action': 'one'})
+        stack = self.client.post('/api/inventory/drop', json={
+            'slot': 9, 'action': 'stack'})
+        matching = self.client.post('/api/inventory/drop', json={
+            'slot': 9, 'action': 'matching'})
+
+        self.assertEqual(one.status_code, 200)
+        self.assertEqual(stack.status_code, 200)
+        self.assertEqual(matching.status_code, 200)
+        self.assertEqual(self.game.drop_calls, [(9, False), (9, True)])
+        self.assertEqual(self.game.dump_item, 1145)
+        self.assertEqual(matching.get_json()['stacks'], 2)
+
+    def test_drop_endpoint_rejects_cross_origin_actions(self):
+        response = self.client.post('/api/inventory/drop', json={
+            'slot': 9, 'action': 'stack'},
+            headers={'Origin': 'https://elsewhere.example'})
+        self.assertEqual(response.status_code, 403)
+
+
+if __name__ == '__main__':
+    unittest.main()