| #!/usr/bin/env python2.6 |
| |
| ''' |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you 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. |
| ''' |
| |
| from unittest import TestCase |
| import unittest |
| import StringIO |
| import socket |
| import os, sys |
| from mock.mock import patch |
| from mock.mock import MagicMock |
| from mock.mock import create_autospec |
| import os, errno, tempfile |
| from ambari_agent import StackVersionsFileHandler |
| import logging |
| |
| stackVersionsFileHandler = \ |
| StackVersionsFileHandler.StackVersionsFileHandler("/tmp") |
| dummyVersionsFile = os.path.join('dummy_files', 'dummy_current_stack') |
| |
| class TestStackVersionsFileHandler(TestCase): |
| |
| logger = logging.getLogger() |
| |
| @patch.object(stackVersionsFileHandler, 'touch_file') |
| def test_read_stack_version(self, touch_method): |
| stackVersionsFileHandler.versionsFilePath = dummyVersionsFile |
| result = stackVersionsFileHandler.read_stack_version("NAGIOS_SERVER") |
| self.assertEquals(result, '{"stackName":"HDP","stackVersion":"1.2.1"}') |
| result = stackVersionsFileHandler.read_stack_version("GANGLIA_SERVER") |
| self.assertEquals(result, '{"stackName":"HDP","stackVersion":"1.2.2"}') |
| result = stackVersionsFileHandler.read_stack_version("NOTEXISTING") |
| self.assertEquals(result, stackVersionsFileHandler.DEFAULT_VER) |
| self.assertTrue(touch_method.called) |
| |
| |
| @patch.object(stackVersionsFileHandler, 'touch_file') |
| def test_read_all_stack_versions(self, touch_method): |
| stackVersionsFileHandler.versionsFilePath = dummyVersionsFile |
| result = stackVersionsFileHandler.read_all_stack_versions() |
| self.assertEquals(len(result.keys()), 4) |
| self.assertEquals(result["NAGIOS_SERVER"], |
| '{"stackName":"HDP","stackVersion":"1.2.1"}') |
| self.assertEquals(result["HCATALOG"], |
| '{"stackName":"HDP","stackVersion":"1.2.2"}') |
| self.assertTrue(touch_method.called) |
| |
| |
| def test_extract(self): |
| s = ' NAGIOS_SERVER \t {"stackName":"HDP","stackVersion":"1.3.0"} ' |
| comp, ver = stackVersionsFileHandler.extract(s) |
| self.assertEqual(comp, "NAGIOS_SERVER") |
| self.assertEqual(ver, '{"stackName":"HDP","stackVersion":"1.3.0"}') |
| # testing wrong value |
| s = " NAGIOS_SERVER " |
| comp, ver = stackVersionsFileHandler.extract(s) |
| self.assertEqual(comp, stackVersionsFileHandler.DEFAULT_VER) |
| self.assertEqual(ver, stackVersionsFileHandler.DEFAULT_VER) |
| |
| |
| def test_touch_file(self): |
| tmpfile = tempfile.mktemp() |
| stackVersionsFileHandler.versionsFilePath = tmpfile |
| stackVersionsFileHandler.touch_file() |
| result = os.path.isfile(tmpfile) |
| self.assertEqual(result, True) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |