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
import tornado.websocket
from tornado.options import define, options

try:
    import json
except ImportError:
    import simplejson as json


class EchoWebSocket(tornado.websocket.WebSocketHandler):
    def open(self):
        logging.info('Connection established from %s' % self.request.remote_ip)

    def on_message(self, message):
        logging.info('Message received: %s' % message)
        board = json.loads(message)['board']
        new_board = {'board': json.dumps(move(board))}
        logging.info('Message sent: %s' % new_board)
        self.write_message(new_board)


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 rows():
    rows = []
    for i in range(3):
        rows.append([(i,j) for j in range(3)])
    return rows
def columns():
    cols = []
    for i in range(3):
        cols.append([(j,i) for j in range(3)])
    return cols
def diagnols():
    return [[(0,0),(1,1),(2,2)],[(0,2),(1,1),(2,0)]]

def winning_move(board, player):
    wins = rows() + columns() + diagnols()
    for i in wins:
        squares = [board[r][c] for r,c in i]
        for r,c in i:
            if board[r][c] == 0 and squares.count(player) == 2:
                return (r,c)
    return None

def move(board):
    #can win?
    if winning_move(board, -1):
        r,c = winning_move(board, -1)
        board[r][c] = -1
        return board
    #can block a win?
    if winning_move(board, 1):
        r,c = winning_move(board, 1)
        board[r][c] = -1
        return board
    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()
    logging.info('Server Online')

    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') #don't mess with unittest
        unittest.main()
    else:
        tornado.ioloop.IOLoop.instance().start()
if __name__ == "__main__":
    main()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.