[PATCH 2/3] missied one rename
Dan Radez
dradez at redhat.com
Tue Dec 6 21:11:11 UTC 2011
---
agent/test_audrey_agent.py | 692 ++++++++++++++++++++++++++++++++++++++++++
agent/test_audrey_startup.py | 692 ------------------------------------------
2 files changed, 692 insertions(+), 692 deletions(-)
create mode 100644 agent/test_audrey_agent.py
delete mode 100644 agent/test_audrey_startup.py
diff --git a/agent/test_audrey_agent.py b/agent/test_audrey_agent.py
new file mode 100644
index 0000000..d87c05f
--- /dev/null
+++ b/agent/test_audrey_agent.py
@@ -0,0 +1,692 @@
+#!/usr/bin/python2.6
+'''
+*
+* Copyright [2011] [Red Hat, Inc.]
+*
+* 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.
+*
+'''
+
+'''
+ test_audrey_agent.py
+
+ Test program for audrey_agent
+'''
+
+import base64
+import logging
+import os
+import os.path
+import tempfile
+import unittest
+import sys
+import tarfile
+
+from audrey_agent import CSClient
+from audrey_agent import ConfigTooling
+from audrey_agent import ASError
+from audrey_agent import parse_args
+from audrey_agent import parse_provides_params
+from audrey_agent import parse_require_config
+from audrey_agent import audrey_script_main
+from audrey_agent import gen_env
+from audrey_agent import _run_cmd, _run_pipe_cmd
+from audrey_agent import generate_provides
+from audrey_agent import setup_logging
+from audrey_agent import discover_config_server
+
+# Helpers and utils
+DUMMY_USER_DATA = '1|http://example.com/|oauthConsumer|oauthSecret'
+DUMMY_CS_CONFIG = {'endpoint': 'http://example.com/',
+ 'oauth_key': 'oauthConsumer',
+ 'oauth_secret': 'oauthSecret',}
+
+try:
+ from cStringIO import StringIO as BIO
+except ImportError: # python 3
+ from io import BytesIO as BIO
+
+class HttpUnitTest(object):
+ '''
+ Description:
+ When testing the http object does not exists. This class provides
+ test methods that could be preformed when doing UNITTESTing.
+ '''
+ class HttpUnitTestResponse(object):
+ '''
+ Description:
+ When testing the http object does not exists. This class
+ provides the test method response that could be preformed
+ when doing UNITTESTing.
+ '''
+ def __init__(self, status):
+ self.status = status
+
+ def add_content_disposition(self):
+ self.__dict__['content-disposition'] = \
+ 'attachment; filename=test.tar.gz'
+
+ def __getitem__(self, key):
+ return self.__dict__[key]
+
+
+ # simple HTTP Response with 200 status code
+ ok_response = HttpUnitTestResponse(200)
+ not_found_response = HttpUnitTestResponse(404)
+
+ def request(self, url, method='GET', body=None, headers=None):
+ '''
+ Handle request when not running live but in test environment.
+ '''
+ body = ''
+ response = HttpUnitTest.ok_response
+ if method == 'GET':
+ if url.find('/configs/') > -1:
+ body = '|service|s1|parameters|param1&%s|param2&%s|' % \
+ (base64.b64encode('value1'), base64.b64encode('value2'))
+ elif url.find('/params/') > -1:
+ body = '|param1¶m2|'
+ elif url.find('/files/') > -1:
+ file_out = BIO()
+ tar = tarfile.open(mode = "w:gz", fileobj = file_out)
+ tar.add('/etc/passwd')
+ tar.close()
+ body = file_out.getvalue()
+ response.add_content_disposition()
+ elif url.endswith('/user-data'):
+ body = base64.b64encode(DUMMY_USER_DATA)
+ elif url.endswith('/no-version-user-data'):
+ body = base64.b64encode('0|endpoint')
+ elif url.endswith('/empty-user-data'):
+ body = base64.b64encode('')
+ elif url.endswith('/gimmie-404'):
+ body = base64.b64encode(DUMMY_USER_DATA)
+ response = HttpUnitTest.not_found_response
+ else:
+ print url
+ response = HttpUnitTest.not_found_response
+ #elif method == 'POST' and url.find('/params/') > -1:
+ # body = ''
+ return response, body
+
+def _write_info_file(filepath, cloud):
+ f = open(filepath, 'w')
+ f.write(cloud)
+ f.close()
+
+# The actual tests
+
+class TestAudreyStarupRunCmds(unittest.TestCase):
+ '''
+ Test the _run*cmd functions
+ '''
+ def test_success_run_pipe_cmd(self):
+ self.assertEqual("'test'\n",
+ _run_pipe_cmd(["echo", "'test'"], ["grep", "test"])['out'])
+
+ def test_cmd2_fail_run_pipe_cmd(self):
+ self.assertEqual("[Errno 2] No such file or directory",
+ _run_pipe_cmd(["echo", "'test'"], ["notreal"])['err'])
+
+ def test_cmd1_fail_run_pipe_cmd(self):
+ self.assertEqual("[Errno 2] No such file or directory",
+ _run_pipe_cmd(["notreal"], ["echo", "'test'"])['err'])
+
+class TestAudreyStartupConfigTooling(unittest.TestCase):
+ '''
+ Make sure all the Config tooling is tested
+ '''
+ def test_is_user_supplied(self):
+ ConfigTooling('test_tooling').is_user_supplied()
+
+ def test_is_rh_supplied(self):
+ ConfigTooling('test_tooling').is_rh_supplied()
+
+ def test_empty_find_tooling(self):
+ self.assertRaises(ASError, ConfigTooling('test_tooling').find_tooling, '')
+
+ def test_fail_to_create_tooling_dir(self):
+ self.assertRaises(ASError, ConfigTooling, tool_dir='/not/real/dir')
+
+class TestAudreyStartupRequiredConfig(unittest.TestCase):
+ '''
+ Class for exercising the parsing of the Required Configs from the CS.
+ '''
+
+ def setUp(self):
+ '''
+ Perform required setup including setting up logging.
+ '''
+ setup_logging(logging.DEBUG, './test_audrey_agent.log')
+
+ def test_success_service_n_params(self):
+ '''
+ Success case:
+ - Exercise parse_require_config() with valid input
+ '''
+ # Establish valid test data:
+
+ src = '|service|jon1' + \
+ '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
+ '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
+ '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
+ '|service|jon2|'
+
+ validation_dict = {'AUDREY_VAR_jon1_jon_server_ip' : '192.168.1.1',
+ 'AUDREY_VAR_jon1_jon_server_ip_2' : '192.168.1.2',
+ 'AUDREY_VAR_jon1_jon_server_ip_3' : '192.168.1.3' }
+
+ print '\nTest Name: test_success_service_n_params()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() success'
+
+ # Exersise code segment
+ services = parse_require_config(src)
+
+ # Validate results
+ self.assertEqual(services[0].name, 'jon1')
+ self.assertEqual(services[1].name, 'jon2')
+
+ for service in services:
+ for param in service.params:
+ name_val = param.split('&')
+ env_var = 'AUDREY_VAR_' + service.name + '_' + name_val[0]
+ print 'name_val[0]: ' + str(name_val[0])
+ print 'param: ' + str(param)
+ print 'services.name: ' + str(service.name)
+
+ cmd = ['/usr/bin/printenv', env_var]
+ ret = _run_cmd(cmd)
+ self.assertEqual(ret['out'][:-1], \
+ validation_dict[env_var])
+
+ def test_success_empty_source(self):
+ '''
+ Success case:
+ - Exercise parse_require_config() with valid empty input
+ '''
+
+ # Establish valid test data:
+ src = '||'
+ print '\nTest Name: test_success_empty_source()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() success'
+
+ # Exersise code segment
+ services = parse_require_config(src)
+ print 'services: ' + str(services)
+
+ # Validate results
+ self.assertEqual(services, [])
+
+ def test_success_empty_service(self):
+ '''
+ Failure case:
+ - Exercise parse_require_config() with valid input
+ '''
+
+ # Establish valid test data:
+ src = '|service|' + \
+ '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
+ '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
+ '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
+ '|service|jon2|'
+
+ validation_dict = {'AUDREY_VAR_jon_server_ip' : '192.168.1.1',
+ 'AUDREY_VAR_jon_server_ip_2' : '192.168.1.2',
+ 'AUDREY_VAR_jon_server_ip_3' : '192.168.1.3' }
+
+ print '\nTest Name: test_success_empty_service()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() success'
+
+ # Exersise code segment
+ services = parse_require_config(src)
+
+ # Validate results
+ self.assertEqual(services[0].name, '')
+ self.assertEqual(services[1].name, 'jon2')
+
+ for service in services:
+ for param in service.params:
+ name_val = param.split('&')
+ env_var = 'AUDREY_VAR_' + name_val[0]
+
+ print 'name_val[0]: ' + str(name_val[0])
+ print 'param: ' + str(param)
+ print 'services.name: ' + str(service.name)
+
+ cmd = ['/usr/bin/printenv', env_var]
+ ret = _run_cmd(cmd)
+ self.assertEqual(ret['out'][:-1], \
+ validation_dict[env_var])
+
+ def test_failure_no_services_name(self):
+ '''
+ Failure case:
+ - Exercise parse_require_config() with valid input
+
+ The slight difference between this test and test_success_empty_services
+ is the success case has an empty service name indicated by "||":
+ |service||paramseters
+
+ and the failure case has no service name:
+ |service|paramseters
+
+ '''
+
+ # Establish valid test data:
+ src = '|service' \
+ '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
+ '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
+ '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
+ '|service|jon2|'
+
+ validation_dict = {'AUDREY_VAR_jon_server_ip' : '192.168.1.1',
+ 'AUDREY_VAR_jon_server_ip_2' : '192.168.1.2',
+ 'AUDREY_VAR_jon_server_ip_3' : '192.168.1.3' }
+
+ print '\nTest Name: test_failure_no_service_names()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() success'
+
+ # Exersise code segment
+ with self.assertRaises(ASError):
+ print 'parse_require_config returned: ' + \
+ str(parse_require_config(src))
+
+ def test_failure_bad_service_name(self):
+ '''
+ Failure case:
+ - Exercise parse_require_config() with valid input
+ '''
+
+ # Establish valid test data:
+ src = '|service|parameters|'
+ print '\nTest Name: test_failure_bad_service_name()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() ASError'
+
+ # Exersise code segment
+ with self.assertRaises(ASError):
+ print 'parse_require_config returned: ' + \
+ str(parse_require_config(src))
+
+class TestAudreyStartupDiscovery(unittest.TestCase):
+ def setUp(self):
+ '''
+ Perform required setup including setting up logging.
+ '''
+ setup_logging(logging.DEBUG, 'test_audrey_agent.log')
+ self.cloud_info_file = 'cloud_info'
+ self.condor_addr_file = 'condor_addr'
+ self.condor_uuid_file = 'condor_uuid'
+
+ def tearDown(self):
+ os.remove(self.cloud_info_file)
+ if os.path.exists(self.condor_addr_file):
+ os.remove(self.condor_addr_file)
+ if os.path.exists(self.condor_uuid_file):
+ os.remove(self.condor_uuid_file)
+
+ def test_ec2(self):
+ _write_info_file(self.cloud_info_file, 'EC2')
+ discover_config_server(self.cloud_info_file, http=HttpUnitTest())
+
+ def test_ec2_404(self):
+ _write_info_file(self.cloud_info_file, 'EC2')
+ self.assertRaises(ASError,
+ discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
+ ec2_user_data='http://169.254.169.254/gimmie-404')
+
+ def test_condorcloud(self):
+ _write_info_file(self.condor_addr_file, '1|endpoint|secret')
+ _write_info_file(self.condor_uuid_file, 'key')
+ _write_info_file(self.cloud_info_file, 'CONDORCLOUD')
+ discover_config_server(self.cloud_info_file,
+ condor_addr_file=self.condor_addr_file,
+ condor_uuid_file=self.condor_uuid_file)
+
+ def test_rhev(self):
+ _write_info_file(self.cloud_info_file, 'RHEV')
+ self.assertRaises(ASError,
+ discover_config_server, self.cloud_info_file)
+
+ def test_vsphere(self):
+ _write_info_file(self.cloud_info_file, 'VSPHERE')
+ self.assertRaises(ASError,
+ discover_config_server, self.cloud_info_file)
+
+ def test_invalid_user_data_version(self):
+ _write_info_file(self.cloud_info_file, 'EC2')
+ self.assertRaises(ASError,
+ discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
+ ec2_user_data='http://169.254.169.254/no-version-user-data')
+
+ def test_invalid_user_data_no_delim(self):
+ _write_info_file(self.cloud_info_file, 'EC2')
+ self.assertRaises(ASError,
+ discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
+ ec2_user_data='http://169.254.169.254/empty-user-data')
+
+
+class TestAudreyStartupProvidesParameters(unittest.TestCase):
+ '''
+ Class for exercising the parsing of the Provides ParametersConfigs
+ from the CS.
+ '''
+
+ def setUp(self):
+ '''
+ Perform required setup including setting up logging.
+ '''
+ setup_logging(logging.DEBUG, './test_audrey_agent.log')
+
+ def test_success_parameters(self):
+ '''
+ Success case:
+ - Exercise parse_provides_params() and generate_provides()
+ with valid input
+ '''
+
+ # Establish valid test data:
+ src = '|operatingsystem&is_virtual|'
+
+ print '\nTest Name: test_success_parameters()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_provides_params() success'
+
+ expected_params_list = ['operatingsystem', 'is_virtual']
+
+ # Exersise code segment
+ params_list = parse_provides_params(src)
+ provides = generate_provides(src)
+ print 'src: ' + str(src)
+ print 'params_list: ' + str(params_list)
+ print 'provides: ' + str(provides)
+ print 'len(provides): ' + str(len(provides))
+
+ # Validate results
+ self.assertEqual(params_list, expected_params_list)
+
+ # The values are not validatable because they are unpredictable
+ # but all the expected parameters should be returned.
+ # Note: %7C is the encoded |, %26 is the encoded &
+ self.assertTrue('audrey_data=%7Coperatingsystem' in provides)
+ for param in expected_params_list:
+ self.assertTrue('%7C' + str(param) in provides)
+
+ def test_success_no_params(self):
+ '''
+ Success case:
+ - Exercise parse_provides_params() and generate_provides()
+ with valid input
+ - Containging an unavailable parameter
+ '''
+
+ # Establish valid test data:
+ src = '|uptime_days&unavailable_dogs&ipaddress|'
+
+ print '\nTest Name: test_success_no_params()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_provides_params() success'
+
+ expected_params_list = ['uptime_days', 'unavailable_dogs', 'ipaddress']
+
+ # Exersise code segment
+ params_list = parse_provides_params(src)
+ provides = generate_provides(src)
+ print 'src: ' + str(src)
+ print 'params_list: ' + str(params_list)
+ print 'provides: ' + str(provides)
+
+ # Validate results
+ self.assertEqual(params_list, expected_params_list)
+
+ # The values are not validatable because they are unpredictable
+ # but all the expected parameters should be returned.
+ # Note: %7C is the encoded |, %26 is the encoded &
+ for param in expected_params_list:
+ self.assertTrue('%7C' + str(param) in provides)
+
+ # Confirm unavailable parameters return an empty string.
+ self.assertTrue('%7C' + 'unavailable_dogs' + '%26%7C' in provides)
+
+ def test_success_one_parameters(self):
+ '''
+ Success case:
+ - Exercise parse_provides_params() and generate_provides()
+ with valid input
+ - with only one parameter
+ '''
+
+ # Establish valid test data:
+ src = '|uptime_days|'
+
+ print '\nTest Name: test_success_parameters()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_provides_params() success'
+
+ expected_params_list = ['uptime_days']
+
+ # Exersise code segment
+ params_list = parse_provides_params(src)
+ provides = generate_provides(src)
+ print 'src: ' + str(src)
+ print 'params_list: ' + str(params_list)
+ print 'provides: ' + str(provides)
+
+ # Validate results
+ self.assertEqual(params_list, expected_params_list)
+
+ # The values are not validatable because they are unpredictable
+ # but all the expected parameters should be returned.
+ # Note: %7C is the encoded |, %26 is the encoded &
+ for param in expected_params_list:
+ self.assertTrue('%7C' + str(param) in provides)
+
+ def test_success_one_parameter(self):
+ '''
+ Success case:
+ - Exercise parse_provides_params() and generate_provides()
+ with valid input
+ - With only one parameter which is unavailable
+ '''
+
+ # Establish valid test data:
+ src = '|unavailable_dogs|'
+
+ print '\nTest Name: test_success_one_parameter()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_provides_params() success'
+
+ expected_params_list = ['unavailable_dogs']
+
+ # Exersise code segment
+ params_list = parse_provides_params(src)
+ provides = generate_provides(src)
+ print 'src: ' + str(src)
+ print 'params_list: ' + str(params_list)
+ print 'provides: ' + str(provides)
+
+ # Validate results
+ self.assertEqual(params_list, expected_params_list)
+
+ # The values are not validatable because they are unpredictable
+ # but all the expected parameters should be returned.
+ # Note: %7C is the encoded |, %26 is the encoded &
+ for param in expected_params_list:
+ self.assertTrue('%7C' + str(param) in provides)
+
+ # Confirm unavailable parameters return an empty string.
+ self.assertTrue('%7C' + 'unavailable_dogs' + '%26%7C' in provides)
+
+ def test_failure_missing_delimiter(self):
+ '''
+ Failure case:
+ - Exercise parse_provides_params() and generate_provides()
+ with invalid input
+ - missing leading delimiter
+ '''
+
+ # Establish valid test data:
+ src = 'unavailable_dogs|'
+
+ print '\nTest Name: test_failure_missing_delimiter()'
+ print 'Test input:\n' + src
+ print 'Expect: parse_require_config() ASError'
+
+ expected_params_list = ['unavailable_dogs']
+
+ # Exersise code segment and validate results
+ with self.assertRaises(ASError):
+ params_list = parse_provides_params(src)
+
+ with self.assertRaises(ASError):
+ provides = generate_provides(src)
+
+class TestConfigServerClient(unittest.TestCase):
+ '''
+ Class for exercising the gets and put to and from the CS
+ '''
+
+ def setUp(self):
+ '''
+ If the cloud info file is not present assume running in a
+ UNITTEST environment. This will allow for exercising some
+ of the code without having to be running in a cloud VM.
+
+ Set up logging.
+ '''
+
+ setup_logging(logging.DEBUG, './test_audrey_agent.log')
+
+ # Create the client Object
+ self.cs_client = CSClient(**DUMMY_CS_CONFIG)
+ self.cs_client.http = HttpUnitTest()
+
+ def tearDown(self):
+ pass
+
+ def test_success_get_cs_configs(self):
+ '''
+ Success case:
+ - Exercise get_cs_configs()
+ '''
+ print '\n\n--- Test Name: test_success_get_cs_configs ---'
+
+ self.cs_client.get_cs_configs()
+
+ # Add asserts
+ print 'test_success_get_cs_configs() Add asserts'
+ print 'self.cs_client : START \n' + str(self.cs_client) + \
+ '\nself.cs_client : END'
+
+ def test_success_get_cs_tooling(self):
+ '''
+ Success case:
+ - Exercise get_cs_tooling()
+ '''
+ self.cs_client.get_cs_tooling()
+
+ def test_success_get_cs_params(self):
+ '''
+ Success case:
+ - Exercise get_cs_params()
+ '''
+ print '\n\n--- Test Name: test_success_get_cs_params ---'
+
+ self.cs_client.get_cs_params()
+
+ # Add asserts
+ print 'test_success_get_cs_params() Add asserts'
+ print 'self.cs_client : START \n' + str(self.cs_client) + \
+ '\nself.cs_client : END'
+
+ def test_success_get_cs_confs_n_params(self):
+ '''
+ Success case:
+ - Exercise get_cs_configs() and get_cs_params()
+ '''
+ print '\n\n--- Test Name: test_success_get_cs_confs_and_params ---'
+
+ self.cs_client.get_cs_configs()
+ self.cs_client.get_cs_params()
+
+ # Add asserts
+ print 'test_success_get_cs_confs_n_params() Add asserts'
+ print 'self.cs_client : START \n' + str(self.cs_client) + \
+ '\nself.cs_client : END'
+ print 'test_success_get_cs_confs_n_params() Add asserts'
+
+ def test_success_put_cs_params_values(self):
+ '''
+ Success case:
+ - Exercise put_cs_params_values()
+ '''
+ self.cs_client.put_cs_params_values('')
+
+ def test_error_http_status(self):
+ '''
+ Success case:
+ - Exercise put_cs_params_values()
+ '''
+ self.assertRaises(ASError, self.cs_client._validate_http_status, 401)
+
+class TestAudreyScript(unittest.TestCase):
+ '''
+ Class for exercising the full audrey script functionality
+ '''
+
+ def setUp(self):
+ '''
+ Perform required setup including setting up logging.
+
+ This test currently require to be run in a cloud VM
+ with a live Config Server.
+ '''
+ setup_logging(logging.DEBUG, './test_audrey_agent.log')
+ # make a copy of argv
+ self.argv = list(sys.argv)
+
+ def tearDown(self):
+ # replace argv
+ sys.argv = list(self.argv)
+
+ def test_audrey_script_main(self):
+ '''
+ Perform what the audrey script will do.
+ '''
+ cloud_info_file = 'cloud_info'
+ sys.argv.extend(['-p'])
+ _write_info_file(cloud_info_file, 'EC2')
+ audrey_script_main(HttpUnitTest())
+ os.remove(cloud_info_file)
+
+ def test_fail_audrey_script_main(self):
+ '''
+ Perform what the audrey script will do.
+ '''
+ self.assertRaises(ASError, audrey_script_main)
+
+ def test_empty_gen_env(self):
+ self.assertRaises(ASError, gen_env, '', '')
+
+ # doesn't actually test what I wanted it to.
+ #def test_parse_require_config(self):
+ # self.assertRaises(ASError, parse_require_config, '')
+
+if __name__ == '__main__':
+
+ setup_logging(logging.DEBUG, logfile_name='./test_audrey_agent.log')
+ unittest.main()
diff --git a/agent/test_audrey_startup.py b/agent/test_audrey_startup.py
deleted file mode 100644
index d87c05f..0000000
--- a/agent/test_audrey_startup.py
+++ /dev/null
@@ -1,692 +0,0 @@
-#!/usr/bin/python2.6
-'''
-*
-* Copyright [2011] [Red Hat, Inc.]
-*
-* 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.
-*
-'''
-
-'''
- test_audrey_agent.py
-
- Test program for audrey_agent
-'''
-
-import base64
-import logging
-import os
-import os.path
-import tempfile
-import unittest
-import sys
-import tarfile
-
-from audrey_agent import CSClient
-from audrey_agent import ConfigTooling
-from audrey_agent import ASError
-from audrey_agent import parse_args
-from audrey_agent import parse_provides_params
-from audrey_agent import parse_require_config
-from audrey_agent import audrey_script_main
-from audrey_agent import gen_env
-from audrey_agent import _run_cmd, _run_pipe_cmd
-from audrey_agent import generate_provides
-from audrey_agent import setup_logging
-from audrey_agent import discover_config_server
-
-# Helpers and utils
-DUMMY_USER_DATA = '1|http://example.com/|oauthConsumer|oauthSecret'
-DUMMY_CS_CONFIG = {'endpoint': 'http://example.com/',
- 'oauth_key': 'oauthConsumer',
- 'oauth_secret': 'oauthSecret',}
-
-try:
- from cStringIO import StringIO as BIO
-except ImportError: # python 3
- from io import BytesIO as BIO
-
-class HttpUnitTest(object):
- '''
- Description:
- When testing the http object does not exists. This class provides
- test methods that could be preformed when doing UNITTESTing.
- '''
- class HttpUnitTestResponse(object):
- '''
- Description:
- When testing the http object does not exists. This class
- provides the test method response that could be preformed
- when doing UNITTESTing.
- '''
- def __init__(self, status):
- self.status = status
-
- def add_content_disposition(self):
- self.__dict__['content-disposition'] = \
- 'attachment; filename=test.tar.gz'
-
- def __getitem__(self, key):
- return self.__dict__[key]
-
-
- # simple HTTP Response with 200 status code
- ok_response = HttpUnitTestResponse(200)
- not_found_response = HttpUnitTestResponse(404)
-
- def request(self, url, method='GET', body=None, headers=None):
- '''
- Handle request when not running live but in test environment.
- '''
- body = ''
- response = HttpUnitTest.ok_response
- if method == 'GET':
- if url.find('/configs/') > -1:
- body = '|service|s1|parameters|param1&%s|param2&%s|' % \
- (base64.b64encode('value1'), base64.b64encode('value2'))
- elif url.find('/params/') > -1:
- body = '|param1¶m2|'
- elif url.find('/files/') > -1:
- file_out = BIO()
- tar = tarfile.open(mode = "w:gz", fileobj = file_out)
- tar.add('/etc/passwd')
- tar.close()
- body = file_out.getvalue()
- response.add_content_disposition()
- elif url.endswith('/user-data'):
- body = base64.b64encode(DUMMY_USER_DATA)
- elif url.endswith('/no-version-user-data'):
- body = base64.b64encode('0|endpoint')
- elif url.endswith('/empty-user-data'):
- body = base64.b64encode('')
- elif url.endswith('/gimmie-404'):
- body = base64.b64encode(DUMMY_USER_DATA)
- response = HttpUnitTest.not_found_response
- else:
- print url
- response = HttpUnitTest.not_found_response
- #elif method == 'POST' and url.find('/params/') > -1:
- # body = ''
- return response, body
-
-def _write_info_file(filepath, cloud):
- f = open(filepath, 'w')
- f.write(cloud)
- f.close()
-
-# The actual tests
-
-class TestAudreyStarupRunCmds(unittest.TestCase):
- '''
- Test the _run*cmd functions
- '''
- def test_success_run_pipe_cmd(self):
- self.assertEqual("'test'\n",
- _run_pipe_cmd(["echo", "'test'"], ["grep", "test"])['out'])
-
- def test_cmd2_fail_run_pipe_cmd(self):
- self.assertEqual("[Errno 2] No such file or directory",
- _run_pipe_cmd(["echo", "'test'"], ["notreal"])['err'])
-
- def test_cmd1_fail_run_pipe_cmd(self):
- self.assertEqual("[Errno 2] No such file or directory",
- _run_pipe_cmd(["notreal"], ["echo", "'test'"])['err'])
-
-class TestAudreyStartupConfigTooling(unittest.TestCase):
- '''
- Make sure all the Config tooling is tested
- '''
- def test_is_user_supplied(self):
- ConfigTooling('test_tooling').is_user_supplied()
-
- def test_is_rh_supplied(self):
- ConfigTooling('test_tooling').is_rh_supplied()
-
- def test_empty_find_tooling(self):
- self.assertRaises(ASError, ConfigTooling('test_tooling').find_tooling, '')
-
- def test_fail_to_create_tooling_dir(self):
- self.assertRaises(ASError, ConfigTooling, tool_dir='/not/real/dir')
-
-class TestAudreyStartupRequiredConfig(unittest.TestCase):
- '''
- Class for exercising the parsing of the Required Configs from the CS.
- '''
-
- def setUp(self):
- '''
- Perform required setup including setting up logging.
- '''
- setup_logging(logging.DEBUG, './test_audrey_agent.log')
-
- def test_success_service_n_params(self):
- '''
- Success case:
- - Exercise parse_require_config() with valid input
- '''
- # Establish valid test data:
-
- src = '|service|jon1' + \
- '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
- '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
- '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
- '|service|jon2|'
-
- validation_dict = {'AUDREY_VAR_jon1_jon_server_ip' : '192.168.1.1',
- 'AUDREY_VAR_jon1_jon_server_ip_2' : '192.168.1.2',
- 'AUDREY_VAR_jon1_jon_server_ip_3' : '192.168.1.3' }
-
- print '\nTest Name: test_success_service_n_params()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() success'
-
- # Exersise code segment
- services = parse_require_config(src)
-
- # Validate results
- self.assertEqual(services[0].name, 'jon1')
- self.assertEqual(services[1].name, 'jon2')
-
- for service in services:
- for param in service.params:
- name_val = param.split('&')
- env_var = 'AUDREY_VAR_' + service.name + '_' + name_val[0]
- print 'name_val[0]: ' + str(name_val[0])
- print 'param: ' + str(param)
- print 'services.name: ' + str(service.name)
-
- cmd = ['/usr/bin/printenv', env_var]
- ret = _run_cmd(cmd)
- self.assertEqual(ret['out'][:-1], \
- validation_dict[env_var])
-
- def test_success_empty_source(self):
- '''
- Success case:
- - Exercise parse_require_config() with valid empty input
- '''
-
- # Establish valid test data:
- src = '||'
- print '\nTest Name: test_success_empty_source()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() success'
-
- # Exersise code segment
- services = parse_require_config(src)
- print 'services: ' + str(services)
-
- # Validate results
- self.assertEqual(services, [])
-
- def test_success_empty_service(self):
- '''
- Failure case:
- - Exercise parse_require_config() with valid input
- '''
-
- # Establish valid test data:
- src = '|service|' + \
- '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
- '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
- '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
- '|service|jon2|'
-
- validation_dict = {'AUDREY_VAR_jon_server_ip' : '192.168.1.1',
- 'AUDREY_VAR_jon_server_ip_2' : '192.168.1.2',
- 'AUDREY_VAR_jon_server_ip_3' : '192.168.1.3' }
-
- print '\nTest Name: test_success_empty_service()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() success'
-
- # Exersise code segment
- services = parse_require_config(src)
-
- # Validate results
- self.assertEqual(services[0].name, '')
- self.assertEqual(services[1].name, 'jon2')
-
- for service in services:
- for param in service.params:
- name_val = param.split('&')
- env_var = 'AUDREY_VAR_' + name_val[0]
-
- print 'name_val[0]: ' + str(name_val[0])
- print 'param: ' + str(param)
- print 'services.name: ' + str(service.name)
-
- cmd = ['/usr/bin/printenv', env_var]
- ret = _run_cmd(cmd)
- self.assertEqual(ret['out'][:-1], \
- validation_dict[env_var])
-
- def test_failure_no_services_name(self):
- '''
- Failure case:
- - Exercise parse_require_config() with valid input
-
- The slight difference between this test and test_success_empty_services
- is the success case has an empty service name indicated by "||":
- |service||paramseters
-
- and the failure case has no service name:
- |service|paramseters
-
- '''
-
- # Establish valid test data:
- src = '|service' \
- '|parameters|jon_server_ip&' + base64.b64encode('192.168.1.1') + \
- '|jon_server_ip_2&' + base64.b64encode('192.168.1.2') + \
- '|jon_server_ip_3&' + base64.b64encode('192.168.1.3') + \
- '|service|jon2|'
-
- validation_dict = {'AUDREY_VAR_jon_server_ip' : '192.168.1.1',
- 'AUDREY_VAR_jon_server_ip_2' : '192.168.1.2',
- 'AUDREY_VAR_jon_server_ip_3' : '192.168.1.3' }
-
- print '\nTest Name: test_failure_no_service_names()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() success'
-
- # Exersise code segment
- with self.assertRaises(ASError):
- print 'parse_require_config returned: ' + \
- str(parse_require_config(src))
-
- def test_failure_bad_service_name(self):
- '''
- Failure case:
- - Exercise parse_require_config() with valid input
- '''
-
- # Establish valid test data:
- src = '|service|parameters|'
- print '\nTest Name: test_failure_bad_service_name()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() ASError'
-
- # Exersise code segment
- with self.assertRaises(ASError):
- print 'parse_require_config returned: ' + \
- str(parse_require_config(src))
-
-class TestAudreyStartupDiscovery(unittest.TestCase):
- def setUp(self):
- '''
- Perform required setup including setting up logging.
- '''
- setup_logging(logging.DEBUG, 'test_audrey_agent.log')
- self.cloud_info_file = 'cloud_info'
- self.condor_addr_file = 'condor_addr'
- self.condor_uuid_file = 'condor_uuid'
-
- def tearDown(self):
- os.remove(self.cloud_info_file)
- if os.path.exists(self.condor_addr_file):
- os.remove(self.condor_addr_file)
- if os.path.exists(self.condor_uuid_file):
- os.remove(self.condor_uuid_file)
-
- def test_ec2(self):
- _write_info_file(self.cloud_info_file, 'EC2')
- discover_config_server(self.cloud_info_file, http=HttpUnitTest())
-
- def test_ec2_404(self):
- _write_info_file(self.cloud_info_file, 'EC2')
- self.assertRaises(ASError,
- discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
- ec2_user_data='http://169.254.169.254/gimmie-404')
-
- def test_condorcloud(self):
- _write_info_file(self.condor_addr_file, '1|endpoint|secret')
- _write_info_file(self.condor_uuid_file, 'key')
- _write_info_file(self.cloud_info_file, 'CONDORCLOUD')
- discover_config_server(self.cloud_info_file,
- condor_addr_file=self.condor_addr_file,
- condor_uuid_file=self.condor_uuid_file)
-
- def test_rhev(self):
- _write_info_file(self.cloud_info_file, 'RHEV')
- self.assertRaises(ASError,
- discover_config_server, self.cloud_info_file)
-
- def test_vsphere(self):
- _write_info_file(self.cloud_info_file, 'VSPHERE')
- self.assertRaises(ASError,
- discover_config_server, self.cloud_info_file)
-
- def test_invalid_user_data_version(self):
- _write_info_file(self.cloud_info_file, 'EC2')
- self.assertRaises(ASError,
- discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
- ec2_user_data='http://169.254.169.254/no-version-user-data')
-
- def test_invalid_user_data_no_delim(self):
- _write_info_file(self.cloud_info_file, 'EC2')
- self.assertRaises(ASError,
- discover_config_server, self.cloud_info_file, http=HttpUnitTest(),
- ec2_user_data='http://169.254.169.254/empty-user-data')
-
-
-class TestAudreyStartupProvidesParameters(unittest.TestCase):
- '''
- Class for exercising the parsing of the Provides ParametersConfigs
- from the CS.
- '''
-
- def setUp(self):
- '''
- Perform required setup including setting up logging.
- '''
- setup_logging(logging.DEBUG, './test_audrey_agent.log')
-
- def test_success_parameters(self):
- '''
- Success case:
- - Exercise parse_provides_params() and generate_provides()
- with valid input
- '''
-
- # Establish valid test data:
- src = '|operatingsystem&is_virtual|'
-
- print '\nTest Name: test_success_parameters()'
- print 'Test input:\n' + src
- print 'Expect: parse_provides_params() success'
-
- expected_params_list = ['operatingsystem', 'is_virtual']
-
- # Exersise code segment
- params_list = parse_provides_params(src)
- provides = generate_provides(src)
- print 'src: ' + str(src)
- print 'params_list: ' + str(params_list)
- print 'provides: ' + str(provides)
- print 'len(provides): ' + str(len(provides))
-
- # Validate results
- self.assertEqual(params_list, expected_params_list)
-
- # The values are not validatable because they are unpredictable
- # but all the expected parameters should be returned.
- # Note: %7C is the encoded |, %26 is the encoded &
- self.assertTrue('audrey_data=%7Coperatingsystem' in provides)
- for param in expected_params_list:
- self.assertTrue('%7C' + str(param) in provides)
-
- def test_success_no_params(self):
- '''
- Success case:
- - Exercise parse_provides_params() and generate_provides()
- with valid input
- - Containging an unavailable parameter
- '''
-
- # Establish valid test data:
- src = '|uptime_days&unavailable_dogs&ipaddress|'
-
- print '\nTest Name: test_success_no_params()'
- print 'Test input:\n' + src
- print 'Expect: parse_provides_params() success'
-
- expected_params_list = ['uptime_days', 'unavailable_dogs', 'ipaddress']
-
- # Exersise code segment
- params_list = parse_provides_params(src)
- provides = generate_provides(src)
- print 'src: ' + str(src)
- print 'params_list: ' + str(params_list)
- print 'provides: ' + str(provides)
-
- # Validate results
- self.assertEqual(params_list, expected_params_list)
-
- # The values are not validatable because they are unpredictable
- # but all the expected parameters should be returned.
- # Note: %7C is the encoded |, %26 is the encoded &
- for param in expected_params_list:
- self.assertTrue('%7C' + str(param) in provides)
-
- # Confirm unavailable parameters return an empty string.
- self.assertTrue('%7C' + 'unavailable_dogs' + '%26%7C' in provides)
-
- def test_success_one_parameters(self):
- '''
- Success case:
- - Exercise parse_provides_params() and generate_provides()
- with valid input
- - with only one parameter
- '''
-
- # Establish valid test data:
- src = '|uptime_days|'
-
- print '\nTest Name: test_success_parameters()'
- print 'Test input:\n' + src
- print 'Expect: parse_provides_params() success'
-
- expected_params_list = ['uptime_days']
-
- # Exersise code segment
- params_list = parse_provides_params(src)
- provides = generate_provides(src)
- print 'src: ' + str(src)
- print 'params_list: ' + str(params_list)
- print 'provides: ' + str(provides)
-
- # Validate results
- self.assertEqual(params_list, expected_params_list)
-
- # The values are not validatable because they are unpredictable
- # but all the expected parameters should be returned.
- # Note: %7C is the encoded |, %26 is the encoded &
- for param in expected_params_list:
- self.assertTrue('%7C' + str(param) in provides)
-
- def test_success_one_parameter(self):
- '''
- Success case:
- - Exercise parse_provides_params() and generate_provides()
- with valid input
- - With only one parameter which is unavailable
- '''
-
- # Establish valid test data:
- src = '|unavailable_dogs|'
-
- print '\nTest Name: test_success_one_parameter()'
- print 'Test input:\n' + src
- print 'Expect: parse_provides_params() success'
-
- expected_params_list = ['unavailable_dogs']
-
- # Exersise code segment
- params_list = parse_provides_params(src)
- provides = generate_provides(src)
- print 'src: ' + str(src)
- print 'params_list: ' + str(params_list)
- print 'provides: ' + str(provides)
-
- # Validate results
- self.assertEqual(params_list, expected_params_list)
-
- # The values are not validatable because they are unpredictable
- # but all the expected parameters should be returned.
- # Note: %7C is the encoded |, %26 is the encoded &
- for param in expected_params_list:
- self.assertTrue('%7C' + str(param) in provides)
-
- # Confirm unavailable parameters return an empty string.
- self.assertTrue('%7C' + 'unavailable_dogs' + '%26%7C' in provides)
-
- def test_failure_missing_delimiter(self):
- '''
- Failure case:
- - Exercise parse_provides_params() and generate_provides()
- with invalid input
- - missing leading delimiter
- '''
-
- # Establish valid test data:
- src = 'unavailable_dogs|'
-
- print '\nTest Name: test_failure_missing_delimiter()'
- print 'Test input:\n' + src
- print 'Expect: parse_require_config() ASError'
-
- expected_params_list = ['unavailable_dogs']
-
- # Exersise code segment and validate results
- with self.assertRaises(ASError):
- params_list = parse_provides_params(src)
-
- with self.assertRaises(ASError):
- provides = generate_provides(src)
-
-class TestConfigServerClient(unittest.TestCase):
- '''
- Class for exercising the gets and put to and from the CS
- '''
-
- def setUp(self):
- '''
- If the cloud info file is not present assume running in a
- UNITTEST environment. This will allow for exercising some
- of the code without having to be running in a cloud VM.
-
- Set up logging.
- '''
-
- setup_logging(logging.DEBUG, './test_audrey_agent.log')
-
- # Create the client Object
- self.cs_client = CSClient(**DUMMY_CS_CONFIG)
- self.cs_client.http = HttpUnitTest()
-
- def tearDown(self):
- pass
-
- def test_success_get_cs_configs(self):
- '''
- Success case:
- - Exercise get_cs_configs()
- '''
- print '\n\n--- Test Name: test_success_get_cs_configs ---'
-
- self.cs_client.get_cs_configs()
-
- # Add asserts
- print 'test_success_get_cs_configs() Add asserts'
- print 'self.cs_client : START \n' + str(self.cs_client) + \
- '\nself.cs_client : END'
-
- def test_success_get_cs_tooling(self):
- '''
- Success case:
- - Exercise get_cs_tooling()
- '''
- self.cs_client.get_cs_tooling()
-
- def test_success_get_cs_params(self):
- '''
- Success case:
- - Exercise get_cs_params()
- '''
- print '\n\n--- Test Name: test_success_get_cs_params ---'
-
- self.cs_client.get_cs_params()
-
- # Add asserts
- print 'test_success_get_cs_params() Add asserts'
- print 'self.cs_client : START \n' + str(self.cs_client) + \
- '\nself.cs_client : END'
-
- def test_success_get_cs_confs_n_params(self):
- '''
- Success case:
- - Exercise get_cs_configs() and get_cs_params()
- '''
- print '\n\n--- Test Name: test_success_get_cs_confs_and_params ---'
-
- self.cs_client.get_cs_configs()
- self.cs_client.get_cs_params()
-
- # Add asserts
- print 'test_success_get_cs_confs_n_params() Add asserts'
- print 'self.cs_client : START \n' + str(self.cs_client) + \
- '\nself.cs_client : END'
- print 'test_success_get_cs_confs_n_params() Add asserts'
-
- def test_success_put_cs_params_values(self):
- '''
- Success case:
- - Exercise put_cs_params_values()
- '''
- self.cs_client.put_cs_params_values('')
-
- def test_error_http_status(self):
- '''
- Success case:
- - Exercise put_cs_params_values()
- '''
- self.assertRaises(ASError, self.cs_client._validate_http_status, 401)
-
-class TestAudreyScript(unittest.TestCase):
- '''
- Class for exercising the full audrey script functionality
- '''
-
- def setUp(self):
- '''
- Perform required setup including setting up logging.
-
- This test currently require to be run in a cloud VM
- with a live Config Server.
- '''
- setup_logging(logging.DEBUG, './test_audrey_agent.log')
- # make a copy of argv
- self.argv = list(sys.argv)
-
- def tearDown(self):
- # replace argv
- sys.argv = list(self.argv)
-
- def test_audrey_script_main(self):
- '''
- Perform what the audrey script will do.
- '''
- cloud_info_file = 'cloud_info'
- sys.argv.extend(['-p'])
- _write_info_file(cloud_info_file, 'EC2')
- audrey_script_main(HttpUnitTest())
- os.remove(cloud_info_file)
-
- def test_fail_audrey_script_main(self):
- '''
- Perform what the audrey script will do.
- '''
- self.assertRaises(ASError, audrey_script_main)
-
- def test_empty_gen_env(self):
- self.assertRaises(ASError, gen_env, '', '')
-
- # doesn't actually test what I wanted it to.
- #def test_parse_require_config(self):
- # self.assertRaises(ASError, parse_require_config, '')
-
-if __name__ == '__main__':
-
- setup_logging(logging.DEBUG, logfile_name='./test_audrey_agent.log')
- unittest.main()
--
1.7.7.3
More information about the aeolus-devel
mailing list