> def __init__(self, *args, **kwargs):
> tornado.web.RequestHandler.__init__(self, *args, **kwargs)
> self.log = self.application.log
> self.db = self.application.db
> class ProductHandler(BaseHandler):
> @tornado.web.removeslash
> def put(self, id = None, *args, **kwargs):
> try:
> self.log.info("Handling PUT request")
> if not id:
> raise Exception('Object Id Required')
> id = { '_id' : id }
> new_values = dict()
> name = self.get_argument('name', None)
> description = self.get_argument('description', None)
> if name:
> new_values['name'] = name
> if description:
> new_values['description'] = description
> self.db.products.update(id, new_values, safe = True)
> except:
> self.log.error("".join(tb.format_exception(*sys.exc_info())))
> raise
> class Application(tornado.web.Application):
> def __init__(self, config_path, test = False, *args, **kwargs):
> handlers = [
> (r"/product/?(.*)", ProductHandler),
> ]
> settings = dict(debug=True)
> tornado.web.Application.__init__(self, handlers, **settings)
> self.log = logging.getLogger(__name__)
> self.config = ConfigParser()
> self.config.read(config_path)
> self.mongo_connection = Connection(
> host = self.config.get('mongo','host'),
> port = self.config.getint('mongo','port'),
> )
> if test:
> db_name = self.config.get('test', 'mongo.db')
> else:
> db_name = self.config.get('mongo', 'db')
> self.log.debug("Using db: %s" % db_name)
> self.db = self.mongo_connection[db_name]
> class Test(AsyncHTTPTestCase):
> def setUp(self):
> clean_db(APP_CONFIG_PATH)
> AsyncHTTPTestCase.setUp(self)
> def tearDown(self):
> clean_db(APP_CONFIG_PATH)
> def get_app(self):
> return Application(config_path = APP_CONFIG_PATH, test = True)
> class CrudTest(Test):
> def setUp(self):
> GonzoTest.setUp(self)
> self.URL = '/product'
> self.log = logging.getLogger(__name__)
> def test_product(self):
> # ...
> expected_product = dict()
> expected_product['description'] = 'Some Other Description'
> expected_product['name'] = 'Some Other Product Name'
> self.log.debug("PUT request with expected field updates: %s" %
> str(expected_product))
> self.http_client.fetch(
> "/".join([self.get_url(self.URL), new_product_id]),
> self.stop,
> method = "PUT",
> body = urllib.urlencode(expected_product),
> )
> expected_product['id'] = new_product_id
> response = self.wait()
> self.assertEqual(response.code, 200)
> self.http_client.fetch(
> "/".join([self.get_url(self.URL), new_product_id]),
> self.stop,
> )
> response = self.wait()
> self.assertEqual(response.code, 200)
> actual_product = json.loads(response.body)[0]
> comparison_count = 0
> for actual_product_key in actual_product:
> for expected_product_key in expected_product:
> if expected_product_key != actual_product_key:
> continue
> self.assertEqual(
> expected_product[expected_product_key],
> actual_product[actual_product_key]
> )
> comparison_count += 1
> self.assertEqual(comparison_count, len(expected_product.keys()))
> self.assertEqual(actual_product.keys(), expected_product.keys())
...
