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_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()