Skip to content

Instantly share code, notes, and snippets.

@Sheeo
Created February 6, 2016 18:47
Show Gist options
  • Save Sheeo/1ef256c0e69296d473ad to your computer and use it in GitHub Desktop.
Save Sheeo/1ef256c0e69296d473ad to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
"""
Usage:
server.py [--nodb | --db TYPE]
Options:
--nodb Don't use a database (Use a mock.Mock). Caution: Will break things.
--db TYPE Use TYPE database driver [default: QMYSQL]
"""
import asyncio
import logging
from server.protocol import QDataStreamProtocol
@asyncio.coroutine
def connect_client(server):
return QDataStreamProtocol(*(yield from asyncio.open_connection(*server)))
@asyncio.coroutine
def get_session(proto):
proto.send_message({'command': 'ask_session', 'user_agent': 'faf-client', 'version': '0.11.46'})
yield from proto.drain()
msg = yield from proto.read_message()
return msg['session']
@asyncio.coroutine
def perform_login(proto, credentials):
login, pw = credentials
proto.send_message({'command': 'hello',
'version': '1.0.0-dev',
'user_agent': 'faf-client',
'login': login,
'password': pw,
'unique_id': 'some_id'})
yield from proto.drain()
@asyncio.coroutine
def read_until(proto, pred):
while True:
msg = yield from proto.read_message()
try:
if pred(msg):
return msg
except (KeyError, ValueError):
logging.getLogger().info("read_until predicate raised during message: {}".format(msg))
pass
async def connect_and_sign_in(credentials, lobby_server):
proto = await connect_client(lobby_server)
session = await get_session(proto)
await perform_login(proto, credentials)
hello = await proto.read_message()
player_id = hello['id']
return player_id, session, proto
test_logins = [
]
if __name__ == '__main__':
logger = logging.getLogger(__name__)
try:
loop = asyncio.get_event_loop()
async def derp(login, pw):
proto = await connect_client(("test.faforever.com", 8001))
session = await get_session(proto)
await perform_login(proto, (login, pw))
hello = await proto.read_message()
player_id = hello['id']
proto.send_message({'command': 'InitiateTest', 'target': 'connectivity', 'args': [6112]})
proto.writer.close()
return player_id, session, proto
async def derp2():
for login, pw in test_logins:
asyncio.ensure_future(derp(login, pw))
await derp(*test_logins[0])
for i in range(1):
print("Connecting {}".format(i))
loop.run_until_complete(derp2())
loop.close()
except Exception as ex:
logger.exception("Failure booting server {}".format(ex))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment