metin2-client/tests/wiki/test_service.py

import sys
import ast
import re
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 ServiceTests(unittest.TestCase):
    def setUp(self):
        self.time=1.;self.sent=[];self.replies=[]
        app=types.ModuleType('app');app.GetTime=lambda:self.time
        net=types.ModuleType('net');net.SendChatPacket=self.sent.append
        game=types.ModuleType('game');game.GameWindow=type('GameWindow',(),dict(Open=lambda s:None,Close=lambda s:None,BINARY_ServerCommand_Run=lambda s,line:7))
        ui=types.ModuleType('ui');ui.Window=object
        data=types.ModuleType('mt_wiki_data');exec(script('WikiDataScript.inc'),data.__dict__)
        self.previous={n:sys.modules.get(n) for n in ('app','net','game','ui','mt_wiki_data')}
        sys.modules.update(app=app,net=net,game=game,ui=ui,mt_wiki_data=data)
        self.ns={};exec(script('FeatureServiceScript.inc'),self.ns)
        self.s=self.ns['Service']()

    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 request(self,kind='mob',subject=101,offset=0):
        rid=self.s.Request(kind,subject,offset,lambda result,error:self.replies.append((result,error)))
        self.time+=.6;self.s.Tick()
        if self.s.pending['kind']=='cap':
            cap=self.s.pending['id'];self.s.Receive('MTF1 cap %d 1 15 1700000000'%cap)
            self.time+=.4;self.s.Tick()
        return rid

    def test_capability_handshake_atomic_page_and_stale_replies(self):
        rid=self.request()
        self.assertEqual(len(self.sent),2)
        self.s.Receive('MTF1 begin %d mob 101 0 1 0 1700000000'%rid)
        self.s.Receive('MTF1 row %d 0 30006 1 1 0 0 100 0 0'%(rid+100))
        self.assertFalse(self.replies)
        self.s.Receive('MTF1 row %d 0 30006 1 1 0 0 100 0 0'%rid)
        self.assertFalse(self.replies)
        self.s.Receive('MTF1 end %d 1'%rid)
        self.assertEqual(self.replies[0][0]['rows'][0][0],30006)
        self.assertIsNone(self.replies[0][1])

    def test_old_server_timeout_is_unsupported_not_empty_results(self):
        self.s.Request('mob',101,0,lambda d,e:self.replies.append((d,e)))
        self.time+=1;self.s.Tick();self.time+=5;self.s.Tick();self.time+=1;self.s.Tick()
        self.assertEqual(self.replies,[(None,'unsupported')])
        self.assertEqual(len(self.sent),1)

    def test_partial_duplicate_and_bad_numeric_rejected(self):
        for bad in ('MTF1 end {r} 1','MTF1 row {r} 1 19 1 1 0 0 1 0 0','MTF1 row {r} 0 19 -1 1 0 0 1 0 0'):
            rid=self.request()
            self.s.Receive('MTF1 begin %d mob 101 0 1 0 1700000000'%rid)
            self.s.Receive(bad.format(r=rid))
            self.assertEqual(self.replies[-1],(None,'bad_reply'))

    def test_stats_signed_values_and_complete_snapshot(self):
        rid=self.request('stats',0)
        self.s.Receive('MTF1 begin %d stats 0 0 2 0 1700000000'%rid)
        self.s.Receive('MTF1 stat %d 0 stone -10'%rid);self.s.Receive('MTF1 stat %d 1 maxhp 1500'%rid)
        self.s.Receive('MTF1 end %d 2'%rid)
        self.assertEqual(dict(self.replies[-1][0]['rows']),dict(stone=-10,maxhp=1500))

    def test_rate_retries_bounded_and_close_cancels_callbacks(self):
        rid=self.request()
        for i in range(3):
            self.s.Receive('MTF1 error %d rate'%rid)
            self.time+=1;self.s.Tick()
        self.assertEqual(self.replies[-1],(None,'rate'))
        rid=self.request();self.s.Close();self.s.Receive('MTF1 end %d 0'%rid)
        self.assertEqual(len(self.replies),1)

    def test_stats_keys_match_current_server_not_ava2_point_numbers(self):
        path=ROOT.parent/'metin2-server/src/server/game/src/port_features.h'
        if not path.exists():self.skipTest('server checkout not installed')
        tree=ast.parse(script('DetailsScript.inc'))
        layout=next(ast.literal_eval(n.value) for n in tree.body if isinstance(n,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='STAT_LAYOUT' for t in n.targets))
        keys=re.findall(r'\{"([a-z_]+)",POINT_[A-Z_]+\}',path.read_text())
        self.assertEqual(set(keys),{r[1] for r in layout})
        self.assertEqual(len(keys),len(set(keys)))

    def test_dungeon_capability_snapshot_and_partial_rejection(self):
        self.s.mask=31
        rows=dict(lair_end=1700003600,lair_join_end=1700000300,lair_phase=1,lair_ticket=30179,
                  lair_tickets=3,lair_boss=2430,lair_owned=7,lair_alive=1,lair_npc=30121,lair_map=208)
        for valid in (True,False):
            rid=self.request('dungeons',0)
            self.assertTrue(self.sent[-1].endswith(' dungeons'))
            values=list(rows.items()) if valid else list(rows.items())[:-1]
            self.s.Receive('MTF1 begin %d dungeons 0 0 %d 0 1700000000'%(rid,len(values)))
            for i,(k,v) in enumerate(values):self.s.Receive('MTF1 stat %d %d %s %d'%(rid,i,k,v))
            self.s.Receive('MTF1 end %d %d'%(rid,len(values)))
            if valid:self.assertEqual(dict(self.replies[-1][0]['rows']),rows)
            else:self.assertEqual(self.replies[-1],(None,'bad_reply'))

    def test_calendar_real_interval_rows_and_invalid_scope(self):
        self.s.mask=63
        for scope,expected in ((0,None),(256,'bad_reply')):
            rid=self.request('calendar',1)
            self.s.Receive('MTF1 begin %d calendar 1 0 1 0 1700000000'%rid)
            self.s.Receive('MTF1 event %d 0 1 1 1700000000 1700003600 %d 0 1'%(rid,scope))
            self.s.Receive('MTF1 end %d 1'%rid)
            self.assertEqual(self.replies[-1][1],expected)
            if expected is None:self.assertEqual(self.replies[-1][0]['rows'][0][1],1)

    def test_boss_rows_signed_positions_and_unsupported_on_old_cap(self):
        self.s.mask=511
        rid=self.request('bosses',0)
        self.assertEqual(self.sent[-1],'/mt_features 1 %d bosses'%rid)
        self.s.Receive('MTF1 begin %d bosses 0 0 2 0 1700000000'%rid)
        self.s.Receive('MTF1 boss %d 0 2307 68 51200 -300 1 1 1 1700000900 9000'%rid)
        self.s.Receive('MTF1 boss %d 1 2191 125 0 0 99 0 1 0 1800'%rid)
        self.s.Receive('MTF1 end %d 2'%rid)
        self.assertIsNone(self.replies[-1][1])
        self.assertEqual(self.replies[-1][0]['rows'],[[2307,68,51200,-300,1,1,1,1700000900,9000],[2191,125,0,0,99,0,1,0,1800]])
        for bad in ('2307 68 0 0 1 2 1 0 9000','0 68 0 0 1 0 1 0 9000','2307 68 0 0 1 0 1 0 0','2307 68 0 0 256 0 1 0 9000','2307 68 0 0 1 0 1 0'):
            rid=self.request('bosses',0)
            self.s.Receive('MTF1 begin %d bosses 0 0 1 0 1700000000'%rid)
            self.s.Receive('MTF1 boss %d 0 %s'%(rid,bad));self.s.Receive('MTF1 end %d 1'%rid)
            self.assertEqual(self.replies[-1],(None,'bad_reply'),bad)
        self.assertEqual(self.s.Request('bosses',1,0,lambda d,e:self.replies.append((d,e))),0)
        self.assertEqual(self.replies[-1],(None,'invalid'))
        self.s.mask=255 # a server deployed before boss tracking
        self.s.Request('bosses',0,0,lambda d,e:self.replies.append((d,e)));self.time+=1;self.s.Tick()
        self.assertEqual(self.replies[-1],(None,'unsupported'))

    def test_future_capability_bits_do_not_disable_known_features(self):
        self.s.Request('stats',0,0,lambda d,e:None)
        self.time+=1;self.s.Tick();rid=self.s.pending['id']
        self.s.Receive('MTF1 cap %d 1 65551 1700000000'%rid)
        self.assertEqual(self.s.mask,65551)
        self.time+=1;self.s.Tick()
        self.assertTrue(self.sent[-1].endswith(' stats'))

    def test_empty_page_distinct_from_unsupported_and_invalid_requests(self):
        rid=self.request()
        self.s.Receive('MTF1 begin %d mob 101 0 0 0 1700000000'%rid)
        self.s.Receive('MTF1 end %d 0'%rid)
        self.assertEqual(self.replies[-1][0]['rows'],[])
        self.assertIsNone(self.replies[-1][1])
        self.s.Request('delete',0,0,lambda d,e:self.replies.append((d,e)))
        self.assertEqual(self.replies[-1],(None,'invalid'))


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