Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

add some basic unit tests

  • Loading branch information...
commit ecffaeb37d32d3307a32dffcf00325e7b7faf3a3 1 parent c69ea50
@eklitzke authored
Showing with 110 additions and 1 deletion.
  1. +2 −0  runtests
  2. +10 −1 tornado_dns/lookup.py
  3. +98 −0 tornado_dns/tests.py
View
2  runtests
@@ -0,0 +1,2 @@
+#!/bin/bash
+python -m qa -m tornado_dns.tests -c process -w 3 -v
View
11 tornado_dns/lookup.py
@@ -1,4 +1,5 @@
import time
+import errno
import socket
import tornado.ioloop
@@ -47,7 +48,15 @@ def lookup(name, callback, errback=None, timeout=None, server=None):
query = DNSPacket.create_a_question(name)
def read_response(fd, events):
- data, addr = sock.recvfrom(1500)
+ try:
+ data, addr = sock.recvfrom(1500)
+ except socket.error, e:
+ # tornado lied to us??
+ if e.errno == errno.EAGAIN:
+ io_loop.remove_handler(fd)
+ io_loop.add_handler(fd, read_response, io_loop.READ)
+ return
+ raise
response = DNSPacket.from_wire(data)
callback(response.get_answer_names())
io_loop.remove_handler(fd)
View
98 tornado_dns/tests.py
@@ -0,0 +1,98 @@
+import qa
+from contextlib import contextmanager
+from functools import wraps
+import tornado.ioloop
+import tornado_dns
+
+io_loop = tornado.ioloop.IOLoop.instance()
+
+class Trit(object):
+
+ OFF = 0
+ ON = 1
+ ERR = 2
+
+ def __init__(self):
+ self.val = self.OFF
+ self.val = False
+
+ def on(self):
+ self.val = True
+
+ def off(self):
+ self.val = False
+
+ def check(self, expected=None):
+ if expected is None:
+ expected = self.ON
+ if self.val != expected:
+ raise AssertionError("Expected %s, got %s" % (self.read_val(expected), self.read_val()))
+
+ def read_val(self, val=None):
+ if val is None:
+ val = self.val
+ if val == self.OFF:
+ return 'OFF'
+ elif val == self.ON:
+ return 'ON'
+ elif val == self.ERR:
+ return 'ERR'
+ else:
+ raise ValueError('val = %r' % (val,))
+
+@contextmanager
+def test_context(ctx):
+ ctx.trit_final = Trit.ON
+ ctx.trit = Trit()
+ yield
+ ctx.trit.check(ctx.trit_final)
+
+def callback(func):
+ @wraps(func)
+ def inner(*args, **kwargs):
+ ret = func(*args, **kwargs)
+ io_loop.stop()
+ return ret
+ return inner
+
+def testcase(*extra_requires):
+ def outer(func):
+ @qa.testcase(requires=[test_context] + list(extra_requires))
+ @wraps(func)
+ def inner(ctx):
+ def run():
+ return func(ctx)
+ io_loop.add_callback(run)
+ io_loop.start()
+ return inner
+ return outer
+
+@testcase()
+def test_basic_a_record(ctx):
+ @callback
+ def success(records):
+ ctx.trit.on()
+ assert records['iomonad.com'] == '173.230.147.249'
+ tornado_dns.lookup('iomonad.com', success)
+
+@testcase()
+def test_simple_cname(ctx):
+ @callback
+ def success(records):
+ ctx.trit.on()
+ assert records['cname1.iomonad.com'] == '173.230.147.249'
+ assert records['cname1.iomonad.com'] == records['iomonad.com']
+ tornado_dns.lookup('cname1.iomonad.com', success)
+
+@testcase()
+def test_complex_cname(ctx):
+ @callback
+ def success(records):
+ ctx.trit.on()
+ assert records['cname2.iomonad.com'] == '173.230.147.249'
+ assert records['cname2.iomonad.com'] == records['cname1.iomonad.com']
+ assert records['cname1.iomonad.com'] == records['iomonad.com']
+ tornado_dns.lookup('cname2.iomonad.com', success)
+
+if __name__ == '__main__':
+ qa.main()
Please sign in to comment.
Something went wrong with that request. Please try again.