Add/refactor functional API tests

More functional tests were added to increase coverage, and
some existing test cases were fixed as they apparently were
not formatted properly.

Change-Id: I3c8a4826c4c52c58978817c0aca8ec6da136185f
This commit is contained in:
Paul Van Eck
2016-03-31 18:08:50 -07:00
parent c78ce638e3
commit c38341c0ec
4 changed files with 234 additions and 106 deletions

View File

@@ -111,6 +111,35 @@ class FunctionalTest(base.BaseTestCase):
conn.close()
raise
def delete(self, url, headers=None, extra_environ=None,
status=None, expect_errors=False, **params):
"""Send HTTP DELETE request.
:param url: url path to target service
:param headers: a dictionary of extra headers to send
:param extra_environ: a dictionary of environmental variables that
should be added to the request
:param status: integer or string of the HTTP status code you expect
in response (if not 200 or 3xx). You can also use a
wildcard, like '3*' or '*'
:param expect_errors: boolean value, if this is False, then if
anything is written to environ wsgi.errors it
will be an error. If it is True, then
non-200/3xx responses are also okay
:param params: a query string, or a dictionary that will be encoded
into a query string. You may also include a URL query
string on the url
"""
response = self.app.delete(url,
headers=headers,
extra_environ=extra_environ,
status=status,
expect_errors=expect_errors,
params=params)
return response
def get_json(self, url, headers=None, extra_environ=None,
status=None, expect_errors=False, **params):
"""Send HTTP GET request.

View File

@@ -0,0 +1,53 @@
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import httmock
from refstack.tests import api
class TestGuidelinesEndpoint(api.FunctionalTest):
"""Test case for the 'guidelines' API endpoint."""
URL = '/v1/guidelines/'
def test_get_guideline_list(self):
@httmock.all_requests
def github_api_mock(url, request):
headers = {'content-type': 'application/json'}
content = [{'name': '2015.03.json', 'type': 'file'},
{'name': '2015.next.json', 'type': 'file'},
{'name': '2015.03', 'type': 'dir'}]
content = json.dumps(content)
return httmock.response(200, content, headers, None, 5, request)
with httmock.HTTMock(github_api_mock):
actual_response = self.get_json(self.URL)
expected_response = ['2015.03.json']
self.assertEqual(expected_response, actual_response)
def test_get_guideline_file(self):
@httmock.all_requests
def github_mock(url, request):
content = {'foo': 'bar'}
return httmock.response(200, content, None, None, 5, request)
url = self.URL + "2015.03"
with httmock.HTTMock(github_mock):
actual_response = self.get_json(url)
expected_response = {'foo': 'bar'}
self.assertEqual(expected_response, actual_response)

View File

@@ -0,0 +1,81 @@
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import binascii
import json
from Crypto.Hash import SHA256
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5
import mock
import webtest.app
from refstack.tests import api
from refstack import db
class TestProfileEndpoint(api.FunctionalTest):
"""Test case for the 'profile' API endpoint."""
URL = '/v1/profile/'
def setUp(self):
super(TestProfileEndpoint, self).setUp()
self.user_info = {
'openid': 'test-open-id',
'email': 'foo@bar.com',
'fullname': 'Foo Bar'
}
db.user_save(self.user_info)
@mock.patch('refstack.api.utils.get_user_id', return_value='test-open-id')
def test_get(self, mock_get_user):
response = self.get_json(self.URL)
self.user_info['is_admin'] = False
self.assertEqual(self.user_info, response)
@mock.patch('refstack.api.utils.get_user_id', return_value='test-open-id')
def test_pubkeys(self, mock_get_user):
"""Test '/v1/profile/pubkeys' API endpoint."""
url = self.URL + 'pubkeys'
data_hash = SHA256.new()
data_hash.update('signature'.encode('utf-8'))
key = RSA.generate(1024)
signer = PKCS1_v1_5.new(key)
sign = signer.sign(data_hash)
pubkey = key.publickey().exportKey('OpenSSH')
body = {'raw_key': pubkey,
'self_signature': binascii.b2a_hex(sign)}
json_params = json.dumps(body)
# POST endpoint
pubkey_id = self.post_json(url, params=json_params)
# GET endpoint
user_pubkeys = self.get_json(url)
self.assertEqual(1, len(user_pubkeys))
self.assertEqual(pubkey.split()[1], user_pubkeys[0]['pubkey'])
self.assertEqual('ssh-rsa', user_pubkeys[0]['format'])
self.assertEqual(pubkey_id, user_pubkeys[0]['id'])
delete_url = '{}/{}'.format(url, pubkey_id)
# DELETE endpoint
response = self.delete(delete_url)
self.assertEqual(204, response.status_code)
user_pubkeys = self.get_json(url)
self.assertEqual(0, len(user_pubkeys))
# DELETE endpoint - nonexistent pubkey
self.assertRaises(webtest.app.AppError, self.delete, delete_url)

View File

@@ -1,4 +1,3 @@
# Copyright (c) 2015 Mirantis, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@@ -13,12 +12,9 @@
# License for the specific language governing permissions and limitations
# under the License.
"""Functional tests for refstack's API."""
import json
import uuid
import httmock
from oslo_config import fixture as config_fixture
import six
import webtest.app
@@ -44,13 +40,13 @@ FAKE_JSON_WITH_EMPTY_RESULTS = {
}
class TestResultsController(api.FunctionalTest):
"""Test case for ResultsController."""
class TestResultsEndpoint(api.FunctionalTest):
"""Test case for the 'results' API endpoint."""
URL = '/v1/results/'
def setUp(self):
super(TestResultsController, self).setUp()
super(TestResultsEndpoint, self).setUp()
self.config_fixture = config_fixture.Config()
self.CONF = self.useFixture(self.config_fixture).conf
@@ -146,115 +142,84 @@ class TestResultsController(api.FunctionalTest):
self.assertEqual(page_two['pagination']['current_page'], 2)
self.assertEqual(page_two['pagination']['total_pages'], 2)
def test_get_with_not_existing_page(self):
self.assertRaises(webtest.app.AppError,
self.get_json,
'/v1/results?page=2')
def test_get_with_not_existing_page(self):
self.assertRaises(webtest.app.AppError,
self.get_json,
'/v1/results?page=2')
def test_get_with_empty_database(self):
results = self.get_json(self.URL)
self.assertEqual(results, [])
def test_get_with_empty_database(self):
results = self.get_json(self.URL)
self.assertEqual([], results['results'])
def test_get_with_cpid_filter(self):
self.CONF.set_override('results_per_page',
2,
'api')
def test_get_with_cpid_filter(self):
self.CONF.set_override('results_per_page',
2,
'api')
responses = []
for i in range(2):
fake_results = {
'cpid': '12345',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
json_result = json.dumps(fake_results)
actual_response = self.post_json(self.URL,
params=json_result)
responses.append(actual_response)
responses = []
for i in range(2):
fake_results = {
'cpid': '12345',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
json_result = json.dumps(fake_results)
actual_response = self.post_json(self.URL,
params=json_result)
responses.append(actual_response)
for i in range(3):
fake_results = {
'cpid': '54321',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
results = self.get_json('/v1/results?page=1&cpid=12345')
self.asserEqual(len(results), 2)
for r in results:
self.assertIn(r['test_id'], responses)
def test_get_with_date_filters(self):
self.CONF.set_override('results_per_page',
10,
'api')
responses = []
for i in range(5):
fake_results = {
'cpid': '12345',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
json_result = json.dumps(fake_results)
actual_response = self.post_json(self.URL,
params=json_result)
responses.append(actual_response)
all_results = self.get_json(self.URL)
slice_results = all_results[1:3]
url = 'v1/results?start_date=%(start)s&end_date=%(end)s' % {
'start': slice_results[2]['created_at'],
'end': slice_results[0]['created_at']
for i in range(3):
fake_results = {
'cpid': '54321',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
filtering_results = self.get_json(url)
self.assertEqual(len(filtering_results), 3)
for r in slice_results:
self.assertEqual(r, filtering_results)
results = self.get_json('/v1/results?page=1&cpid=12345')
self.assertEqual(len(results), 2)
response_test_ids = [test['test_id'] for test in responses[0:2]]
for r in results['results']:
self.assertIn(r['id'], response_test_ids)
def test_get_with_date_filters(self):
self.CONF.set_override('results_per_page',
10,
'api')
class TestGuidelinesController(api.FunctionalTest):
"""Test case for GuidelinesController."""
responses = []
for i in range(5):
fake_results = {
'cpid': '12345',
'duration_seconds': i,
'results': [
{'name': 'tempest.foo'},
{'name': 'tempest.bar'}
]
}
json_result = json.dumps(fake_results)
actual_response = self.post_json(self.URL,
params=json_result)
responses.append(actual_response)
URL = '/v1/guidelines/'
all_results = self.get_json(self.URL)
def test_get_guideline_list(self):
@httmock.all_requests
def github_api_mock(url, request):
headers = {'content-type': 'application/json'}
content = [{'name': '2015.03.json', 'type': 'file'},
{'name': '2015.next.json', 'type': 'file'},
{'name': '2015.03', 'type': 'dir'}]
content = json.dumps(content)
return httmock.response(200, content, headers, None, 5, request)
slice_results = all_results['results'][1:4]
with httmock.HTTMock(github_api_mock):
actual_response = self.get_json(self.URL)
url = '/v1/results?start_date=%(start)s&end_date=%(end)s' % {
'start': slice_results[2]['created_at'],
'end': slice_results[0]['created_at']
}
expected_response = ['2015.03.json']
self.assertEqual(expected_response, actual_response)
filtering_results = self.get_json(url)
for r in slice_results:
self.assertIn(r, filtering_results['results'])
def test_get_guideline_file(self):
@httmock.all_requests
def github_mock(url, request):
content = {'foo': 'bar'}
return httmock.response(200, content, None, None, 5, request)
url = self.URL + "2015.03"
with httmock.HTTMock(github_mock):
actual_response = self.get_json(url)
expected_response = {'foo': 'bar'}
self.assertEqual(expected_response, actual_response)
url = '/v1/results?end_date=1000-01-01 12:00:00'
filtering_results = self.get_json(url)
self.assertEqual([], filtering_results['results'])