1# -*- coding: utf-8 -*- 

2from __future__ import annotations 

3 

4# -- stdlib -- 

5# -- third party -- 

6import gevent 

7 

8# -- own -- 

9from ..mock import Environ, EventTap 

10from utils.misc import BatchList 

11from .common import UserInputFuzzingHandler, let_it_go 

12from game.base import sync_primitive 

13from thb.actions import THBEventHandler 

14from thb.actions import COMMON_EVENT_HANDLERS 

15 

16 

17# -- code -- 

18class GameEnded(Exception): 

19 pass 

20 

21 

22class ParanoidSyncHandler(THBEventHandler): 

23 interested = ['action_after'] 

24 

25 def handle(self, evt_type, arg): 

26 g = self.game 

27 if hasattr(g, 'players'): 

28 me = [len(ch.cards) for ch in g.players] 

29 svr = sync_primitive(me, g.players) 

30 assert me == svr, (me, svr) 

31 return arg 

32 

33 

34class TestFuzzTHBattleRole(object): 

35 def testFuzzTHBattleRole(self): 

36 env = Environ() 

37 t = EventTap() 

38 

39 # from thb.cards import definition 

40 # definition.card_definition = [(definition.ExinwanCard, 1, 1)] * 1000 

41 

42 COMMON_EVENT_HANDLERS.add(ParanoidSyncHandler) 

43 

44 me = gevent.getcurrent() 

45 

46 def fail_crash(g): 

47 e = Exception('GAME CRASH') 

48 e.__cause__ = g.runner.exception 

49 gevent.kill(me, e) 

50 return g 

51 

52 s = env.server_core() 

53 cl = BatchList([env.client_core() for _ in range(8)]) 

54 t.tap(s, *cl) 

55 

56 c1r = BatchList(cl[1:]) 

57 c = cl[0] 

58 names = ( 

59 'Reimu', 'Marisa', 'Youmu', 'Sakuya', 

60 'Satori', 'Koishi', 'Remilia', 'Flandre', 

61 ) 

62 for i, name in zip(cl, names): 

63 i.auth.login(name) 

64 gevent.idle(-100) 

65 assert all(cl.auth.uid) 

66 c.room.create('Test1', 'THBattleRole', {}) 

67 gevent.idle(-100) 

68 gid = c.game.gid_of(t[c.events.game_joined]) 

69 c1r.room.join(gid) 

70 gevent.idle(-100) 

71 assert [gid] * 8 == [i.game.gid_of(t[i.events.game_joined]) for i in cl] 

72 

73 s.events.game_crashed += fail_crash 

74 for i in cl: 

75 g = t[i.events.game_joined] 

76 i.events.game_crashed += fail_crash 

77 g.event_observer = UserInputFuzzingHandler(g) 

78 

79 cl.room.get_ready() 

80 gevent.idle(-100) 

81 assert [gid] * 8 == [i.game.gid_of(t[i.events.game_started]) for i in cl] 

82 

83 gevent.idle(-100) 

84 

85 def game_ended(g): 

86 gevent.kill(me, GameEnded()) 

87 return g 

88 

89 s.events.game_ended += game_ended 

90 

91 [i.game.start_game(t[i.events.game_started]) for i in cl] 

92 

93 try: 

94 # gevent.sleep(5) 

95 let_it_go(*cl) 

96 except GameEnded: 

97 pass