Embedded module mt_switchbot: a paced request queue (0.3 s gap, one in flight, 4 s timeout -> unsupported), atomic attrs lists from the server, two alternatives of up to 5 targets, 10-1000 roll limits, live state and stop reasons. Entry button under the wiki button by the minimap. The selection follows the server-reported cell when the item moves mid-run and falls back to the same vnum when it moves while idle. Live tested against 203 (channel-5 test core, port 13040) with the macOS native client. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
155 lines
9.4 KiB
Python
155 lines
9.4 KiB
Python
import re
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT=Path(__file__).resolve().parents[2]
|
|
|
|
def script(name):return (ROOT/'src/platform/ScriptLib'/name).read_text().split('R"PY(',1)[1].rsplit(')PY";',1)[0]
|
|
|
|
|
|
class Widget(object):
|
|
def __init__(self,*a,**k):self.shown=False;self.text='';self.pos=(0,0);self.size=(0,0)
|
|
def __getattr__(self,name):return lambda *a,**k:None
|
|
def Show(self):self.shown=True
|
|
def Hide(self):self.shown=False
|
|
def IsShow(self):return self.shown
|
|
def SetText(self,text):self.text=text
|
|
def SetPosition(self,x,y):self.pos=(x,y)
|
|
def GetLocalPosition(self):return self.pos
|
|
def GetGlobalPosition(self):return self.pos
|
|
def SetSize(self,w,h):self.size=(w,h)
|
|
def GetWidth(self):return self.size[0]
|
|
def GetHeight(self):return self.size[1]
|
|
|
|
|
|
class Action(Widget):
|
|
def __init__(self,parent,x,y,width,text,event):
|
|
Widget.__init__(self);self.pos=(x,y);self.event=event;self.enabled=True;self.label=Widget();self.text=text;self.shown=True
|
|
def SetEnabled(self,enabled):self.enabled=enabled
|
|
|
|
|
|
class SwitchbotTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.time=1.;self.sent=[];self.chat=[];self.events=[]
|
|
self.inventory={10:(11209,1,[(3,2),(15,1)]),20:(71084,7,[]),21:(71084,3,[]),30:(27001,5,[])}
|
|
protos={11209:(2,0),71084:(3,18),27001:(3,0)}
|
|
mods={n:types.ModuleType(n) for n in ('app','chat','game','grp','item','net','player','ui','uiToolTip','wndMgr','mt_biolog','mt_wiki')}
|
|
mods['app'].GetTime=lambda:self.time
|
|
mods['chat'].CHAT_TYPE_INFO=1;mods['chat'].AppendChat=lambda t,m:self.chat.append(m)
|
|
mods['net'].SendChatPacket=self.sent.append
|
|
mods['game'].GameWindow=type('GameWindow',(),dict(Open=lambda s:None,Close=lambda s:None,BINARY_ServerCommand_Run=lambda s,line:7))
|
|
grp=mods['grp'];grp.SetColor=grp.RenderBar=grp.RenderBox=lambda *a:None
|
|
it=mods['item'];it.ITEM_TYPE_WEAPON=1;it.ITEM_TYPE_ARMOR=2;it.ITEM_TYPE_USE=3;self.selected=None
|
|
it.SelectItem=lambda v:setattr(self,'selected',v);it.GetItemType=lambda:protos[self.selected][0];it.GetItemSubType=lambda:protos[self.selected][1]
|
|
it.GetItemName=lambda:'N%d'%self.selected
|
|
pl=mods['player'];pl.INVENTORY=1;pl.INVENTORY_PAGE_SIZE=45;pl.INVENTORY_PAGE_COUNT=5
|
|
pl.GetItemIndex=lambda w,c:self.inventory.get(c,(0,))[0];pl.GetItemCount=lambda w,c:self.inventory[c][1]
|
|
pl.GetItemAttribute=lambda w,c,i:(self.inventory[c][2]+[(0,0)]*5)[i]
|
|
mods['ui'].Window=Widget
|
|
mods['uiToolTip'].ItemToolTip=type('ItemToolTip',(),dict(AFFECT_DICT={3:lambda v:'STR+%d'%v}))
|
|
mods['wndMgr'].GetScreenWidth=lambda:800;mods['wndMgr'].GetScreenHeight=lambda:600
|
|
mods['mt_biolog'].Action=Action;mods['mt_biolog'].Text=lambda p,x,y,t='',c=0:Widget();mods['mt_biolog'].L=lambda t:t if isinstance(t,str) else t.decode('utf-8')
|
|
mods['mt_wiki'].U=lambda t:t if isinstance(t,str) else t.decode('utf-8');mods['mt_wiki'].B=lambda t,limit=0:t
|
|
self.previous={n:sys.modules.get(n) for n in mods}
|
|
sys.modules.update(mods)
|
|
self.ns={'__name__':'mt_switchbot'};exec(script('SwitchbotScript.inc'),self.ns)
|
|
self.bot=self.ns['Bot'](self.sent.append,lambda:self.time,self.events.append)
|
|
|
|
def tearDown(self):
|
|
for name,value in self.previous.items():
|
|
if value is None:sys.modules.pop(name,None)
|
|
else:sys.modules[name]=value
|
|
|
|
def advance(self,seconds=.4):self.time+=seconds;self.bot.Tick()
|
|
|
|
def load(self,bot,cell=10):
|
|
for line in ('MTSB1 attrs %d 11209 3'%cell,'MTSB1 attr 0 3 12','MTSB1 attr 1 63 6','MTSB1 attr 2 71 30','MTSB1 attrs_end %d 3'%cell):bot.Receive(line)
|
|
|
|
def test_commands_are_paced_and_attrs_requests_collapse(self):
|
|
self.bot.RequestAttrs(10);self.bot.RequestAttrs(11);self.bot.RequestAttrs(12);self.bot.Status()
|
|
self.assertEqual(self.sent,['/mt_switchbot 1 attrs 10'])
|
|
self.load(self.bot);self.advance()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 attrs 12')
|
|
self.bot.Receive('MTSB1 error attrs no_attr');self.advance()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 status')
|
|
self.assertEqual(len(self.sent),3)
|
|
|
|
def test_atomic_attr_list_and_rejected_partials(self):
|
|
self.load(self.bot)
|
|
self.assertEqual(self.bot.allowed[10]['rows'],[(3,12),(63,6),(71,30)])
|
|
for bad in (['MTSB1 attrs 11 1 2','MTSB1 attr 0 3 12','MTSB1 attrs_end 11 2'],
|
|
['MTSB1 attrs 11 1 1','MTSB1 attr 1 3 12','MTSB1 attrs_end 11 1'],
|
|
['MTSB1 attrs 11 1 1','MTSB1 attr 0 3 -1','MTSB1 attrs_end 11 1'],
|
|
['MTSB1 attrs_end 11 0']):
|
|
for line in bad:self.assertTrue(self.bot.Receive(line))
|
|
self.assertNotIn(11,self.bot.allowed)
|
|
self.assertFalse(self.bot.Receive('MTF1 cap 1 1 127 0'))
|
|
|
|
def test_start_validates_against_server_list_and_builds_command(self):
|
|
self.load(self.bot)
|
|
self.assertEqual(self.bot.Start(10,71084,100,[[(3,13)],[]]),'target')
|
|
self.assertEqual(self.bot.Start(10,71084,100,[[(1,1)],[]]),'target')
|
|
self.assertEqual(self.bot.Start(10,71084,100,[[(3,1),(3,2)],[]]),'invalid')
|
|
self.assertEqual(self.bot.Start(10,71084,1001,[[(3,1)],[]]),'invalid')
|
|
self.assertEqual(self.bot.Start(10,71084,100,[[],[]]),'invalid')
|
|
self.assertIsNone(self.bot.Start(10,71084,100,[[(3,8),(63,6)],[(71,30)]]))
|
|
self.advance()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 start 10 71084 100 2 3 8 63 6 1 71 30')
|
|
self.bot.Receive('MTSB1 error start running')
|
|
self.assertIsNone(self.bot.Start(10,71084,5,[[],[(71,1)]]));self.advance()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 start 10 71084 5 1 71 1') # empty first alternative skipped
|
|
|
|
def test_state_errors_and_timeout(self):
|
|
self.bot.Receive('MTSB1 state 1 10 11209 3 100 71084 4 running')
|
|
self.assertTrue(self.bot.Running());self.assertEqual(self.bot.state['scrolls'],4)
|
|
self.bot.Receive('MTSB1 state 0 10 11209 4 100 71084 3 matched')
|
|
self.assertFalse(self.bot.Running());self.assertEqual(self.bot.state['reason'],'matched')
|
|
self.bot.Receive('MTSB1 state 0 10 11209 4 100 71084 3 exploded')
|
|
self.assertEqual(self.bot.state['reason'],'matched') # unknown reason ignored
|
|
self.bot.Receive('MTSB1 error start busy');self.assertEqual(self.bot.error,'busy')
|
|
self.bot.Status();self.advance(4.1);self.advance()
|
|
self.assertEqual(self.bot.error,'unsupported');self.assertTrue(self.bot.unsupported)
|
|
|
|
def test_server_constants_match(self):
|
|
server=ROOT.parent/'metin2-server/src/server/game/src/port_switchbot.h'
|
|
if not server.exists():self.skipTest('server checkout not installed')
|
|
text=server.read_text(encoding='latin1')
|
|
reasons=set(re.findall(r'return "([a-z_]+)"',text))|{'running','stopped','idle'}
|
|
errors=set(re.findall(r'error [a-z%]+ ([a-z_]+)"',text))|{'invalid','rate'}
|
|
self.assertTrue(reasons-{'invalid','running','item','equipped','no_attr','scroll','target','busy'}<=set(self.ns['REASONS']))
|
|
self.assertTrue(errors<=set(self.ns['ERRORS']))
|
|
self.assertIn('kPortSwitchLimit=1000',text);self.assertEqual(max(self.ns['LIMITS']),1000)
|
|
self.assertEqual(self.ns['MAX_TARGETS'],5)
|
|
|
|
def test_window_scan_pick_start_and_notify(self):
|
|
game=sys.modules['game'];window_type=game.GameWindow
|
|
gw=window_type();gw.interface=types.SimpleNamespace(wndMiniMap=Widget())
|
|
game.GameWindow.Open(gw);w=self.ns['g_window'];w.Open()
|
|
self.assertEqual(w.items,[(10,11209)]);self.assertEqual(w.scrolls,[(71084,10)])
|
|
self.assertEqual(self.sent,['/mt_switchbot 1 attrs 10']) # the status request waits for this reply
|
|
self.load(w.bot);self.time+=.5;w.OnUpdate()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 status')
|
|
self.assertEqual(game.GameWindow.BINARY_ServerCommand_Run(gw,'MTSB1 state 0 0 0 0 0 0 0 idle'),1)
|
|
self.assertEqual(game.GameWindow.BINARY_ServerCommand_Run(gw,'other'),7)
|
|
w.TogglePick();w.Pick(0)
|
|
self.assertEqual(w.alternatives[0],[[3,12]])
|
|
w.Adjust(0,-1);self.assertEqual(w.alternatives[0][0][1],10);w.Adjust(0,1);w.Adjust(0,1);self.assertEqual(w.alternatives[0][0][1],12)
|
|
w.Start();self.time+=.5;w.OnUpdate()
|
|
self.assertEqual(self.sent[-1],'/mt_switchbot 1 start 10 71084 100 1 3 12')
|
|
w.bot.Receive('MTSB1 state 1 10 11209 1 100 71084 9 running');self.assertFalse(w.startButton.enabled)
|
|
self.inventory[40]=self.inventory.pop(10);sent=len(self.sent) # moved while running: follow the server's cell
|
|
w.bot.Receive('MTSB1 state 1 40 11209 2 100 71084 8 running');w.Scan()
|
|
self.assertEqual(w.Selected(),(40,11209));self.assertEqual(w.alternatives[0],[[3,12]])
|
|
self.assertEqual(w.Allowed(),[(3,12),(63,6),(71,30)]);self.time+=.5;w.OnUpdate();self.assertEqual(len(self.sent),sent)
|
|
w.bot.Receive('MTSB1 state 0 40 11209 2 100 71084 8 stopped')
|
|
self.inventory[5]=self.inventory.pop(40);w.Scan();self.assertEqual(w.Selected(),(5,11209)) # moved while idle: same vnum
|
|
self.inventory[10]=self.inventory.pop(5);w.Scan()
|
|
w.Hide();w.bot.Receive('MTSB1 state 0 10 11209 2 100 71084 8 matched')
|
|
self.assertTrue(w.IsShow());self.assertIn('已获得目标属性',self.chat[-1])
|
|
game.GameWindow.Close(gw);self.assertIsNone(self.ns['g_window'])
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|