Source

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

#!/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

logging.basicConfig(level=logging.DEBUG)

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),
    ])

    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()