1
from twisted.internet import task
2
from twisted.internet.protocol import connectionDone
3
from twisted.test import proto_helpers
4
from twisted.trial import unittest
6
from wlms import MetaServer
7
from wlms.protocol import GamePingFactory, NETCMD_METASERVER_PING
8
from wlms.utils import make_packet
9
from wlms.db.flatfile import FlatFileDatabase
12
logging.basicConfig(level=logging.CRITICAL)
15
class ClientStringTransport(proto_helpers.StringTransport):
16
def __init__(self, ip):
18
proto_helpers.StringTransport.__init__(self)
22
db = FlatFileDatabase("SirVer\t123456\tSUPERUSER\n" + "otto\tottoiscool\tREGISTERED\n")
23
self.ms = MetaServer(db)
24
self.clock = task.Clock()
25
self._cons = [ self.ms.buildProtocol(None) for i in range(10) ]
26
self._trs = [ ClientStringTransport("192.168.0.%i" % i) for i in range(10) ]
30
def create_game_pinger(gc, timeout):
31
self._gpfac = GamePingFactory(gc, timeout)
32
p = self._gpfac.buildProtocol(None)
33
tr = ClientStringTransport("GamePinger")
38
for c,tr in zip(self._cons, self._trs):
39
c.callLater = self.clock.callLater
40
c.seconds = self.clock.seconds
41
c.create_game_pinger = create_game_pinger
46
for idx,c in enumerate(self._cons):
49
raise RuntimeError("unexpected packet from client %i: %r" % (idx, p))
51
c.transport.loseConnection()
52
c.connectionLost(connectionDone)
54
def _recv(self, client):
55
d = self._trs[client].value()
56
self._trs[client].clear()
59
size = (ord(d[0]) << 8) + ord(d[1])
60
self.assertTrue(size <= len(d))
61
rv.append(d[2:size][:-1].split("\x00"))
63
self.clock.advance(1e-5)
66
def _mult_receive(self, clients):
67
"""Make sure everybody received the same data,
68
return this data once"""
69
ps = [ self._recv(c) for c in clients ]
70
for idx,p in enumerate(ps[1:], 1):
71
self.assertEqual(ps[idx-1], ps[idx])
74
def _send(self, client, *args, **kwargs):
75
c = self._cons[client]
76
c.dataReceived(make_packet(*args))
77
self.clock.advance(1e-5)
78
# End: Helper classes }}}
81
class TestBasics(_Base, unittest.TestCase):
82
def test_sending_absolute_garbage_too_short(self):
83
self._cons[0].dataReceived("\xff")
84
self.assertFalse(self._recv(0))
85
def test_sending_absolute_garbage(self):
86
self._cons[0].dataReceived("\xff\x37lkjdflsjflkjsf")
87
self.assertFalse(self._recv(0))
88
def test_sending_nonexisting_packet(self):
89
self._send(0, "BLUMBAQUATSCH")
91
self.assertEqual(p1, ['ERROR', 'GARBAGE_RECEIVED', "INVALID_CMD"])
92
def test_sending_toolittle_arguments_packet(self):
93
self._send(0, "LOGIN", "hi")
95
self.assertEqual(p1, ['ERROR', 'LOGIN', "Invalid integer: 'hi'"])
97
def test_sendtwopacketsinone(self):
98
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
100
self._cons[0].dataReceived("\x00\x0aCLIENTS\x00"*2)
101
p1,p2 = self._recv(0)
102
self.assertEqual(p1, p2)
103
self.assertEqual(p1, ['CLIENTS', '1', 'testuser', 'build-17', '', 'UNREGISTERED', ''])
105
def test_fragmented_packages(self):
106
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
109
self._cons[0].dataReceived("\x00\x0aCLI")
110
self._cons[0].dataReceived("ENTS\x00\x00\x0a")
112
self._cons[0].dataReceived("CLIENTS\x00\x00\x08")
114
self.assertEqual(p1, p2)
116
self.assertEqual(p1, ['CLIENTS', '1', 'testuser', 'build-17', '', 'UNREGISTERED', ''])
117
# End: Sending Basics }}}
118
# Test Regular Pinging {{{
119
class TestRegularPinging(_Base, unittest.TestCase):
122
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
125
def test_regular_cycle(self):
126
self.clock.advance(15) # Don't speak for 15 seconds
127
p1, = self._recv(0) # Expect a ping packet
128
self.assertEqual(p1, ["PING"])
129
self._send(0, "PONG")
131
self.clock.advance(15)
132
p1, = self._recv(0) # Expect a ping packet
133
self.assertEqual(p1, ["PING"])
134
self._send(0, "PONG")
136
def test_stay_quiet(self):
137
self.clock.advance(10.5)
138
p1, = self._recv(0) # Expect a ping packet
139
self.assertEqual(p1, ["PING"])
140
self.clock.advance(10.5)
141
p1, = self._recv(0) # Expect a timout packet
142
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
144
def test_delay_ping_by_regular_packet(self):
145
self.clock.advance(9.9)
146
self._send(0, "CHAT", "hello there", "")
148
self.assertEqual(p1, ["CHAT", 'bert', 'hello there', 'public'])
150
self.clock.advance(9.9)
151
self._send(0, "CHAT", "hello there", "")
153
self.assertEqual(p1, ["CHAT", 'bert', 'hello there', 'public'])
155
self.clock.advance(10.1)
156
p1, = self._recv(0) # Expect a ping packet
157
self.assertEqual(p1, ["PING"])
158
# End: Test Regular Pinging }}}
161
class TestLogin(_Base, unittest.TestCase):
162
def test_loginanon(self):
163
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
164
p1,p2,p3 = list(self._recv(0))
165
self.assertEqual(p1, ['LOGIN', 'testuser', "UNREGISTERED"])
166
self.assertEqual(p2[0], 'TIME')
167
self.assertEqual(p3, ['CLIENTS_UPDATE'])
169
def test_loginanon_unknown_protocol(self):
170
self._send(0, "LOGIN", 10, "testuser", "build-17", "false")
172
self.assertEqual(p1, ['ERROR', 'LOGIN', "UNSUPPORTED_PROTOCOL"])
174
def test_loginanon_withknownusername(self):
175
self._send(0, "LOGIN", 0, "SirVer", "build-17", "false")
176
p1,p2,p3 = list(self._recv(0))
177
self.assertEqual(p1, ['LOGIN', 'SirVer1', "UNREGISTERED"])
178
self.assertEqual(p2[0], 'TIME')
179
self.assertEqual(p3, ['CLIENTS_UPDATE'])
181
def test_loginanon_onewasalreadythere(self):
183
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
184
p1, p2, p3 = self._recv(0)
185
self.assertEqual(p1, ['LOGIN', 'testuser', "UNREGISTERED"])
186
self.assertEqual(p2[0], 'TIME')
187
self.assertEqual(p3, ['CLIENTS_UPDATE'])
190
self._send(1, "LOGIN", 0, "testuser", "build-17", "false")
191
p1, p2, p3 = self._recv(1)
192
# Note: Other username
193
self.assertEqual(p1, ['LOGIN', 'testuser1', "UNREGISTERED"])
194
self.assertEqual(p2[0], 'TIME')
195
self.assertEqual(p3, ['CLIENTS_UPDATE'])
197
# 0 should get an CLIENTS_UPDATE
199
self.assertEqual(p1, ['CLIENTS_UPDATE'])
201
def test_nonanon_login_correct_password(self):
202
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
203
p1, p2, p3 = self._recv(0)
204
self.assertEqual(p1, ['LOGIN', 'SirVer', "SUPERUSER"])
205
self.assertEqual(p2[0], 'TIME')
206
self.assertEqual(p3, ['CLIENTS_UPDATE'])
208
def test_nonanon_login_onewasalreadythere(self):
210
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
214
self._send(1, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
217
self.assertEqual(p1, ['ERROR', 'LOGIN', 'ALREADY_LOGGED_IN'])
219
def test_nonanon_login_twousers_password(self):
220
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
221
p1, p2, p3 = self._recv(0)
222
self.assertEqual(p1, ['LOGIN', 'SirVer', "SUPERUSER"])
223
self.assertEqual(p2[0], 'TIME')
224
self.assertEqual(p3, ['CLIENTS_UPDATE'])
226
self._send(1, "LOGIN", 0, "otto", "build-18", 1, "ottoiscool")
227
p1, p2, p3 = self._recv(1)
228
self.assertEqual(p1, ['LOGIN', 'otto', "REGISTERED"])
229
self.assertEqual(p2[0], 'TIME')
230
self.assertEqual(p3, ['CLIENTS_UPDATE'])
232
# 0 should get an CLIENTS_UPDATE
234
self.assertEqual(p1, ['CLIENTS_UPDATE'])
236
def test_nonanon_login_incorrect_password(self):
237
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "12345")
239
self.assertEqual(p, ['ERROR', 'LOGIN', 'WRONG_PASSWORD'])
240
# End: Test Login }}}
242
class TestMotd(_Base, unittest.TestCase):
245
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
246
self._send(1, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
247
self._send(2, "LOGIN", 0, "SirVer", "build-17", "true", "123456")
252
def test_setting_motd(self):
253
self._send(2, "MOTD", "Schnulz is cool!")
255
p1, = self._mult_receive(range(3))
256
self.assertEqual(p1, ["CHAT", "", "Schnulz is cool!", "system"])
258
def test_login_with_motd_set(self):
259
self._send(2, "MOTD", "Schnulz is cool!")
260
p1, = self._mult_receive(range(3))
262
self._send(4, "LOGIN", 0, "fasel", "build-18", "false")
263
p1,p2,p3,p4 = self._recv(4)
265
self.assertEqual(p1, ['LOGIN', 'fasel', "UNREGISTERED"])
266
self.assertEqual(p2[0], 'TIME')
267
self.assertEqual(p3, ['CLIENTS_UPDATE'])
268
self.assertEqual(p4, ["CHAT", "", "Schnulz is cool!", "system"])
270
p1, = self._mult_receive(range(3))
271
self.assertEqual(p1, ["CLIENTS_UPDATE"])
274
def test_setting_motd_forbidden(self):
275
self._send(1, "MOTD", "Schnulz is cool!")
276
self._send(0, "MOTD", "Schnulz is cool!")
278
p1, = self._mult_receive([0,1])
279
self.assertEqual(p1, ["ERROR", "MOTD", "DEFICIENT_PERMISSION"])
281
# TestAnnouncement {{{
282
class TestAnnouncement(_Base, unittest.TestCase):
285
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
286
self._send(1, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
287
self._send(2, "LOGIN", 0, "SirVer", "build-17", "true", "123456")
292
def test_setting_announcement(self):
293
self._send(2, "ANNOUNCEMENT", "Schnulz is cool!")
295
p1, = self._mult_receive(range(3))
296
self.assertEqual(p1, ["CHAT", "", "Schnulz is cool!", "system"])
298
def test_setting_announcement_forbidden(self):
299
self._send(1, "ANNOUNCEMENT", "Schnulz is cool!")
300
self._send(0, "ANNOUNCEMENT", "Schnulz is cool!")
302
p1, = self._mult_receive([0,1])
303
self.assertEqual(p1, ["ERROR", "ANNOUNCEMENT", "DEFICIENT_PERMISSION"])
304
# End: TestAnnouncement }}}
306
class TestRelogin_Anon(_Base, unittest.TestCase):
309
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
310
self._send(2, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
311
self._send(3, "LOGIN", 0, "SirVer", "build-17", "true", "123456")
316
def test_relogin_ping_and_reply(self):
317
self._send(1, "RELOGIN", 0, "bert", "build-17", "false")
318
p1, = self._recv(0) # Should have gotten a ping request
319
self.assertEqual(p1, ["PING"])
321
self._send(0, "PONG")
322
self.assertFalse(self._recv(0))
325
self.assertEqual(p1, ["ERROR", "RELOGIN", "CONNECTION_STILL_ALIVE"])
327
def test_relogin_ping_and_noreply(self):
328
self._send(1, "RELOGIN", 0, "bert", "build-17", "false")
330
self.assertEqual(p1, ["PING"])
332
self.clock.advance(6)
334
# Connection was terminated for old user
336
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
340
self.assertEqual(p1, ["RELOGIN"])
342
def test_relogin_notloggedin(self):
343
self._send(1, "RELOGIN", 0, "iamnotbert", "build-17", "false")
346
self.assertEqual(p1, ["ERROR", "RELOGIN", "NOT_LOGGED_IN"])
347
self.assertFalse(self._recv(0))
349
def test_relogin_wronginformation_wrong_proto(self):
350
self._send(1, "RELOGIN", 1, "bert", "build-17", "false")
352
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
353
self.assertFalse(self._recv(0))
355
def test_relogin_wronginformation_wrong_buildid(self):
356
self._send(1, "RELOGIN", 0, "bert", "uild-17", "false")
358
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
359
self.assertFalse(self._recv(0))
360
def test_relogin_wronginformation_wrong_loggedin(self):
361
self._send(1, "RELOGIN", 0, "bert", "build-17", "true")
363
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
364
self.assertFalse(self._recv(0))
365
def test_relogin_wronginformation_wrong_loggedin_nonanon(self):
366
self._send(1, "RELOGIN", 0, "otto", "build-17", "false")
368
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
369
self.assertFalse(self._recv(0))
370
def test_relogin_wronginformation_wrong_passwordl(self):
371
self._send(1, "RELOGIN", 0, "otto", "build-17", "true", "12345")
373
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
375
self.assertFalse(self._recv(0))
376
def test_relogin_loggedin_allcorrect(self):
377
self._send(1, "RELOGIN", 0, "otto", "build-17", "true", "ottoiscool")
379
self.assertEqual(p1, ["PING"])
381
self.clock.advance(6)
383
# Connection was terminated for old user
385
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
389
self.assertEqual(p1, ["RELOGIN"])
391
class TestReloginWhileInGame(_Base, unittest.TestCase):
394
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
395
self._send(2, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
396
self._send(0, "GAME_OPEN", "my cool game", 8)
397
self._send(2, "GAME_CONNECT", "my cool game")
398
self._gpc.dataReceived(NETCMD_METASERVER_PING)
402
def test_relogin_ping_and_noreply(self):
403
self.clock.advance(10)
404
p1, = self._mult_receive([0,2])
405
self.assertEqual(p1, ["PING"])
406
self._send(2, "PONG")
408
self.clock.advance(10)
410
self._send(2, "PONG")
412
# Connection was terminated for old user
414
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
415
self._cons[0].connectionLost(connectionDone)
416
self.assertEqual([], self._mult_receive([1,2]))
418
self._send(1, "RELOGIN", 0, "bert", "build-17", "false")
422
self.assertEqual(p1, ["RELOGIN"])
423
# p1, = self._recv(2)
424
# self.assertEqual(p1, ["CLIENTS_UPDATE"])
426
# Some garbage might have reached zero
429
self._send(1, "CLIENTS")
431
self.assertEqual(p1, ["CLIENTS", '2',
432
"bert", "build-17", "my cool game", "UNREGISTERED", "",
433
"otto", "build-17", "my cool game", "REGISTERED", "",
437
# End: Test Relogin }}}
439
class TestChat(_Base, unittest.TestCase):
442
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
443
self._send(1, "LOGIN", 0, "ernie", "build-17", "false")
447
def test_public_chat(self):
448
self._send(0, "CHAT", "hello there", "")
449
p0, = self._mult_receive([0,1])
450
self.assertEqual(p0, ["CHAT", "bert", "hello there", "public"])
452
def test_sanitize_public_chat(self):
453
self._send(0, "CHAT", "hello <rt>there</rt>\nhow<rtdoyoudo", "")
454
p0, = self._mult_receive([0,1])
455
self.assertEqual(p0, ["CHAT", "bert", "hello <rt>there</rt>\nhow<rtdoyoudo", "public"])
457
def test_private_chat(self):
458
self._send(0, "CHAT", "hello there", "ernie")
459
self.assertEqual([], self._recv(0))
461
self.assertEqual(p1, ["CHAT", "bert", "hello there", "private"])
463
def test_sanitize_private_chat(self):
464
self._send(0, "CHAT", "hello <rt>there</rt>\nhow<rtdoyoudo", "ernie")
465
self.assertEqual([], self._recv(0))
467
self.assertEqual(p1, ["CHAT", "bert", "hello <rt>there</rt>\nhow<rtdoyoudo", "private"])
471
# Test Game Creation/Joining {{{
472
class TestGameCreation(_Base, unittest.TestCase):
475
self._send(0, "LOGIN", 0, "bert", "build-16", "false")
476
self._send(1, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
477
self._send(2, "LOGIN", 0, "SirVer", "build-18", "true", "123456")
482
def test_create_game_and_ping_reply(self):
483
self._send(0, "GAME_OPEN", "my cool game", 8)
484
b1,b2 = self._mult_receive(range(3))
485
self.assertEqual(b1, ["GAMES_UPDATE"])
486
self.assertEqual(b2, ["CLIENTS_UPDATE"])
488
self._send(1, "CLIENTS")
491
self.assertEqual(p, ["CLIENTS", "3",
492
"bert", "build-16", "my cool game", "UNREGISTERED", "",
493
"otto", "build-17", "", "REGISTERED", "",
494
"SirVer", "build-18", "", "SUPERUSER", ""
497
data = self._gptr.value(); self._gptr.clear()
498
self.assertEqual(data, NETCMD_METASERVER_PING)
499
self._gpc.dataReceived(NETCMD_METASERVER_PING)
501
p1,p2 = self._recv(0)
502
self.assertEqual(p1, ["GAME_OPEN"])
503
self.assertEqual(p2, ["GAMES_UPDATE"])
504
p1, = self._mult_receive([1,2])
505
self.assertEqual(p1, ["GAMES_UPDATE"])
507
self._send(1, "CLIENTS")
510
self.assertEqual(p, ["CLIENTS", "3",
511
"bert", "build-16", "my cool game", "UNREGISTERED", "",
512
"otto", "build-17", "", "REGISTERED", "",
513
"SirVer", "build-18", "", "SUPERUSER", ""
516
def test_create_game_no_connection_to_game(self):
517
self._send(0, "GAME_OPEN", "my cool game", 8)
518
b1,b2 = self._mult_receive(range(3))
519
self.assertEqual(b1, ["GAMES_UPDATE"])
520
self.assertEqual(b2, ["CLIENTS_UPDATE"])
522
self._gpfac.clientConnectionFailed(None, None)
523
p1,p2 = self._recv(0)
524
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_TIMEOUT"])
525
self.assertEqual(p2, ["GAMES_UPDATE"])
527
p1, = self._mult_receive([1,2])
528
self.assertEqual(p1, ["GAMES_UPDATE"])
530
self._send(2, "CLIENTS")
533
self.assertEqual(p, ["CLIENTS", "3",
534
"bert", "build-16", "my cool game", "UNREGISTERED", "",
535
"otto", "build-17", "", "REGISTERED", "",
536
"SirVer", "build-18", "", "SUPERUSER", ""
539
self._send(2, "GAMES")
541
self.assertEqual(p, ["GAMES", "1", "my cool game", "build-16", "false"])
543
def test_create_game_twice_pre_ping(self):
544
self._send(0, "GAME_OPEN", "my cool game", 8)
545
b1,b2 = self._mult_receive(range(3))
547
self._send(1, "GAME_OPEN", "my cool game", 12)
549
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_EXISTS"])
551
data = self._gptr.value(); self._gptr.clear()
552
self.assertEqual(data, NETCMD_METASERVER_PING)
553
self._gpc.dataReceived(NETCMD_METASERVER_PING)
555
self._mult_receive([1,2])
557
self._send(2, "CLIENTS")
560
self.assertEqual(p, ["CLIENTS", "3",
561
"bert", "build-16", "my cool game", "UNREGISTERED", "",
562
"otto", "build-17", "", "REGISTERED", "",
563
"SirVer", "build-18", "", "SUPERUSER", ""
565
def test_create_game_twice_post_ping(self):
566
self._send(0, "GAME_OPEN", "my cool game", 8)
567
b1,b2 = self._mult_receive(range(3))
569
data = self._gptr.value(); self._gptr.clear()
570
self.assertEqual(data, NETCMD_METASERVER_PING)
571
self._gpc.dataReceived(NETCMD_METASERVER_PING)
573
self._mult_receive([1,2])
575
self._send(1, "GAME_OPEN", "my cool game", 12)
577
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_EXISTS"])
579
self._send(2, "CLIENTS")
582
self.assertEqual(p, ["CLIENTS", "3",
583
"bert", "build-16", "my cool game", "UNREGISTERED", "",
584
"otto", "build-17", "", "REGISTERED", "",
585
"SirVer", "build-18", "", "SUPERUSER", ""
588
def test_create_game_and_no_first_ping_reply(self):
589
self._send(0, "GAME_OPEN", "my cool game", 8)
590
b1,b2 = self._mult_receive(range(3))
591
self.assertEqual(b1, ["GAMES_UPDATE"])
592
self.assertEqual(b2, ["CLIENTS_UPDATE"])
594
self.clock.advance(6)
595
p1,p2 = self._recv(0)
596
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_TIMEOUT"])
597
self.assertEqual(p2, ["GAMES_UPDATE"])
599
p1, = self._mult_receive([1,2])
600
self.assertEqual(p2, ["GAMES_UPDATE"])
602
self._send(2, "CLIENTS")
605
self.assertEqual(p, ["CLIENTS", "3",
606
"bert", "build-16", "my cool game", "UNREGISTERED", "",
607
"otto", "build-17", "", "REGISTERED", "",
608
"SirVer", "build-18", "", "SUPERUSER", ""
611
self._send(2, "GAMES")
613
self.assertEqual(p, ["GAMES", "1", "my cool game", "build-16", "false"])
615
def test_create_game_and_no_second_ping_reply(self):
616
self._send(0, "GAME_OPEN", "my cool game", 8)
617
b1,b2 = self._mult_receive(range(3))
618
self.assertEqual(b1, ["GAMES_UPDATE"])
619
self.assertEqual(b2, ["CLIENTS_UPDATE"])
621
data = self._gptr.value(); self._gptr.clear()
622
self.assertEqual(data, NETCMD_METASERVER_PING)
623
self._gpc.dataReceived(NETCMD_METASERVER_PING)
624
p1,p2 = self._recv(0)
625
self.assertEqual(p1, ["GAME_OPEN"])
626
self.assertEqual(p2, ["GAMES_UPDATE"])
627
p1, = self._mult_receive([1,2])
628
self.assertEqual(p1, ["GAMES_UPDATE"])
630
self.clock.advance(119)
631
p1, = self._mult_receive(range(3))
632
self.assertEqual(p1, ["PING"])
633
for i in range(3): self._send(i, "PONG")
635
data = self._gptr.value(); self._gptr.clear()
636
self.assertEqual(data, '')
638
self.clock.advance(2)
639
data = self._gptr.value(); self._gptr.clear()
640
self.assertEqual(data, NETCMD_METASERVER_PING)
642
# Now, do not answer ping
643
self.clock.advance(118)
644
p1, = self._mult_receive(range(3))
645
self.assertEqual(p1, ["PING"])
646
for i in range(3): self._send(i, "PONG")
648
self.clock.advance(3)
650
p1, = self._mult_receive(range(3))
651
self.assertEqual(p1, ["GAMES_UPDATE"])
653
self._send(2, "CLIENTS")
656
self.assertEqual(p, ["CLIENTS", "3",
657
"bert", "build-16", "my cool game", "UNREGISTERED", "",
658
"otto", "build-17", "", "REGISTERED", "",
659
"SirVer", "build-18", "", "SUPERUSER", ""
662
self._send(2, "GAMES")
664
self.assertEqual(p, ["GAMES", "0"])
666
def test_join_game(self):
667
self._send(0, "GAME_OPEN", "my cool game", 8)
669
p1,p2 = self._mult_receive([1,2])
671
self._send(1, "GAME_CONNECT", "my cool game")
672
p1,p2 = self._recv(1)
673
self.assertEqual(p1, ["GAME_CONNECT", "192.168.0.0"])
674
self.assertEqual(p2, ["CLIENTS_UPDATE"])
676
p1, = self._mult_receive([0,2])
677
self.assertEqual(p1, ["CLIENTS_UPDATE"])
679
self._send(2, "CLIENTS")
681
self.assertEqual(p, ["CLIENTS", "3",
682
"bert", "build-16", "my cool game", "UNREGISTERED", "",
683
"otto", "build-17", "my cool game", "REGISTERED", "",
684
"SirVer", "build-18", "", "SUPERUSER", ""
687
def test_join_full_game(self):
688
self._send(0, "GAME_OPEN", "my cool game", 1)
690
p1,p2 = self._mult_receive([1,2])
692
self._send(1, "GAME_CONNECT", "my cool game")
694
self.assertEqual(p1, ["ERROR", "GAME_CONNECT", "GAME_FULL"])
696
self._send(2, "CLIENTS")
698
self.assertEqual(p, ["CLIENTS", "3",
699
"bert", "build-16", "my cool game", "UNREGISTERED", "",
700
"otto", "build-17", "", "REGISTERED", "",
701
"SirVer", "build-18", "", "SUPERUSER", ""
704
def test_non_existing_Game(self):
705
self._send(1, "GAME_CONNECT", "my cool game")
707
self.assertEqual(p1, ["ERROR", "GAME_CONNECT", "NO_SUCH_GAME"])
709
self._send(2, "CLIENTS")
711
self.assertEqual(p, ["CLIENTS", "3",
712
"bert", "build-16", "", "UNREGISTERED", "",
713
"otto", "build-17", "", "REGISTERED", "",
714
"SirVer", "build-18", "", "SUPERUSER", ""
716
# End: Test Game Creation/Joining }}}
717
# Test Game Starting {{{
718
class TestGameStarting(_Base, unittest.TestCase):
721
self._send(0, "LOGIN", 0, "bert", "build-16", "false")
722
self._send(1, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
723
self._send(2, "LOGIN", 0, "SirVer", "build-18", "true", "123456")
724
self._send(0, "GAME_OPEN", "my cool game", 8)
725
self._send(1, "GAME_CONNECT", "my cool game")
730
def test_start_game_without_being_in_one(self):
731
self._send(2, "GAME_START")
734
self.assertEqual(p1, ["ERROR", "GARBAGE_RECEIVED", "INVALID_CMD"])
736
def test_start_game_not_host(self):
737
self._send(1, "GAME_START")
740
self.assertEqual(p1, ["ERROR", "GAME_START", "DEFICIENT_PERMISSION"])
742
def test_start_game(self):
743
self._send(0, "GAME_START")
744
p1,p2 = self._recv(0)
745
self.assertEqual(p1, ["GAME_START"])
747
p1, = self._mult_receive([1,2])
748
self.assertEqual(p1, p2)
749
self.assertEqual(p2, ["GAMES_UPDATE"])
750
# End: Game Starting }}}
751
# Test Game Leaving {{{
752
class TestGameLeaving(_Base, unittest.TestCase):
755
self._send(0, "LOGIN", 0, "bert", "build-16", "false")
756
self._send(1, "LOGIN", 0, "otto", "build-17", "true", "ottoiscool")
757
self._send(2, "LOGIN", 0, "SirVer", "build-18", "true", "123456")
758
self._send(0, "GAME_OPEN", "my cool game", 8)
759
self._send(1, "GAME_CONNECT", "my cool game")
764
def test_leave_game_nothost(self):
765
self._send(1, "GAME_DISCONNECT")
767
self.assertEqual(p1, ["CLIENTS_UPDATE"])
769
p1, = self._mult_receive([0,2])
770
self.assertEqual(p1, ["CLIENTS_UPDATE"])
772
self._send(2, "CLIENTS")
774
self.assertEqual(p, ["CLIENTS", "3",
775
"bert", "build-16", "my cool game", "UNREGISTERED", "",
776
"otto", "build-17", "", "REGISTERED", "",
777
"SirVer", "build-18", "", "SUPERUSER", ""
780
def test_leave_game_host(self):
781
self._send(0, "GAME_DISCONNECT")
782
p1,p2 = self._recv(0)
783
self.assertEqual(p1, ["CLIENTS_UPDATE"])
784
self.assertEqual(p2, ["GAMES_UPDATE"])
786
p1,p2 = self._mult_receive([1,2])
787
self.assertEqual(p1, ["CLIENTS_UPDATE"])
788
self.assertEqual(p2, ["GAMES_UPDATE"])
790
self._send(2, "CLIENTS")
792
self.assertEqual(p, ["CLIENTS", "3",
793
"bert", "build-16", "", "UNREGISTERED", "",
794
"otto", "build-17", "my cool game", "REGISTERED", "",
795
"SirVer", "build-18", "", "SUPERUSER", ""
798
def test_leave_not_in_game(self):
799
self._send(2, "GAME_DISCONNECT")
801
self.assertEqual(p1, ["ERROR", "GARBAGE_RECEIVED", "INVALID_CMD"])
803
self._send(2, "CLIENTS")
805
self.assertEqual(p, ["CLIENTS", "3",
806
"bert", "build-16", "my cool game", "UNREGISTERED", "",
807
"otto", "build-17", "my cool game", "REGISTERED", "",
808
"SirVer", "build-18", "", "SUPERUSER", ""
810
# End: Game Leaving }}}
added
removed
wlms/test/test.py
