querying.py 1.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142
  1. from lettuce import *
  2. import subprocess
  3. import re
  4. #
  5. # define a class to easily access different parts
  6. # We may consider using our full library for this, but for now
  7. # simply store several parts of the response as text values in
  8. # this structure
  9. #
  10. # this will set 'rcode' as the result code, we 'define' one additional
  11. # rcode, "NO_ANSWER", if the dig process returned an error code itself
  12. # we will extend as necessary
  13. class QueryResult:
  14. def __init__(self, name, qtype = None, qclass = None, port = 47806):
  15. args = [ 'dig', '@localhost', '-p', str(port) ]
  16. if qtype is not None:
  17. args.append('-t')
  18. args.append(str(qtype))
  19. if qclass is not None:
  20. args.append('-c')
  21. args.append(str(qclass))
  22. args.append(name)
  23. dig_process = subprocess.Popen(args, 1, None, None, subprocess.PIPE,
  24. None)
  25. result = dig_process.wait()
  26. if result != 0:
  27. self.rcode = "NO_ANSWER"
  28. else:
  29. rcode_re = re.compile("status: ([A-Z]+)")
  30. self.rcode = None
  31. for out in dig_process.stdout:
  32. rcode_match = rcode_re.search(out)
  33. if rcode_match is not None:
  34. self.rcode = rcode_match.group(1)
  35. @step(u'A query for ([\w.]+) should have rcode ([\w.]+)')
  36. def query(step, query_name, rcode):
  37. query_result = QueryResult(query_name)
  38. assert query_result.rcode == rcode, "Got " + query_result.rcode