Source

caktus-snake / tic-tac-toe / server / tttserv.py

Full commit
#!/usr/bin/env python
#adapted from Facebook's helloworld.py
#testing from:
#http://groups.google.com/group/python-tornado/browse_thread/thread/867cfb2665ea10a9/fc9a53000f4c104d

import sys
import unittest
import logging
import pprint

import tornado.httpserver
import tornado.httpclient
import tornado.ioloop
import tornado.options
import tornado.web

from tornado.options import define, options

import simplejson as json

import websocket

logging.basicConfig(level=logging.DEBUG)


class EchoWebSocket(websocket.WebSocketHandler):
    def open(self):
        print 'open'
        self.receive_message(self.on_message)

    def on_message(self, message):
        print message
        self.write_message(message)


class TestTicTacToe(unittest.TestCase):
    response = None

    def setUp(self):
        self.client = tornado.httpclient.AsyncHTTPClient()
        #self.client = tornado.httpclient.HTTPClient()

    def get(self, path=''):
        self.client.fetch(
            'http://localhost:%d/%s' % (options.port, path),
            self.handle_request,
        )

    def post(self, path='', body=None):
        self.client.fetch(
            'http://localhost:%d/%s' % (options.port, path),
            self.handle_request,
            method='POST',
            body=body,
        )
        tornado.ioloop.IOLoop.instance().start()

    def handle_request(self, response):
        self.response = response
        tornado.ioloop.IOLoop.instance().stop()

    def test_no_board(self):
        self.post(body='')
        self.assertEqual(self.response.code,500)

    def test_get(self):
        self.get()
        #self.assertEqual(self.response.code,200)

    def test_board(self):
        body = {
            'board' : [
                [0,1,0],
                [0,-1,0],
                [1,0,-1]
            ],
        }
        self.post(body=json.dumps(body))
        self.assertEqual(self.response.code,200)
        new_board = json.loads(self.response.body)
        self.failIf(new_board == body)

def move(board):
    import random
    row = random.choice(range(len(board)))
    col = random.choice(range(len(board[0])))
    new_options = [-1,0,1]
    new_options.remove(board[row][col])
    board[row][col] = random.choice( new_options )
    return board

class MainHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def post(self):
        board_json = self.request.body
        if not board_json:
            raise tornado.web.HTTPError(500, "Board not found.")
        board = json.loads(board_json)['board']
        new_board = move(board)
        self.write({'board' : new_board})
        self.flush()
        self.finish()

    def get(self):
        pass

define("port", default=8888, help="run on the given port", type=int)
define("test", type=bool, help="run unit tests for server")
def main():
    tornado.options.parse_command_line()

    application = tornado.web.Application([
        (r"/", MainHandler),
        (r"/websocket", EchoWebSocket),
    ])

    http_server = tornado.httpserver.HTTPServer(application)
    http_server.listen(options.port)

    if options.test:
        sys.argv.remove('--test')
        unittest.main()
    else:
        tornado.ioloop.IOLoop.instance().start()
if __name__ == "__main__":
    main()