openstack/ironic-python-agent
Code Issues Proposed changes

Merge "Adds more functional tests for commands"

Browse Source
This commit is contained in:
Jenkins 2015-10-07 09:23:18 +00:00 committed by Gerrit Code Review
commit 2c0ebdea09
3 changed files with 101 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
ironic_python_agent/tests/functional/base.py
View File

@ -24,32 +24,61 @@ from ironic_python_agent import agent
class FunctionalBase(test_base.BaseTestCase):
def setUp(self):
"""Start the agent and wait for it to start"""
super(FunctionalBase, self).setUp()
mpl = multiprocessing.log_to_stderr()
mpl.setLevel(logging.INFO)
test_port = os.environ.get('TEST_PORT', '9999')
self.test_port = os.environ.get('TEST_PORT', '9999')
# Build a basic standalone agent using the config option defaults.
# 127.0.0.1:6835 is the fake Ironic client.
self.agent = agent.IronicPythonAgent(
'http://127.0.0.1:6835', 'localhost',
('0.0.0.0', int(test_port)), 3, 10, None, 300, 1, 'agent_ipmitool',
True)
('0.0.0.0', int(self.test_port)), 3, 10, None, 300, 1,
'agent_ipmitool', True)
self.process = multiprocessing.Process(
target=self.agent.run)
self.process.start()
self.addCleanup(self.process.terminate)
# Wait for process to start, otherwise we have a race for tests
sleep_time = 0.1
tries = 0
max_tries = os.environ.get('IPA_WAIT_TIME', '2')
while tries < int(max_tries):
max_tries = int(os.environ.get('IPA_WAIT_TRIES', '100'))
while tries < max_tries:
try:
return requests.get(
'http://localhost:%s/v1/commands' % test_port)
return self.request('get', 'commands')
except requests.ConnectionError:
time.sleep(.1)
time.sleep(sleep_time)
tries += 1
raise IOError('Agent did not start after %s seconds.' % max_tries)
raise IOError('Agent did not start after %s seconds.' % (max_tries *
sleep_time))
def request(self, method, path, expect_error=None, expect_json=True,
**kwargs):
"""Send a request to the agent and verifies response.
:param: method type of request to send as a string
:param: path desired API endpoint to request, for example 'commands'
:param: expect_error error code to expect, if an error is expected
:param: expect_json whether to expect a JSON response. if True, convert
it to a dict before returning, otherwise return the
Response object
:param **kwargs: keyword args to pass to the request method
:raises: HTTPError if an error is returned that was not expected
:raises: AssertionError if a received HTTP status code does not match
expect_error
:returns: the response object
"""
res = requests.request(method, 'http://localhost:%s/v1/%s' %
(self.test_port, path), **kwargs)
if expect_error is not None:
self.assertEqual(expect_error, res.status_code)
else:
res.raise_for_status()
if expect_json:
return res.json()
else:
return res

71
ironic_python_agent/tests/functional/test_commands.py
View File

@ -12,16 +12,69 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import requests
from ironic_python_agent.tests.functional import base
class TestCommands(base.FunctionalBase):
def test_empty_commands(self):
commands = requests.get(
'http://localhost:%s/v1/commands' % os.environ.get('TEST_PORT',
'9999'))
self.assertEqual(200, commands.status_code)
self.assertEqual({'commands': []}, commands.json())
"""Tests the commands API.
These tests are structured monolithically as one test with multiple steps
to preserve ordering and ensure IPA state remains consistent across
different test runs.
"""
def step_1_get_empty_commands(self):
response = self.request('get', 'commands')
self.assertEqual({'commands': []}, response)
def step_2_run_command(self):
# NOTE(mariojv): get_clean_steps always returns the default
# HardwareManager clean steps if there's not a more specific HWM. So,
# this command succeeds even with an empty node and port. This test's
# success is required for steps 3 and 4 to succeed.
command = {'name': 'clean.get_clean_steps',
'params': {'node': {}, 'ports': {}}}
response = self.request('post', 'commands', json=command,
headers={'Content-Type': 'application/json'})
self.assertIsNone(response['command_error'])
def step_3_get_commands(self):
# This test relies on step 2 to succeed since step 2 runs the command
# we're checking for
response = self.request('get', 'commands')
self.assertEqual(1, len(response['commands']))
self.assertEqual(
'get_clean_steps', response['commands'][0]['command_name'])
def step_4_get_command_by_id(self):
# First, we have to query the commands API to retrieve the ID. Make
# sure this API call succeeds again, just in case it fails for some
# reason after the last test. This test relies on step 2 to succeed
# since step 2 runs the command we're checking for.
response = self.request('get', 'commands')
command_id = response['commands'][0]['id']
command_from_id = self.request(
'get', 'commands/%s' % command_id)
self.assertEqual('get_clean_steps', command_from_id['command_name'])
def step_5_run_non_existent_command(self):
fake_command = {'name': 'bad_extension.fake_command', 'params': {}}
self.request('post', 'commands', expect_error=404, json=fake_command)
def positive_get_post_command_steps(self):
"""Returns generator with test steps sorted by step number."""
steps_unsorted = [step for step in dir(self)
if step.startswith('step_')]
# The lambda retrieves the step number from the function name and casts
# it to an integer. This is necessary, otherwise a lexicographic sort
# would return ['step_1', 'step_12', 'step_3'] after sorting instead of
# ['step_1', 'step_3', 'step_12'].
steps = sorted(steps_unsorted, key=lambda s: int(s.split('_', 2)[1]))
for name in steps:
yield getattr(self, name)
def test_positive_get_post_commands(self):
for step in self.positive_get_post_command_steps():
step()

2
tox.ini
View File

@ -26,7 +26,7 @@ setenv =
VIRTUAL_ENV={envdir}
OS_TEST_PATH=./ironic_python_agent/tests/functional
TEST_PORT=9999
IPA_WAIT_TIME=5
IPA_WAIT_TRIES=100
commands =
python setup.py testr --slowest --testr-args='{posargs:}'