metin2-client/tests/wiki/test_channel.py

import os
import sys
import tempfile
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 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.text=text;self.shown=True
    def SetEnabled(self,enabled):self.enabled=enabled


class Stream(object):
    def __init__(self):
        self.id='admin';self.pwd='secret';self.addr='10.0.0.1';self.port=13000;self.account_addr='10.0.0.1';self.account_port=11000
        self.slot=2;self.isAutoSelect=0
    def SetConnectInfo(self,addr,port,account_addr=0,account_port=0):
        self.addr=addr;self.port=port;self.account_addr=account_addr;self.account_port=account_port
    def SetCharacterSlot(self,slot):self.slot=slot
    def GetCharacterSlot(self):return self.slot


class Login(object):
    # introLogin.LoginWindow.Open resets the connect info and the slot from loginInfo.xml.
    def __init__(self,stream,log):self.stream=stream;self.log=log
    def Open(self):self.stream.SetConnectInfo('',0,'',0);self.stream.SetCharacterSlot(0);self.stream.isAutoSelect=0
    def OnUpdate(self):pass
    def OnLoginFailure(self,error):self.log.append(('failure',error))
    def OnConnectFailure(self):self.log.append(('connect',))
    def Connect(self,id,pwd):self.log.append(('connect_to',self.stream.addr,self.stream.port,id,pwd))


class ChannelTests(unittest.TestCase):
    def setUp(self):
        self.time=1.;self.sent=[];self.chat=[];self.log=[];self.serverinfo=[]
        self.cwd=os.getcwd();self.tmp=tempfile.TemporaryDirectory();os.chdir(self.tmp.name)
        mods={n:types.ModuleType(n) for n in ('app','chat','game','grp','introLogin','net','serverInfo','ui','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['net'].SetServerInfo=self.serverinfo.append
        mods['game'].GameWindow=type('GameWindow',(),dict(Open=lambda s:None,Close=lambda s:None,OnUpdate=lambda s:None))
        mods['introLogin'].LoginWindow=Login
        si=mods['serverInfo']
        si.REGION_DICT={0:{1:{'name':'Metin2','channel':{
            1:{'name':'CH1','ip':'10.0.0.1','tcp_port':13000},2:{'name':'CH2','ip':'10.0.0.1','tcp_port':13010},
            3:{'name':'CH3','ip':'10.0.0.1','tcp_port':13020}}},
            2:{'name':'Test','channel':{1:{'name':'CH1','ip':'127.0.0.1','tcp_port':13000}}}}}
        si.REGION_AUTH_SERVER_DICT={0:{1:{'ip':'10.0.0.1','port':11000},2:{'ip':'127.0.0.1','port':11000}}}
        grp=mods['grp'];grp.SetColor=grp.RenderBar=grp.RenderBox=lambda *a:None
        mods['ui'].Window=Widget
        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'].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_channel'};exec(script('ChannelScript.inc'),self.ns)
        self.stream=Stream();self.game=sys.modules['game'];self.gw=self.game.GameWindow()
        self.gw.stream=self.stream;self.gw.interface=types.SimpleNamespace(wndMiniMap=Widget())

    def tearDown(self):
        os.chdir(self.cwd);self.tmp.cleanup()
        for name,value in self.previous.items():
            if value is None:sys.modules.pop(name,None)
            else:sys.modules[name]=value

    def test_locate(self):
        Locate=self.ns['Locate']
        self.assertEqual(Locate('10.0.0.1',13010),(0,1,2))
        self.assertEqual(Locate('127.0.0.1',13000),(0,2,1))
        self.assertEqual(Locate('10.0.0.1',13040),(0,1,0)) # unknown port: first server, current unknown
        Path('channel.inf').write_text('2 1 0')
        self.assertEqual(Locate('10.0.0.1',13040),(0,2,0)) # the server last picked on the login board
        Path('channel.inf').write_text('9 1 0');self.assertEqual(Locate('x',1),(0,1,0))

    def test_full_switch(self):
        game=self.game;gw=self.gw
        game.GameWindow.Open(gw);w=self.ns['g_window'];self.assertEqual(self.ns['g_entry'].pos,(4,304))
        w.Open();self.assertEqual(len(w.channelButtons),3);self.assertIn('CH1',w.current.text)
        self.assertFalse(w.channelButtons[0].enabled);self.assertIn('【',w.channelButtons[0].text)
        w.channelButtons[1].event();self.assertEqual(self.sent,['/logout'])
        self.assertTrue(all(not b.enabled for b in w.channelButtons));self.assertIn('正在退出',w.status.text)
        w.channelButtons[2].event();self.assertEqual(self.sent,['/logout']) # busy: a second /logout would cancel it
        game.GameWindow.Close(gw) # server closed the session -> SetLoginPhase
        login=sys.modules['introLogin'].LoginWindow(self.stream,self.log);login.Open()
        self.assertEqual(self.log,[('connect_to','10.0.0.1',13010,'admin','secret')])
        self.assertEqual((self.stream.account_addr,self.stream.account_port,self.stream.slot,self.stream.isAutoSelect),('10.0.0.1',11000,2,1))
        self.assertEqual(self.serverinfo,['Metin2, CH2']);self.assertEqual(Path('channel.inf').read_text(),'1 2 0')
        game.GameWindow.Open(gw) # introSelect auto-selected slot 2 and the game started
        self.assertEqual(self.stream.isAutoSelect,0);self.assertIsNone(self.ns['g_switcher'].stage)
        self.assertIn('已换到 CH2',self.chat[-1])
        w=self.ns['g_window'];w.Open();self.assertIn('CH2',w.current.text);self.assertTrue(w.channelButtons[0].enabled)
        login.Open();self.assertEqual(len(self.log),1) # a later login window is left alone

    def test_logout_refused_times_out(self):
        self.game.GameWindow.Open(self.gw);w=self.ns['g_window'];w.Open();w.channelButtons[2].event()
        self.time+=19;self.game.GameWindow.OnUpdate(self.gw);self.assertEqual(self.ns['g_switcher'].stage,'logout')
        self.time+=1;self.game.GameWindow.OnUpdate(self.gw);self.assertIsNone(self.ns['g_switcher'].stage)
        self.assertIn('换线已取消',self.chat[-1]);self.assertTrue(w.channelButtons[2].enabled)

    def test_login_failure_clears_auto_select(self):
        self.game.GameWindow.Open(self.gw);w=self.ns['g_window'];w.Open();w.channelButtons[1].event()
        login=sys.modules['introLogin'].LoginWindow(self.stream,self.log);login.Open()
        self.assertEqual(self.stream.isAutoSelect,1)
        login.OnLoginFailure('ALREADY');self.assertEqual(self.stream.isAutoSelect,0)
        self.assertEqual(self.log[-1],('failure','ALREADY'));self.assertIn('登录目标频道失败',self.chat[-1])
        self.assertIsNone(self.ns['g_switcher'].stage)

    def test_refuses_without_account_or_same_channel(self):
        s=self.ns['g_switcher'];place=(0,1,1)
        self.assertFalse(s.Start(self.stream,place,1));self.assertEqual(s.error,'channel')
        self.assertFalse(s.Start(self.stream,place,7));self.assertEqual(s.error,'channel')
        self.stream.pwd='';self.assertFalse(s.Start(self.stream,place,2));self.assertEqual(s.error,'account')
        self.assertEqual(self.sent,[])

    def test_server_logout_countdown_fits_timeout(self):
        server=ROOT.parent/'metin2-server/src/server/game/src/cmd_general.cpp'
        if not server.exists():self.skipTest('server checkout not installed')
        text=server.read_text(encoding='latin1');body=text.split('ACMD(do_cmd)',1)[1].split('\n}\n',1)[0]
        self.assertIn('info->left_second = 10;',body);self.assertLess(10+2,self.ns['LOGOUT_TIMEOUT'])
        self.assertIn('event_cancel(&ch->m_pkTimedEvent)',body) # why the switcher never sends /logout twice


if __name__=='__main__':unittest.main()