~widelands-dev/widelands-website/trunk
» Revision
247.1.1
(compared to revision 323)
: wlms/test/test.py
« back to all changes in this revision
Viewing changes to wlms/test/test.py
- browse files at revision 247.1.1
- compare with another revision
- download diff
- download tarball
- view history from revision 247.1.1
- Committer: Holger Rapp
- Date: 2011-08-24 13:09:49 UTC
- mto: This revision was merged to the branch mainline in revision 249.
- Revision ID: sirver@gmx.de-20110824130949-vv0yn4nq3hogq10i
Model.save() should always take args and kwargs and pass them one. Whitespace fixes. Removed dependency on Site being configured on initial run of syncdb
- files added:
- news/tests.py
- pybb/migrations
- files removed:
- media/js/highcharts_219
- media/js/highcharts_219/adapters
- media/js/highcharts_219/modules
- media/js/highcharts_219/themes
- pybb/migrations
- wlms
- wlms/db
- wlms/test
- files modified:
- README.txt
added
removed
wlms/test/test.py
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
5
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
10
11
import logging
12
logging.basicConfig(level=logging.CRITICAL)
13
14
# Helper classes {{{
15
class ClientStringTransport(proto_helpers.StringTransport):
16
def __init__(self, ip):
17
self.client = [ip]
18
proto_helpers.StringTransport.__init__(self)
19
20
class _Base(object):
21
def setUp(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) ]
27
self._gptr = None
28
self._gpc = None
29
30
def create_game_pinger(gc, timeout):
31
self._gpfac = GamePingFactory(gc, timeout)
32
p = self._gpfac.buildProtocol(None)
33
tr = ClientStringTransport("GamePinger")
34
p.makeConnection(tr)
35
self._gptr = tr
36
self._gpc = p
37
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
42
43
c.makeConnection(tr)
44
45
def tearDown(self):
46
for idx,c in enumerate(self._cons):
47
p = self._recv(idx)
48
if p:
49
raise RuntimeError("unexpected packet from client %i: %r" % (idx, p))
50
for c in self._cons:
51
c.transport.loseConnection()
52
c.connectionLost(connectionDone)
53
54
def _recv(self, client):
55
d = self._trs[client].value()
56
self._trs[client].clear()
57
rv = []
58
while len(d):
59
size = (ord(d[0]) << 8) + ord(d[1])
60
self.assertTrue(size <= len(d))
61
rv.append(d[2:size][:-1].split("\x00"))
62
d = d[size:]
63
self.clock.advance(1e-5)
64
return rv
65
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])
72
return ps[0]
73
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 }}}
79
80
# Sending Basics {{{
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")
90
p1, = self._recv(0)
91
self.assertEqual(p1, ['ERROR', 'GARBAGE_RECEIVED', "INVALID_CMD"])
92
def test_sending_toolittle_arguments_packet(self):
93
self._send(0, "LOGIN", "hi")
94
p1, = self._recv(0)
95
self.assertEqual(p1, ['ERROR', 'LOGIN', "Invalid integer: 'hi'"])
96
97
def test_sendtwopacketsinone(self):
98
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
99
self._recv(0)
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', ''])
104
105
def test_fragmented_packages(self):
106
self._send(0, "LOGIN", 0, "testuser", "build-17", "false")
107
self._recv(0)
108
109
self._cons[0].dataReceived("\x00\x0aCLI")
110
self._cons[0].dataReceived("ENTS\x00\x00\x0a")
111
p1, = self._recv(0)
112
self._cons[0].dataReceived("CLIENTS\x00\x00\x08")
113
p2, = self._recv(0)
114
self.assertEqual(p1, p2)
115
116
self.assertEqual(p1, ['CLIENTS', '1', 'testuser', 'build-17', '', 'UNREGISTERED', ''])
117
# End: Sending Basics }}}
118
# Test Regular Pinging {{{
119
class TestRegularPinging(_Base, unittest.TestCase):
120
def setUp(self):
121
_Base.setUp(self)
122
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
123
self._recv(0)
124
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")
130
131
self.clock.advance(15)
132
p1, = self._recv(0) # Expect a ping packet
133
self.assertEqual(p1, ["PING"])
134
self._send(0, "PONG")
135
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"])
143
144
def test_delay_ping_by_regular_packet(self):
145
self.clock.advance(9.9)
146
self._send(0, "CHAT", "hello there", "")
147
p1, = self._recv(0)
148
self.assertEqual(p1, ["CHAT", 'bert', 'hello there', 'public'])
149
150
self.clock.advance(9.9)
151
self._send(0, "CHAT", "hello there", "")
152
p1, = self._recv(0)
153
self.assertEqual(p1, ["CHAT", 'bert', 'hello there', 'public'])
154
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 }}}
159
160
# Test Login {{{
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'])
168
169
def test_loginanon_unknown_protocol(self):
170
self._send(0, "LOGIN", 10, "testuser", "build-17", "false")
171
p1, = self._recv(0)
172
self.assertEqual(p1, ['ERROR', 'LOGIN', "UNSUPPORTED_PROTOCOL"])
173
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'])
180
181
def test_loginanon_onewasalreadythere(self):
182
# Login client 0
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'])
188
189
# Login client 1
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'])
196
197
# 0 should get an CLIENTS_UPDATE
198
p1, = self._recv(0)
199
self.assertEqual(p1, ['CLIENTS_UPDATE'])
200
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'])
207
208
def test_nonanon_login_onewasalreadythere(self):
209
# Login client 0
210
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
211
self._recv(0)
212
213
# Login client 0
214
self._send(1, "LOGIN", 0, "SirVer", "build-17", 1, "123456")
215
p1, = self._recv(1)
216
217
self.assertEqual(p1, ['ERROR', 'LOGIN', 'ALREADY_LOGGED_IN'])
218
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'])
225
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'])
231
232
# 0 should get an CLIENTS_UPDATE
233
p1, = self._recv(0)
234
self.assertEqual(p1, ['CLIENTS_UPDATE'])
235
236
def test_nonanon_login_incorrect_password(self):
237
self._send(0, "LOGIN", 0, "SirVer", "build-17", 1, "12345")
238
p, = self._recv(0)
239
self.assertEqual(p, ['ERROR', 'LOGIN', 'WRONG_PASSWORD'])
240
# End: Test Login }}}
241
# TestMotd {{{
242
class TestMotd(_Base, unittest.TestCase):
243
def setUp(self):
244
_Base.setUp(self)
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")
248
self._recv(0)
249
self._recv(1)
250
self._recv(2)
251
252
def test_setting_motd(self):
253
self._send(2, "MOTD", "Schnulz is cool!")
254
255
p1, = self._mult_receive(range(3))
256
self.assertEqual(p1, ["CHAT", "", "Schnulz is cool!", "system"])
257
258
def test_login_with_motd_set(self):
259
self._send(2, "MOTD", "Schnulz is cool!")
260
p1, = self._mult_receive(range(3))
261
262
self._send(4, "LOGIN", 0, "fasel", "build-18", "false")
263
p1,p2,p3,p4 = self._recv(4)
264
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"])
269
270
p1, = self._mult_receive(range(3))
271
self.assertEqual(p1, ["CLIENTS_UPDATE"])
272
273
274
def test_setting_motd_forbidden(self):
275
self._send(1, "MOTD", "Schnulz is cool!")
276
self._send(0, "MOTD", "Schnulz is cool!")
277
278
p1, = self._mult_receive([0,1])
279
self.assertEqual(p1, ["ERROR", "MOTD", "DEFICIENT_PERMISSION"])
280
# End: TestMotd }}}
281
# TestAnnouncement {{{
282
class TestAnnouncement(_Base, unittest.TestCase):
283
def setUp(self):
284
_Base.setUp(self)
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")
288
self._recv(0)
289
self._recv(1)
290
self._recv(2)
291
292
def test_setting_announcement(self):
293
self._send(2, "ANNOUNCEMENT", "Schnulz is cool!")
294
295
p1, = self._mult_receive(range(3))
296
self.assertEqual(p1, ["CHAT", "", "Schnulz is cool!", "system"])
297
298
def test_setting_announcement_forbidden(self):
299
self._send(1, "ANNOUNCEMENT", "Schnulz is cool!")
300
self._send(0, "ANNOUNCEMENT", "Schnulz is cool!")
301
302
p1, = self._mult_receive([0,1])
303
self.assertEqual(p1, ["ERROR", "ANNOUNCEMENT", "DEFICIENT_PERMISSION"])
304
# End: TestAnnouncement }}}
305
# Test Relogin {{{
306
class TestRelogin_Anon(_Base, unittest.TestCase):
307
def setUp(self):
308
_Base.setUp(self)
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")
312
self._recv(0)
313
self._recv(2)
314
self._recv(3)
315
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"])
320
321
self._send(0, "PONG")
322
self.assertFalse(self._recv(0))
323
324
p1, = self._recv(1)
325
self.assertEqual(p1, ["ERROR", "RELOGIN", "CONNECTION_STILL_ALIVE"])
326
327
def test_relogin_ping_and_noreply(self):
328
self._send(1, "RELOGIN", 0, "bert", "build-17", "false")
329
p1, = self._recv(0)
330
self.assertEqual(p1, ["PING"])
331
332
self.clock.advance(6)
333
334
# Connection was terminated for old user
335
p1, = self._recv(0)
336
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
337
338
# Relogin accepted
339
p1, = self._recv(1)
340
self.assertEqual(p1, ["RELOGIN"])
341
342
def test_relogin_notloggedin(self):
343
self._send(1, "RELOGIN", 0, "iamnotbert", "build-17", "false")
344
p1, = self._recv(1)
345
346
self.assertEqual(p1, ["ERROR", "RELOGIN", "NOT_LOGGED_IN"])
347
self.assertFalse(self._recv(0))
348
349
def test_relogin_wronginformation_wrong_proto(self):
350
self._send(1, "RELOGIN", 1, "bert", "build-17", "false")
351
p1, = self._recv(1)
352
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
353
self.assertFalse(self._recv(0))
354
355
def test_relogin_wronginformation_wrong_buildid(self):
356
self._send(1, "RELOGIN", 0, "bert", "uild-17", "false")
357
p1, = self._recv(1)
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")
362
p1, = self._recv(1)
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")
367
p1, = self._recv(1)
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")
372
p1, = self._recv(1)
373
self.assertEqual(p1, ["ERROR", "RELOGIN", "WRONG_INFORMATION"])
374
375
self.assertFalse(self._recv(0))
376
def test_relogin_loggedin_allcorrect(self):
377
self._send(1, "RELOGIN", 0, "otto", "build-17", "true", "ottoiscool")
378
p1, = self._recv(2)
379
self.assertEqual(p1, ["PING"])
380
381
self.clock.advance(6)
382
383
# Connection was terminated for old user
384
p1, = self._recv(2)
385
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
386
387
# Relogin accepted
388
p1, = self._recv(1)
389
self.assertEqual(p1, ["RELOGIN"])
390
391
class TestReloginWhileInGame(_Base, unittest.TestCase):
392
def setUp(self):
393
_Base.setUp(self)
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)
399
self._recv(0)
400
self._recv(2)
401
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")
407
408
self.clock.advance(10)
409
p1, = self._recv(2)
410
self._send(2, "PONG")
411
412
# Connection was terminated for old user
413
p1, = self._recv(0)
414
self.assertEqual(p1, ["DISCONNECT", "CLIENT_TIMEOUT"])
415
self._cons[0].connectionLost(connectionDone)
416
self.assertEqual([], self._mult_receive([1,2]))
417
418
self._send(1, "RELOGIN", 0, "bert", "build-17", "false")
419
420
# Relogin accepted
421
p1, = self._recv(1)
422
self.assertEqual(p1, ["RELOGIN"])
423
# p1, = self._recv(2)
424
# self.assertEqual(p1, ["CLIENTS_UPDATE"])
425
426
# Some garbage might have reached zero
427
self._recv(0)
428
429
self._send(1, "CLIENTS")
430
p1, = self._recv(1)
431
self.assertEqual(p1, ["CLIENTS", '2',
432
"bert", "build-17", "my cool game", "UNREGISTERED", "",
433
"otto", "build-17", "my cool game", "REGISTERED", "",
434
])
435
436
437
# End: Test Relogin }}}
438
# Test Chat {{{
439
class TestChat(_Base, unittest.TestCase):
440
def setUp(self):
441
_Base.setUp(self)
442
self._send(0, "LOGIN", 0, "bert", "build-17", "false")
443
self._send(1, "LOGIN", 0, "ernie", "build-17", "false")
444
self._recv(0)
445
self._recv(1)
446
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"])
451
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"])
456
457
def test_private_chat(self):
458
self._send(0, "CHAT", "hello there", "ernie")
459
self.assertEqual([], self._recv(0))
460
p1, = self._recv(1)
461
self.assertEqual(p1, ["CHAT", "bert", "hello there", "private"])
462
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))
466
p1, = self._recv(1)
467
self.assertEqual(p1, ["CHAT", "bert", "hello <rt>there</rt>\nhow<rtdoyoudo", "private"])
468
469
# End: Test Chat }}}
470
#
471
# Test Game Creation/Joining {{{
472
class TestGameCreation(_Base, unittest.TestCase):
473
def setUp(self):
474
_Base.setUp(self)
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")
478
self._recv(0)
479
self._recv(1)
480
self._recv(2)
481
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"])
487
488
self._send(1, "CLIENTS")
489
p, = self._recv(1)
490
491
self.assertEqual(p, ["CLIENTS", "3",
492
"bert", "build-16", "my cool game", "UNREGISTERED", "",
493
"otto", "build-17", "", "REGISTERED", "",
494
"SirVer", "build-18", "", "SUPERUSER", ""
495
])
496
497
data = self._gptr.value(); self._gptr.clear()
498
self.assertEqual(data, NETCMD_METASERVER_PING)
499
self._gpc.dataReceived(NETCMD_METASERVER_PING)
500
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"])
506
507
self._send(1, "CLIENTS")
508
p, = self._recv(1)
509
510
self.assertEqual(p, ["CLIENTS", "3",
511
"bert", "build-16", "my cool game", "UNREGISTERED", "",
512
"otto", "build-17", "", "REGISTERED", "",
513
"SirVer", "build-18", "", "SUPERUSER", ""
514
])
515
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"])
521
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"])
526
527
p1, = self._mult_receive([1,2])
528
self.assertEqual(p1, ["GAMES_UPDATE"])
529
530
self._send(2, "CLIENTS")
531
p, = self._recv(2)
532
533
self.assertEqual(p, ["CLIENTS", "3",
534
"bert", "build-16", "my cool game", "UNREGISTERED", "",
535
"otto", "build-17", "", "REGISTERED", "",
536
"SirVer", "build-18", "", "SUPERUSER", ""
537
])
538
539
self._send(2, "GAMES")
540
p, = self._recv(2)
541
self.assertEqual(p, ["GAMES", "1", "my cool game", "build-16", "false"])
542
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))
546
547
self._send(1, "GAME_OPEN", "my cool game", 12)
548
p1, = self._recv(1)
549
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_EXISTS"])
550
551
data = self._gptr.value(); self._gptr.clear()
552
self.assertEqual(data, NETCMD_METASERVER_PING)
553
self._gpc.dataReceived(NETCMD_METASERVER_PING)
554
self._recv(0)
555
self._mult_receive([1,2])
556
557
self._send(2, "CLIENTS")
558
p, = self._recv(2)
559
560
self.assertEqual(p, ["CLIENTS", "3",
561
"bert", "build-16", "my cool game", "UNREGISTERED", "",
562
"otto", "build-17", "", "REGISTERED", "",
563
"SirVer", "build-18", "", "SUPERUSER", ""
564
])
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))
568
569
data = self._gptr.value(); self._gptr.clear()
570
self.assertEqual(data, NETCMD_METASERVER_PING)
571
self._gpc.dataReceived(NETCMD_METASERVER_PING)
572
self._recv(0)
573
self._mult_receive([1,2])
574
575
self._send(1, "GAME_OPEN", "my cool game", 12)
576
p1, = self._recv(1)
577
self.assertEqual(p1, ["ERROR", "GAME_OPEN", "GAME_EXISTS"])
578
579
self._send(2, "CLIENTS")
580
p, = self._recv(2)
581
582
self.assertEqual(p, ["CLIENTS", "3",
583
"bert", "build-16", "my cool game", "UNREGISTERED", "",
584
"otto", "build-17", "", "REGISTERED", "",
585
"SirVer", "build-18", "", "SUPERUSER", ""
586
])
587
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"])
593
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"])
598
599
p1, = self._mult_receive([1,2])
600
self.assertEqual(p2, ["GAMES_UPDATE"])
601
602
self._send(2, "CLIENTS")
603
p, = self._recv(2)
604
605
self.assertEqual(p, ["CLIENTS", "3",
606
"bert", "build-16", "my cool game", "UNREGISTERED", "",
607
"otto", "build-17", "", "REGISTERED", "",
608
"SirVer", "build-18", "", "SUPERUSER", ""
609
])
610
611
self._send(2, "GAMES")
612
p, = self._recv(2)
613
self.assertEqual(p, ["GAMES", "1", "my cool game", "build-16", "false"])
614
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"])
620
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"])
629
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")
634
635
data = self._gptr.value(); self._gptr.clear()
636
self.assertEqual(data, '')
637
638
self.clock.advance(2)
639
data = self._gptr.value(); self._gptr.clear()
640
self.assertEqual(data, NETCMD_METASERVER_PING)
641
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")
647
648
self.clock.advance(3)
649
650
p1, = self._mult_receive(range(3))
651
self.assertEqual(p1, ["GAMES_UPDATE"])
652
653
self._send(2, "CLIENTS")
654
p, = self._recv(2)
655
656
self.assertEqual(p, ["CLIENTS", "3",
657
"bert", "build-16", "my cool game", "UNREGISTERED", "",
658
"otto", "build-17", "", "REGISTERED", "",
659
"SirVer", "build-18", "", "SUPERUSER", ""
660
])
661
662
self._send(2, "GAMES")
663
p, = self._recv(2)
664
self.assertEqual(p, ["GAMES", "0"])
665
666
def test_join_game(self):
667
self._send(0, "GAME_OPEN", "my cool game", 8)
668
self._recv(0)
669
p1,p2 = self._mult_receive([1,2])
670
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"])
675
676
p1, = self._mult_receive([0,2])
677
self.assertEqual(p1, ["CLIENTS_UPDATE"])
678
679
self._send(2, "CLIENTS")
680
p, = self._recv(2)
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", ""
685
])
686
687
def test_join_full_game(self):
688
self._send(0, "GAME_OPEN", "my cool game", 1)
689
self._recv(0)
690
p1,p2 = self._mult_receive([1,2])
691
692
self._send(1, "GAME_CONNECT", "my cool game")
693
p1, = self._recv(1)
694
self.assertEqual(p1, ["ERROR", "GAME_CONNECT", "GAME_FULL"])
695
696
self._send(2, "CLIENTS")
697
p, = self._recv(2)
698
self.assertEqual(p, ["CLIENTS", "3",
699
"bert", "build-16", "my cool game", "UNREGISTERED", "",
700
"otto", "build-17", "", "REGISTERED", "",
701
"SirVer", "build-18", "", "SUPERUSER", ""
702
])
703
704
def test_non_existing_Game(self):
705
self._send(1, "GAME_CONNECT", "my cool game")
706
p1, = self._recv(1)
707
self.assertEqual(p1, ["ERROR", "GAME_CONNECT", "NO_SUCH_GAME"])
708
709
self._send(2, "CLIENTS")
710
p, = self._recv(2)
711
self.assertEqual(p, ["CLIENTS", "3",
712
"bert", "build-16", "", "UNREGISTERED", "",
713
"otto", "build-17", "", "REGISTERED", "",
714
"SirVer", "build-18", "", "SUPERUSER", ""
715
])
716
# End: Test Game Creation/Joining }}}
717
# Test Game Starting {{{
718
class TestGameStarting(_Base, unittest.TestCase):
719
def setUp(self):
720
_Base.setUp(self)
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")
726
self._recv(0)
727
self._recv(1)
728
self._recv(2)
729
730
def test_start_game_without_being_in_one(self):
731
self._send(2, "GAME_START")
732
p1, = self._recv(2)
733
734
self.assertEqual(p1, ["ERROR", "GARBAGE_RECEIVED", "INVALID_CMD"])
735
736
def test_start_game_not_host(self):
737
self._send(1, "GAME_START")
738
p1, = self._recv(1)
739
740
self.assertEqual(p1, ["ERROR", "GAME_START", "DEFICIENT_PERMISSION"])
741
742
def test_start_game(self):
743
self._send(0, "GAME_START")
744
p1,p2 = self._recv(0)
745
self.assertEqual(p1, ["GAME_START"])
746
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):
753
def setUp(self):
754
_Base.setUp(self)
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")
760
self._recv(0)
761
self._recv(1)
762
self._recv(2)
763
764
def test_leave_game_nothost(self):
765
self._send(1, "GAME_DISCONNECT")
766
p1, = self._recv(1)
767
self.assertEqual(p1, ["CLIENTS_UPDATE"])
768
769
p1, = self._mult_receive([0,2])
770
self.assertEqual(p1, ["CLIENTS_UPDATE"])
771
772
self._send(2, "CLIENTS")
773
p, = self._recv(2)
774
self.assertEqual(p, ["CLIENTS", "3",
775
"bert", "build-16", "my cool game", "UNREGISTERED", "",
776
"otto", "build-17", "", "REGISTERED", "",
777
"SirVer", "build-18", "", "SUPERUSER", ""
778
])
779
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"])
785
786
p1,p2 = self._mult_receive([1,2])
787
self.assertEqual(p1, ["CLIENTS_UPDATE"])
788
self.assertEqual(p2, ["GAMES_UPDATE"])
789
790
self._send(2, "CLIENTS")
791
p, = self._recv(2)
792
self.assertEqual(p, ["CLIENTS", "3",
793
"bert", "build-16", "", "UNREGISTERED", "",
794
"otto", "build-17", "my cool game", "REGISTERED", "",
795
"SirVer", "build-18", "", "SUPERUSER", ""
796
])
797
798
def test_leave_not_in_game(self):
799
self._send(2, "GAME_DISCONNECT")
800
p1,= self._recv(2)
801
self.assertEqual(p1, ["ERROR", "GARBAGE_RECEIVED", "INVALID_CMD"])
802
803
self._send(2, "CLIENTS")
804
p, = self._recv(2)
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", ""
809
])
810
# End: Game Leaving }}}
811
812
813
Loggerhead is a web-based interface for Breezy
Version: 2.0.1