diff options
Diffstat (limited to 'tests/distro')
| -rw-r--r-- | tests/distro/__init__.py | 3 | ||||
| -rw-r--r-- | tests/distro/test_daemon.py | 65 | ||||
| -rw-r--r-- | tests/distro/test_extension.py | 191 | ||||
| -rw-r--r-- | tests/distro/test_monitor.py | 32 | ||||
| -rw-r--r-- | tests/distro/test_protocol_util.py | 89 | ||||
| -rw-r--r-- | tests/distro/test_provision.py | 47 | ||||
| -rw-r--r-- | tests/distro/test_resourceDisk.py (renamed from tests/distro/test_loader.py) | 26 | ||||
| -rw-r--r-- | tests/distro/test_scvmm.py | 84 |
8 files changed, 100 insertions, 437 deletions
diff --git a/tests/distro/__init__.py b/tests/distro/__init__.py index 9bdb27e..2ef4c16 100644 --- a/tests/distro/__init__.py +++ b/tests/distro/__init__.py @@ -14,6 +14,3 @@ # # Requires Python 2.4+ and Openssl 1.0+ # -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx diff --git a/tests/distro/test_daemon.py b/tests/distro/test_daemon.py deleted file mode 100644 index 9d3c45b..0000000 --- a/tests/distro/test_daemon.py +++ /dev/null @@ -1,65 +0,0 @@ -# Copyright 2014 Microsoft Corporation -# -# 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. -# -# Requires Python 2.4+ and Openssl 1.0+ -# -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx - -from tests.tools import * -from azurelinuxagent.distro.loader import get_distro -from azurelinuxagent.exception import * -from azurelinuxagent.distro.default.daemon import * - -class MockDaemonCall(object): - def __init__(self, daemon_handler, count): - self.daemon_handler = daemon_handler - self.count = count - - def __call__(self, *args, **kw): - self.count = self.count - 1 - #Stop daemon after restarting for n times - if self.count <= 0: - self.daemon_handler.running = False - raise Exception("Mock unhandled exception") - -@patch("time.sleep") -class TestDaemon(AgentTestCase): - def test_daemon_restart(self, mock_sleep): - distro = get_distro() - mock_daemon = Mock(side_effect=MockDaemonCall(distro.daemon_handler, 2)) - distro.daemon_handler.daemon = mock_daemon - distro.daemon_handler.check_pid = Mock() - distro.daemon_handler.run() - - mock_sleep.assert_any_call(15) - self.assertEquals(2, distro.daemon_handler.daemon.call_count) - - @patch("azurelinuxagent.distro.default.daemon.conf") - @patch("azurelinuxagent.distro.default.daemon.sys.exit") - def test_check_pid(self, mock_exit, mock_conf, mock_sleep): - distro = get_distro() - mock_pid_file = os.path.join(self.tmp_dir, "pid") - mock_conf.get_agent_pid_file_path = Mock(return_value=mock_pid_file) - - distro.daemon_handler.check_pid() - self.assertTrue(os.path.isfile(mock_pid_file)) - - distro.daemon_handler.check_pid() - mock_exit.assert_any_call(0) - -if __name__ == '__main__': - unittest.main() - diff --git a/tests/distro/test_extension.py b/tests/distro/test_extension.py deleted file mode 100644 index d0b631f..0000000 --- a/tests/distro/test_extension.py +++ /dev/null @@ -1,191 +0,0 @@ -# Copyright 2014 Microsoft Corporation -# -# 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. -# -# Requires Python 2.4+ and Openssl 1.0+ -# -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx - -from tests.tools import * -from tests.protocol.mockwiredata import * -from azurelinuxagent.exception import * -from azurelinuxagent.distro.loader import get_distro -from azurelinuxagent.protocol.restapi import get_properties -from azurelinuxagent.protocol.wire import WireProtocol - -@patch("time.sleep") -@patch("azurelinuxagent.protocol.wire.CryptUtil") -@patch("azurelinuxagent.utils.restutil.http_get") -class TestExtension(AgentTestCase): - - def _assert_handler_status(self, report_vm_status, expected_status, - expected_ext_count, version): - self.assertTrue(report_vm_status.called) - args, kw = report_vm_status.call_args - vm_status = args[0] - self.assertNotEquals(0, len(vm_status.vmAgent.extensionHandlers)) - handler_status = vm_status.vmAgent.extensionHandlers[0] - self.assertEquals(expected_status, handler_status.status) - self.assertEquals("OSTCExtensions.ExampleHandlerLinux", - handler_status.name) - self.assertEquals(version, handler_status.version) - self.assertEquals(expected_ext_count, len(handler_status.extensions)) - - def _assert_no_handler_status(self, report_vm_status): - self.assertTrue(report_vm_status.called) - args, kw = report_vm_status.call_args - vm_status = args[0] - self.assertEquals(0, len(vm_status.vmAgent.extensionHandlers)) - - def _create_mock(self, test_data, mock_http_get, MockCryptUtil, _): - """Test enable/disable/unistall of an extension""" - distro = get_distro() - - #Mock protocol to return test data - mock_http_get.side_effect = test_data.mock_http_get - MockCryptUtil.side_effect = test_data.mock_crypt_util - - protocol = WireProtocol("foo.bar") - protocol.detect() - protocol.report_ext_status = MagicMock() - protocol.report_vm_status = MagicMock() - distro.protocol_util.get_protocol = Mock(return_value=protocol) - - return distro, protocol - - def test_ext_handler(self, *args): - test_data = WireProtocolData(DATA_FILE) - distro, protocol = self._create_mock(test_data, *args) - - #Test enable scenario. - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - self._assert_ext_status(protocol.report_ext_status, "success", 0) - - #Test goal state not changed - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - - #Test goal state changed - test_data.goal_state = test_data.goal_state.replace("<Incarnation>1<", - "<Incarnation>2<") - test_data.ext_conf = test_data.ext_conf.replace("seqNo=\"0\"", - "seqNo=\"1\"") - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - self._assert_ext_status(protocol.report_ext_status, "success", 1) - - #Test upgrade - test_data.goal_state = test_data.goal_state.replace("<Incarnation>2<", - "<Incarnation>3<") - test_data.ext_conf = test_data.ext_conf.replace("1.0", "1.1") - test_data.ext_conf = test_data.ext_conf.replace("seqNo=\"1\"", - "seqNo=\"2\"") - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.1") - self._assert_ext_status(protocol.report_ext_status, "success", 2) - - #Test disable - test_data.goal_state = test_data.goal_state.replace("<Incarnation>3<", - "<Incarnation>4<") - test_data.ext_conf = test_data.ext_conf.replace("enabled", "disabled") - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "NotReady", - 1, "1.1") - - #Test uninstall - test_data.goal_state = test_data.goal_state.replace("<Incarnation>4<", - "<Incarnation>5<") - test_data.ext_conf = test_data.ext_conf.replace("disabled", "uninstall") - distro.ext_handlers_handler.run() - self._assert_no_handler_status(protocol.report_vm_status) - - #Test uninstall again! - test_data.goal_state = test_data.goal_state.replace("<Incarnation>5<", - "<Incarnation>6<") - distro.ext_handlers_handler.run() - self._assert_no_handler_status(protocol.report_vm_status) - - def test_ext_handler_no_settings(self, *args): - test_data = WireProtocolData(DATA_FILE_EXT_NO_SETTINGS) - distro, protocol = self._create_mock(test_data, *args) - - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 0, "1.0") - - def test_ext_handler_no_public_settings(self, *args): - test_data = WireProtocolData(DATA_FILE_EXT_NO_PUBLIC) - distro, protocol = self._create_mock(test_data, *args) - - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - - def test_ext_handler_no_ext(self, *args): - test_data = WireProtocolData(DATA_FILE_NO_EXT) - distro, protocol = self._create_mock(test_data, *args) - - #Assert no extension handler status - distro.ext_handlers_handler.run() - self._assert_no_handler_status(protocol.report_vm_status) - - @patch('azurelinuxagent.distro.default.extension.add_event') - def test_ext_handler_download_failure(self, mock_add_event, *args): - test_data = WireProtocolData(DATA_FILE) - distro, protocol = self._create_mock(test_data, *args) - protocol.download_ext_handler_pkg = Mock(side_effect=ProtocolError) - - distro.ext_handlers_handler.run() - args, kw = mock_add_event.call_args - self.assertEquals(False, kw['is_success']) - self.assertEquals("OSTCExtensions.ExampleHandlerLinux", kw['name']) - self.assertEquals("Download", kw['op']) - - @patch('azurelinuxagent.distro.default.extension.fileutil') - def test_ext_handler_io_error(self, mock_fileutil, *args): - test_data = WireProtocolData(DATA_FILE) - distro, protocol = self._create_mock(test_data, *args) - - mock_fileutil.write_file.return_value = IOError("Mock IO Error") - distro.ext_handlers_handler.run() - - def _assert_ext_status(self, report_ext_status, expected_status, - expected_seq_no): - self.assertTrue(report_ext_status.called) - args, kw = report_ext_status.call_args - ext_status = args[-1] - self.assertEquals(expected_status, ext_status.status) - self.assertEquals(expected_seq_no, ext_status.sequenceNumber) - - def test_ext_handler_no_reporting_status(self, *args): - test_data = WireProtocolData(DATA_FILE) - distro, protocol = self._create_mock(test_data, *args) - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - - #Remove status file and re-run collecting extension status - status_file = os.path.join(self.tmp_dir, - "OSTCExtensions.ExampleHandlerLinux-1.0", - "status", "0.status") - self.assertTrue(os.path.isfile(status_file)) - os.remove(status_file) - - distro.ext_handlers_handler.run() - self._assert_handler_status(protocol.report_vm_status, "Ready", 1, "1.0") - self._assert_ext_status(protocol.report_ext_status, "error", 0) - - -if __name__ == '__main__': - unittest.main() - diff --git a/tests/distro/test_monitor.py b/tests/distro/test_monitor.py deleted file mode 100644 index 1dd7740..0000000 --- a/tests/distro/test_monitor.py +++ /dev/null @@ -1,32 +0,0 @@ -# Copyright 2014 Microsoft Corporation -# -# 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. -# -# Requires Python 2.4+ and Openssl 1.0+ -# -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx - -from tests.tools import * -from azurelinuxagent.exception import * -from azurelinuxagent.distro.default.monitor import * - -class TestMonitor(AgentTestCase): - def test_parse_xml_event(self): - data_str = load_data('ext/event.xml') - event = parse_xml_event(data_str) - self.assertNotEquals(None, event) - self.assertNotEquals(0, event.parameters) - self.assertNotEquals(None, event.parameters[0]) - diff --git a/tests/distro/test_protocol_util.py b/tests/distro/test_protocol_util.py deleted file mode 100644 index 61339f3..0000000 --- a/tests/distro/test_protocol_util.py +++ /dev/null @@ -1,89 +0,0 @@ -# Copyright 2014 Microsoft Corporation -# -# 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. -# -# Requires Python 2.4+ and Openssl 1.0+ -# -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx - -from tests.tools import * -from azurelinuxagent.distro.loader import get_distro -from azurelinuxagent.exception import * -from azurelinuxagent.distro.default.protocolUtil import * - -@patch("time.sleep") -class TestProtocolUtil(AgentTestCase): - - @distros() - @patch("azurelinuxagent.distro.default.protocolUtil.MetadataProtocol") - @patch("azurelinuxagent.distro.default.protocolUtil.WireProtocol") - def test_detect_protocol(self, distro_name, distro_version, distro_full_name, - WireProtocol, MetadataProtocol, _, *distro_args): - - WireProtocol.return_value = MagicMock() - MetadataProtocol.return_value = MagicMock() - - distro = get_distro(distro_name, distro_version, distro_full_name) - distro.dhcp_handler = MagicMock() - distro.dhcp_handler.endpoint = "foo.bar" - - #Test wire protocol is available - protocol = distro.protocol_util.detect_protocol() - self.assertEquals(WireProtocol.return_value, protocol) - - #Test wire protocol is not available - distro.protocol_util.protocol = None - WireProtocol.side_effect = ProtocolError() - - protocol = distro.protocol_util.detect_protocol() - self.assertEquals(MetadataProtocol.return_value, protocol) - - #Test no protocol is available - distro.protocol_util.protocol = None - WireProtocol.side_effect = ProtocolError() - MetadataProtocol.side_effect = ProtocolError() - self.assertRaises(ProtocolError, distro.protocol_util.detect_protocol) - - @distros() - def test_detect_protocol_by_file(self, distro_name, distro_version, - distro_full_name, _): - distro = get_distro(distro_name, distro_version, distro_full_name) - protocol_util = distro.protocol_util - - protocol_util._detect_wire_protocol = Mock() - protocol_util._detect_metadata_protocol = Mock() - - tag_file = os.path.join(self.tmp_dir, TAG_FILE_NAME) - - #Test tag file doesn't exist - protocol_util.detect_protocol_by_file() - protocol_util._detect_wire_protocol.assert_any_call() - protocol_util._detect_metadata_protocol.assert_not_called() - - #Test tag file exists - protocol_util.protocol = None - protocol_util._detect_wire_protocol.reset_mock() - protocol_util._detect_metadata_protocol.reset_mock() - with open(tag_file, "w+") as tag_fd: - tag_fd.write("") - - protocol_util.detect_protocol_by_file() - protocol_util._detect_metadata_protocol.assert_any_call() - protocol_util._detect_wire_protocol.assert_not_called() - - -if __name__ == '__main__': - unittest.main() - diff --git a/tests/distro/test_provision.py b/tests/distro/test_provision.py deleted file mode 100644 index 60249ce..0000000 --- a/tests/distro/test_provision.py +++ /dev/null @@ -1,47 +0,0 @@ -# Copyright 2014 Microsoft Corporation -# -# 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. -# -# Requires Python 2.4+ and Openssl 1.0+ -# -# Implements parts of RFC 2131, 1541, 1497 and -# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx -# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx - -from tests.tools import * -from azurelinuxagent.distro.loader import get_distro -from azurelinuxagent.distro.default.protocolUtil import * -import azurelinuxagent.utils.fileutil as fileutil - -class TestProvision(AgentTestCase): - - @distros("redhat") - def test_provision(self, distro_name, distro_version, distro_full_name): - distro = get_distro(distro_name, distro_version, distro_full_name) - distro.osutil = MagicMock() - distro.osutil.decode_customdata = Mock(return_value="") - - distro.protocol_util.detect_protocol_by_file = MagicMock() - distro.protocol_util.get_protocol = MagicMock() - conf.get_dvd_mount_point = Mock(return_value=self.tmp_dir) - - ovfenv_file = os.path.join(self.tmp_dir, OVF_FILE_NAME) - ovfenv_data = load_data("ovf-env.xml") - fileutil.write_file(ovfenv_file, ovfenv_data) - - handler = distro.provision_handler - handler.run() - -if __name__ == '__main__': - unittest.main() - diff --git a/tests/distro/test_loader.py b/tests/distro/test_resourceDisk.py index 94ca913..198fd49 100644 --- a/tests/distro/test_loader.py +++ b/tests/distro/test_resourceDisk.py @@ -19,18 +19,24 @@ # http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx from tests.tools import * -from azurelinuxagent.distro.loader import get_distro -from azurelinuxagent.distro.default.distro import DefaultDistro +from azurelinuxagent.daemon.resourcedisk import get_resourcedisk_handler -class TestDistroLoader(AgentTestCase): +class TestResourceDisk(AgentTestCase): + def test_mkfile(self): + # setup + test_file = os.path.join(self.tmp_dir, 'test_file') + file_size = 1024 * 128 + if os.path.exists(test_file): + os.remove(test_file) - @distros() - def test_distro_loader(self, *distro_args): - distro = get_distro(*distro_args) - self.assertNotEquals(None, distro) - self.assertNotEquals(DefaultDistro, type(distro)) - + # execute + get_resourcedisk_handler().mkfile(test_file, file_size) + + # assert + assert os.path.exists(test_file) + + # cleanup + os.remove(test_file) if __name__ == '__main__': unittest.main() - diff --git a/tests/distro/test_scvmm.py b/tests/distro/test_scvmm.py new file mode 100644 index 0000000..2437170 --- /dev/null +++ b/tests/distro/test_scvmm.py @@ -0,0 +1,84 @@ +# Copyright 2014 Microsoft Corporation +# +# 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. +# +# Requires Python 2.4+ and Openssl 1.0+ +# +# Implements parts of RFC 2131, 1541, 1497 and +# http://msdn.microsoft.com/en-us/library/cc227282%28PROT.10%29.aspx +# http://msdn.microsoft.com/en-us/library/cc227259%28PROT.13%29.aspx + +import mock +from tests.tools import * + +import azurelinuxagent.daemon.scvmm as scvmm +from azurelinuxagent.daemon.main import * +from azurelinuxagent.common.osutil.default import DefaultOSUtil + +class TestSCVMM(AgentTestCase): + def test_scvmm_detection_with_file(self): + # setup + conf.get_dvd_mount_point = Mock(return_value=self.tmp_dir) + conf.get_detect_scvmm_env = Mock(return_value=True) + scvmm_file = os.path.join(self.tmp_dir, scvmm.VMM_CONF_FILE_NAME) + fileutil.write_file(scvmm_file, "") + + with patch.object(scvmm.ScvmmHandler, 'start_scvmm_agent') as po: + with patch('os.listdir', return_value=["sr0", "sr1", "sr2"]): + with patch('time.sleep', return_value=0): + # execute + failed = False + try: + scvmm.get_scvmm_handler().run() + except: + failed = True + # assert + self.assertTrue(failed) + self.assertTrue(po.call_count == 1) + # cleanup + os.remove(scvmm_file) + + + def test_scvmm_detection_with_multiple_cdroms(self): + # setup + conf.get_dvd_mount_point = Mock(return_value=self.tmp_dir) + conf.get_detect_scvmm_env = Mock(return_value=True) + + # execute + with mock.patch.object(DefaultOSUtil, 'mount_dvd') as patch_mount: + with patch('os.listdir', return_value=["sr0", "sr1", "sr2"]): + scvmm.ScvmmHandler().detect_scvmm_env() + # assert + assert patch_mount.call_count == 3 + assert patch_mount.call_args_list[0][1]['dvd_device'] == '/dev/sr0' + assert patch_mount.call_args_list[1][1]['dvd_device'] == '/dev/sr1' + assert patch_mount.call_args_list[2][1]['dvd_device'] == '/dev/sr2' + + + def test_scvmm_detection_without_file(self): + # setup + conf.get_dvd_mount_point = Mock(return_value=self.tmp_dir) + conf.get_detect_scvmm_env = Mock(return_value=True) + scvmm_file = os.path.join(self.tmp_dir, scvmm.VMM_CONF_FILE_NAME) + if os.path.exists(scvmm_file): + os.remove(scvmm_file) + + with mock.patch.object(scvmm.ScvmmHandler, 'start_scvmm_agent') as patch_start: + # execute + scvmm.ScvmmHandler().detect_scvmm_env() + # assert + patch_start.assert_not_called() + + +if __name__ == '__main__': + unittest.main() |