Split TestTablesApi into multiple tests
Change-Id: I77c949ac1a9fb846dd2c9e690ed85b4d2a8a07f0
This commit is contained in:
parent
e05c9023b5
commit
57102b5457
|
@ -43,7 +43,7 @@ class AbstractApiTest(unittest.TestCase):
|
||||||
@classmethod
|
@classmethod
|
||||||
def setUpClass(cls):
|
def setUpClass(cls):
|
||||||
cls.server = subprocess.Popen(
|
cls.server = subprocess.Popen(
|
||||||
[cls.API_SERVER_PATH] + cls.API_SERVER_ARGS)
|
[cls.API_SERVER_PATH] + cls.API_SERVER_ARGS)
|
||||||
hconn = httplib.HTTPConnection(cls.API_SERVER_ADDR,
|
hconn = httplib.HTTPConnection(cls.API_SERVER_ADDR,
|
||||||
cls.API_SERVER_PORT)
|
cls.API_SERVER_PORT)
|
||||||
starttm = time.time() * 1000
|
starttm = time.time() * 1000
|
||||||
|
@ -70,7 +70,7 @@ class AbstractApiTest(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.hconn = httplib.HTTPConnection(self.API_SERVER_ADDR,
|
self.hconn = httplib.HTTPConnection(self.API_SERVER_ADDR,
|
||||||
self.API_SERVER_PORT)
|
self.API_SERVER_PORT)
|
||||||
self.hconn.connect()
|
self.hconn.connect()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
|
@ -80,14 +80,14 @@ class AbstractApiTest(unittest.TestCase):
|
||||||
content_type='text/plain'):
|
content_type='text/plain'):
|
||||||
body = response.read()
|
body = response.read()
|
||||||
self.assertTrue(response.status == status,
|
self.assertTrue(response.status == status,
|
||||||
'%s response status == %s' %(description, status))
|
'%s response status == %s' % (description, status))
|
||||||
|
|
||||||
if content_type is not None:
|
if content_type is not None:
|
||||||
if ';' in content_type:
|
if ';' in content_type:
|
||||||
self.assertTrue(
|
self.assertTrue(
|
||||||
response.getheader('content-type') == content_type,
|
response.getheader('content-type') == content_type,
|
||||||
"'%s response Content-Type (with params) is '%s'"
|
"'%s response Content-Type (with params) is '%s'"
|
||||||
% (description, content_type))
|
% (description, content_type))
|
||||||
else:
|
else:
|
||||||
ct_start = response.getheader('content-type').split(';', 1)[0]
|
ct_start = response.getheader('content-type').split(';', 1)[0]
|
||||||
self.assertTrue(ct_start == content_type,
|
self.assertTrue(ct_start == content_type,
|
||||||
|
@ -97,7 +97,7 @@ class AbstractApiTest(unittest.TestCase):
|
||||||
|
|
||||||
def check_json_response(self, response, description, status=httplib.OK):
|
def check_json_response(self, response, description, status=httplib.OK):
|
||||||
raw_body = self.check_response(
|
raw_body = self.check_response(
|
||||||
response, description, status, 'application/json')
|
response, description, status, 'application/json')
|
||||||
body = json.loads(raw_body)
|
body = json.loads(raw_body)
|
||||||
return body
|
return body
|
||||||
|
|
||||||
|
@ -106,9 +106,7 @@ class TestTablesApi(AbstractApiTest):
|
||||||
API_SERVER_PATH = os.path.join(SRC_PATH, 'server', 'server.py')
|
API_SERVER_PATH = os.path.join(SRC_PATH, 'server', 'server.py')
|
||||||
STATIC_TABLES = ['ad-groups']
|
STATIC_TABLES = ['ad-groups']
|
||||||
|
|
||||||
def test_tables(self):
|
def test_tables_get(self):
|
||||||
"""Test table list API method."""
|
|
||||||
# List (empty)
|
|
||||||
self.hconn.request('GET', '/tables')
|
self.hconn.request('GET', '/tables')
|
||||||
r = self.hconn.getresponse()
|
r = self.hconn.getresponse()
|
||||||
body = self.check_json_response(r, 'List tables (empty)')
|
body = self.check_json_response(r, 'List tables (empty)')
|
||||||
|
@ -116,93 +114,147 @@ class TestTablesApi(AbstractApiTest):
|
||||||
self.assertTrue(len(body) == len(self.STATIC_TABLES),
|
self.assertTrue(len(body) == len(self.STATIC_TABLES),
|
||||||
'List tables (empty) contains default tables only')
|
'List tables (empty) contains default tables only')
|
||||||
|
|
||||||
# Create
|
def test_table_create(self):
|
||||||
ids = []
|
try:
|
||||||
for table in ['table1', 'table2', 'table3']:
|
ids = []
|
||||||
|
for table in ['table1']:
|
||||||
|
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Create table',
|
||||||
|
status=httplib.CREATED)
|
||||||
|
self.assertIsInstance(body, dict,
|
||||||
|
'Create table returns a dict')
|
||||||
|
self.assertIsNotNone(body['table1'])
|
||||||
|
self.assertIsNotNone(body['id'])
|
||||||
|
ids.append(body['id'])
|
||||||
|
|
||||||
|
self.hconn.request('GET', '/tables')
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'List tables')
|
||||||
|
self.assertIsInstance(body, list,
|
||||||
|
'List tables returns a list')
|
||||||
|
self.assertTrue(
|
||||||
|
len(body) == 1 + len(self.STATIC_TABLES),
|
||||||
|
'List contains proper number of results')
|
||||||
|
finally:
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % ids[0])
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
|
||||||
|
def test_create_named(self):
|
||||||
|
id = None
|
||||||
|
try:
|
||||||
|
table = 'table1'
|
||||||
|
self.hconn.request('PUT', '/tables/foo_id',
|
||||||
|
'{"%s": "foo"}' % table)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r,
|
||||||
|
'Create named table',
|
||||||
|
status=httplib.CREATED)
|
||||||
|
id = body['id']
|
||||||
|
self.assertIsInstance(body, dict, 'Create table returns a dict')
|
||||||
|
self.assertTrue(body['id'] ==
|
||||||
|
'foo_id', 'Created table has specified ID')
|
||||||
|
finally:
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
|
||||||
|
def test_read(self):
|
||||||
|
id = None
|
||||||
|
try:
|
||||||
|
table = 'table1'
|
||||||
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
||||||
r = self.hconn.getresponse()
|
r = self.hconn.getresponse()
|
||||||
body = self.check_json_response(r, 'Create table',
|
body = self.check_json_response(r, 'Create named table',
|
||||||
status=httplib.CREATED)
|
status=httplib.CREATED)
|
||||||
self.assertIsInstance(body, dict, 'Create table returns a dict')
|
id = body['id']
|
||||||
#TODO: validate object
|
self.hconn.request('GET', '/tables/%s' % id)
|
||||||
#TODO: validate location header
|
r = self.hconn.getresponse()
|
||||||
ids.append(body['id'])
|
body = self.check_json_response(r, 'Read table')
|
||||||
|
self.assertIsInstance(body, dict, 'Read table returns a dict')
|
||||||
|
self.assertEqual(body['id'], id, 'Read expected table instance')
|
||||||
|
self.assertTrue('table1' in body,
|
||||||
|
'Read expected table instance data')
|
||||||
|
|
||||||
self.hconn.request('GET', '/tables')
|
finally:
|
||||||
r = self.hconn.getresponse()
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
body = self.check_json_response(r, 'List tables')
|
r = self.hconn.getresponse()
|
||||||
self.assertIsInstance(body, list, 'List tables returns a list')
|
|
||||||
self.assertTrue(len(body) == 3 + len(self.STATIC_TABLES),
|
|
||||||
'List contains proper number of results')
|
|
||||||
|
|
||||||
# Create Named
|
def test_read_invalid(self):
|
||||||
self.hconn.request('PUT', '/tables/foo_id', '{"%s": "foo"}' % table)
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Create named table',
|
|
||||||
status=httplib.CREATED)
|
|
||||||
self.assertIsInstance(body, dict, 'Create table returns a dict')
|
|
||||||
self.assertTrue(body['id'] == 'foo_id',
|
|
||||||
'Created table has specified ID')
|
|
||||||
|
|
||||||
# List
|
|
||||||
self.hconn.request('GET', '/tables')
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'List tables')
|
|
||||||
self.assertIsInstance(body, list, 'List tables returns a list')
|
|
||||||
self.assertTrue(len(body) == 4 + len(self.STATIC_TABLES),
|
|
||||||
'List contains proper # results (after create)')
|
|
||||||
|
|
||||||
# Read
|
|
||||||
self.hconn.request('GET', '/tables/%s' % ids[0])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Read table')
|
|
||||||
self.assertIsInstance(body, dict, 'Read table returns a dict')
|
|
||||||
self.assertEqual(body['id'], ids[0], 'Read expected table instance')
|
|
||||||
self.assertTrue('table1' in body, 'Read expected table instance data')
|
|
||||||
|
|
||||||
# Read Invalid
|
|
||||||
self.hconn.request('GET', '/tables/%s' % uuid.uuid4())
|
self.hconn.request('GET', '/tables/%s' % uuid.uuid4())
|
||||||
r = self.hconn.getresponse()
|
r = self.hconn.getresponse()
|
||||||
body = self.check_json_response(r, 'Read missing table',
|
body = self.check_json_response(r, 'Read missing table',
|
||||||
status=httplib.NOT_FOUND)
|
|
||||||
|
|
||||||
# Replace
|
|
||||||
new = json.loads('{"id": "%s", "table4": "bar"}' % ids[0])
|
|
||||||
self.hconn.request('PUT', '/tables/%s' % ids[0], json.dumps(new))
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Replace table')
|
|
||||||
self.assertEqual(body, new, 'Replaced table returns new data')
|
|
||||||
self.hconn.request('GET', '/tables/%s' % ids[0])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Read replaced table')
|
|
||||||
self.assertEqual(body, new, 'GET replaced table returns new data')
|
|
||||||
|
|
||||||
# Update
|
|
||||||
self.hconn.request('GET', '/tables/%s' % ids[1])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
old_body = self.check_json_response(r, 'Read old table')
|
|
||||||
new = json.loads('{"newkey": "baz"}')
|
|
||||||
self.hconn.request('PATCH', '/tables/%s' % ids[1], json.dumps(new))
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Update table')
|
|
||||||
expected = old_body.copy()
|
|
||||||
expected.update(new)
|
|
||||||
self.assertEqual(body, expected, 'Updated table returns new data')
|
|
||||||
self.hconn.request('GET', '/tables/%s' % ids[1])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Read updated table')
|
|
||||||
self.assertEqual(body, expected, 'GET replaced table returns new data')
|
|
||||||
|
|
||||||
# Delete
|
|
||||||
self.hconn.request('DELETE', '/tables/%s' % ids[1])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Delete table')
|
|
||||||
|
|
||||||
self.hconn.request('DELETE', '/tables/%s' % ids[1])
|
|
||||||
r = self.hconn.getresponse()
|
|
||||||
body = self.check_json_response(r, 'Delete missing table',
|
|
||||||
status=httplib.NOT_FOUND)
|
status=httplib.NOT_FOUND)
|
||||||
#TODO: validate objects
|
|
||||||
|
def test_replace(self):
|
||||||
|
id = None
|
||||||
|
try:
|
||||||
|
table = 'table1'
|
||||||
|
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Create named table',
|
||||||
|
status=httplib.CREATED)
|
||||||
|
id = body['id']
|
||||||
|
new = json.loads('{"id": "%s", "table1": "bar"}' % id)
|
||||||
|
self.hconn.request('PUT', '/tables/%s' % id, json.dumps(new))
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Replace table')
|
||||||
|
self.assertEqual(body, new, 'Replaced table returns new data')
|
||||||
|
self.hconn.request('GET', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Read replaced table')
|
||||||
|
self.assertEqual(body, new, 'GET replaced table returns new data')
|
||||||
|
finally:
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
|
||||||
|
def test_update(self):
|
||||||
|
id = None
|
||||||
|
try:
|
||||||
|
table = 'table1'
|
||||||
|
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Create named table',
|
||||||
|
status=httplib.CREATED)
|
||||||
|
id = body['id']
|
||||||
|
self.hconn.request('GET', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
old_body = self.check_json_response(r, 'Read old table')
|
||||||
|
new = json.loads('{"newkey": "baz"}')
|
||||||
|
self.hconn.request('PATCH', '/tables/%s' % id, json.dumps(new))
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Update table')
|
||||||
|
expected = old_body.copy()
|
||||||
|
expected.update(new)
|
||||||
|
self.assertEqual(body, expected, 'Updated table returns new data')
|
||||||
|
self.hconn.request('GET', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Read updated table')
|
||||||
|
self.assertEqual(body, expected,
|
||||||
|
'GET replaced table returns new data')
|
||||||
|
finally:
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
|
||||||
|
def test_delete(self):
|
||||||
|
id = None
|
||||||
|
try:
|
||||||
|
table = 'table1'
|
||||||
|
self.hconn.request('POST', '/tables', '{"%s": "foo"}' % table)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Create named table',
|
||||||
|
status=httplib.CREATED)
|
||||||
|
id = body['id']
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Delete table')
|
||||||
|
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
body = self.check_json_response(r, 'Delete missing table',
|
||||||
|
status=httplib.NOT_FOUND)
|
||||||
|
finally:
|
||||||
|
self.hconn.request('DELETE', '/tables/%s' % id)
|
||||||
|
r = self.hconn.getresponse()
|
||||||
|
|
||||||
|
|
||||||
class TestPolicyApi(AbstractApiTest):
|
class TestPolicyApi(AbstractApiTest):
|
||||||
|
|
|
@ -23,10 +23,15 @@ ROOTDIR=`dirname $SCRIPTDIR`
|
||||||
|
|
||||||
PYSRCDIR=$ROOTDIR/src
|
PYSRCDIR=$ROOTDIR/src
|
||||||
THIRDPARTYDIR=$ROOTDIR/thirdparty
|
THIRDPARTYDIR=$ROOTDIR/thirdparty
|
||||||
|
TESTDIR=$ROOTDIR/tests
|
||||||
export PYTHONPATH=$SRCDIR:$THIRDPARTYDIR
|
export PYTHONPATH=$SRCDIR:$THIRDPARTYDIR
|
||||||
|
|
||||||
# Use nosetests to find all unitests in tree
|
# Use nosetests to find all unitests in tree
|
||||||
ARGS="$@"
|
ARGS="$@"
|
||||||
cd $ROOTDIR
|
cd $ROOTDIR
|
||||||
nosetests $ARGS
|
nosetests $ARGS
|
||||||
|
|
||||||
|
cd $ROOTDIR
|
||||||
|
export PYTHONPATH=$TESTDIR/functional:$PYTHONPATH
|
||||||
|
python -m unittest test_api
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue