Skip to content
This repository has been archived by the owner on Jan 19, 2022. It is now read-only.

Commit

Permalink
wrote the -r flag to search.
Browse files Browse the repository at this point in the history
  • Loading branch information
uberj committed Nov 8, 2012
1 parent 765efa4 commit 9cb72a8
Show file tree
Hide file tree
Showing 4 changed files with 113 additions and 1 deletion.
3 changes: 2 additions & 1 deletion Makefile
@@ -1,5 +1,6 @@
test: tests:
python tests/cli_tests.py python tests/cli_tests.py
python tests/search_tests.py


inspect: inspect:
./bin/invdns search -q "testfqdn" | awk '{ print "./bin/invdns " $5 " delete --pk " $1}' ./bin/invdns search -q "testfqdn" | awk '{ print "./bin/invdns " $5 " delete --pk " $1}'
Expand Down
6 changes: 6 additions & 0 deletions invdns/dispatch.py
Expand Up @@ -173,6 +173,10 @@ def build_dns_parsers(base_parser):
"Specify the range using: <ip-start>,<ip-end> format " "Specify the range using: <ip-start>,<ip-end> format "
"(no spaces", default=None, required=False) "(no spaces", default=None, required=False)


search.add_argument('--display-integers', dest='d_integers',
help="Return integers when showing free ip ranges.",
action='store_true', default=False, required=False)

# Build all the records # Build all the records


for dispatch in registrar.dns_dispatches: for dispatch in registrar.dns_dispatches:
Expand Down Expand Up @@ -205,6 +209,8 @@ def irange(self, nas):
headers = {'content-type': 'application/json'} headers = {'content-type': 'application/json'}
start, end = nas.irange.split(',') start, end = nas.irange.split(',')
search = {'start': start, 'end': end} search = {'start': start, 'end': end}
if nas.d_integers:
search['format'] = 'integers'
resp = requests.get(url, params=search, headers=headers, auth=auth) resp = requests.get(url, params=search, headers=headers, auth=auth)
if resp.status_code == 500: if resp.status_code == 500:
resp_list = [_("CLIENT ERROR! (Please email this output to " resp_list = [_("CLIENT ERROR! (Please email this output to "
Expand Down
2 changes: 2 additions & 0 deletions tests/all.py
@@ -0,0 +1,2 @@
from search_tests import *
from cli_tests import *
103 changes: 103 additions & 0 deletions tests/search_tests.py
@@ -0,0 +1,103 @@
import pdb
import subprocess, shlex
import unittest
import sys
sys.path.insert(0, "/home/juber/repositories/inv-tool")
import simplejson as json
from gettext import gettext as _

from invdns.dispatch import registrar
from invdns.dispatch import DNSDispatch

EXEC = "./bin/invdns --json"

def test_method_to_params(test_case):
if not test_case:
return ''
elif not test_case[0]:
return test_case[1]
else:
return "--{0} {1}".format(*test_case)

def call_to_json(command_str):
"""Given a string, this function will shell out, execute the command
and parse the json returned by that command"""
p = subprocess.Popen(shlex.split(command_str),
stderr=subprocess.PIPE, stdout=subprocess.PIPE)
stdout, stderr = p.communicate()
if stderr:
return None, stderr, p.returncode

stdout = stdout.replace('u\'', '"').replace('\'', '"').strip('\n')
try:
return json.loads(stdout, 'unicode'), None, p.returncode
except json.decoder.JSONDecodeError, e:
return (None,
"Ret was: {0}. Got error: {1}".format(stdout, str(e)),
p.returncode)

def run_tests():

def build_testcase(commands):
"""The first command is used to create an object. Using the pk returned
from the object's creation we look up the object and use the second
command to update the object. The object is then deleted and finally
the object is looked up again to ensure a 404."""
class _TestCase(unittest.TestCase):
pass

def place_holder(self):
# Search
expected_status, command = commands[0]
ret, errors, rc = call_to_json(command)

if errors:
self.fail(errors)

self.assertEqual(0, rc)

self.assertTrue('http_status' in ret)
self.assertEqual(ret['http_status'], expected_status)

# Range
expected_status, command = commands[1]
ret, errors, rc = call_to_json(command)

if errors:
self.fail(errors)

self.assertEqual(0, rc)

self.assertTrue('http_status' in ret)
self.assertEqual(ret['http_status'], expected_status)


test_name = "test_{0}".format('search')
place_holder.__name__ = test_name
setattr(_TestCase, test_name, place_holder)
return _TestCase

def build_testcases():
commands = []
command = [EXEC, 'search', ' -q "foopy32"']
commands.append((200, ' '.join(command)))

command = [EXEC, 'search', ' -r 10.0.0.0,10.0.20.3']
commands.append((200, ' '.join(command)))

return build_testcase(commands)

ts = unittest.TestSuite()
test_cases = [build_testcases()]
return test_cases

if __name__ == "__main__":
tcs = run_tests()
loader = unittest.TestLoader()
suite = unittest.TestSuite()
for test_class in tcs:
tests = loader.loadTestsFromTestCase(test_class)
suite.addTests(tests)

unittest.TextTestRunner(verbosity=2).run(suite)

0 comments on commit 9cb72a8

Please sign in to comment.