diff options
Diffstat (limited to 'tests/utils')
| -rw-r--r-- | tests/utils/test_file_util.py | 57 | ||||
| -rw-r--r-- | tests/utils/test_rest_util.py | 85 | ||||
| -rw-r--r-- | tests/utils/test_text_util.py | 21 |
3 files changed, 114 insertions, 49 deletions
diff --git a/tests/utils/test_file_util.py b/tests/utils/test_file_util.py index 9a5479e..f16f409 100644 --- a/tests/utils/test_file_util.py +++ b/tests/utils/test_file_util.py @@ -24,8 +24,9 @@ from azurelinuxagent.common.future import ustr import azurelinuxagent.common.utils.fileutil as fileutil class TestFileOperations(AgentTestCase): + def test_read_write_file(self): - test_file=os.path.join(self.tmp_dir, 'test_file') + test_file=os.path.join(self.tmp_dir, self.test_file) content = ustr(uuid.uuid4()) fileutil.write_file(test_file, content) @@ -34,7 +35,7 @@ class TestFileOperations(AgentTestCase): os.remove(test_file) def test_rw_utf8_file(self): - test_file=os.path.join(self.tmp_dir, 'test_file') + test_file=os.path.join(self.tmp_dir, self.test_file) content = u"\u6211" fileutil.write_file(test_file, content, encoding="utf-8") @@ -43,14 +44,14 @@ class TestFileOperations(AgentTestCase): os.remove(test_file) def test_remove_bom(self): - test_file=os.path.join(self.tmp_dir, 'test_file') + test_file=os.path.join(self.tmp_dir, self.test_file) data = b'\xef\xbb\xbfhehe' fileutil.write_file(test_file, data, asbin=True) data = fileutil.read_file(test_file, remove_bom=True) self.assertNotEquals(0xbb, ord(data[0])) def test_append_file(self): - test_file=os.path.join(self.tmp_dir, 'test_file') + test_file=os.path.join(self.tmp_dir, self.test_file) content = ustr(uuid.uuid4()) fileutil.append_file(test_file, content) @@ -68,5 +69,53 @@ class TestFileOperations(AgentTestCase): filename = fileutil.base_name(filepath) self.assertEquals('abc', filename) + def test_remove_files(self): + import random + import string + import glob + random_word = lambda : ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(5)) + + #Create 10 test files + test_file = os.path.join(self.tmp_dir, self.test_file) + test_file2 = os.path.join(self.tmp_dir, 'another_file') + test_files = [test_file + random_word() for _ in range(5)] + \ + [test_file2 + random_word() for _ in range(5)] + for file in test_files: + open(file, 'a').close() + + #Remove files using fileutil.rm_files + test_file_pattern = test_file + '*' + test_file_pattern2 = test_file2 + '*' + fileutil.rm_files(test_file_pattern, test_file_pattern2) + + self.assertEqual(0, len(glob.glob(os.path.join(self.tmp_dir, test_file_pattern)))) + self.assertEqual(0, len(glob.glob(os.path.join(self.tmp_dir, test_file_pattern2)))) + + def test_get_all_files(self): + import random + import string + random_word = lambda: ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(5)) + + # Create 10 test files at the root dir and 10 other in the sub dir + test_file = os.path.join(self.tmp_dir, self.test_file) + test_file2 = os.path.join(self.tmp_dir, 'another_file') + expected_files = [test_file + random_word() for _ in range(5)] + \ + [test_file2 + random_word() for _ in range(5)] + + test_subdir = os.path.join(self.tmp_dir, 'test_dir') + os.mkdir(test_subdir) + test_file_in_subdir = os.path.join(test_subdir, self.test_file) + test_file_in_subdir2 = os.path.join(test_subdir, 'another_file') + expected_files.extend([test_file_in_subdir + random_word() for _ in range(5)] + \ + [test_file_in_subdir2 + random_word() for _ in range(5)]) + + for file in expected_files: + open(file, 'a').close() + + # Get All files using fileutil.get_all_files + actual_files = fileutil.get_all_files(self.tmp_dir) + + self.assertEqual(set(expected_files), set(actual_files)) + if __name__ == '__main__': unittest.main() diff --git a/tests/utils/test_rest_util.py b/tests/utils/test_rest_util.py index 874e527..5f084a6 100644 --- a/tests/utils/test_rest_util.py +++ b/tests/utils/test_rest_util.py @@ -15,16 +15,13 @@ # Requires Python 2.4+ and Openssl 1.0+ # -from tests.tools import AgentTestCase, patch, Mock, MagicMock -import uuid import unittest -import os import azurelinuxagent.common.utils.restutil as restutil -from azurelinuxagent.common.future import ustr, httpclient -import azurelinuxagent.common.logger as logger +from azurelinuxagent.common.future import httpclient +from tests.tools import AgentTestCase, patch, Mock, MagicMock -class TestHttpOperations(AgentTestCase): +class TestHttpOperations(AgentTestCase): def test_parse_url(self): test_uri = "http://abc.def/ghi#hash?jkl=mn" host, port, secure, rel_uri = restutil._parse_url(test_uri) @@ -36,11 +33,11 @@ class TestHttpOperations(AgentTestCase): self.assertEquals("abc.def", host) self.assertEquals("/", rel_uri) self.assertEquals(False, secure) - + test_uri = "https://abc.def/ghi?jkl=mn" host, port, secure, rel_uri = restutil._parse_url(test_uri) self.assertEquals(True, secure) - + test_uri = "http://abc.def:80/" host, port, secure, rel_uri = restutil._parse_url(test_uri) self.assertEquals("abc.def", host) @@ -53,71 +50,71 @@ class TestHttpOperations(AgentTestCase): self.assertEquals(None, host) self.assertEquals(rel_uri, "None") - @patch("azurelinuxagent.common.future.httpclient.HTTPSConnection") @patch("azurelinuxagent.common.future.httpclient.HTTPConnection") def test_http_request(self, HTTPConnection, HTTPSConnection): - mock_httpconn = MagicMock() - mock_httpresp = MagicMock() - mock_httpconn.getresponse = Mock(return_value=mock_httpresp) - HTTPConnection.return_value = mock_httpconn - HTTPSConnection.return_value = mock_httpconn - - mock_httpresp.read = Mock(return_value="_(:3| <)_") - - #Test http get + mock_http_conn = MagicMock() + mock_http_resp = MagicMock() + mock_http_conn.getresponse = Mock(return_value=mock_http_resp) + HTTPConnection.return_value = mock_http_conn + HTTPSConnection.return_value = mock_http_conn + + mock_http_resp.read = Mock(return_value="_(:3| <)_") + + # Test http get resp = restutil._http_request("GET", "foo", "bar") self.assertNotEquals(None, resp) self.assertEquals("_(:3| <)_", resp.read()) - - #Test https get + + # Test https get resp = restutil._http_request("GET", "foo", "bar", secure=True) self.assertNotEquals(None, resp) self.assertEquals("_(:3| <)_", resp.read()) - - #Test http get with proxy - mock_httpresp.read = Mock(return_value="_(:3| <)_") + + # Test http get with proxy + mock_http_resp.read = Mock(return_value="_(:3| <)_") resp = restutil._http_request("GET", "foo", "bar", proxy_host="foo.bar", proxy_port=23333) self.assertNotEquals(None, resp) self.assertEquals("_(:3| <)_", resp.read()) - - #Test https get + + # Test https get resp = restutil._http_request("GET", "foo", "bar", secure=True) self.assertNotEquals(None, resp) self.assertEquals("_(:3| <)_", resp.read()) - - #Test https get with proxy - mock_httpresp.read = Mock(return_value="_(:3| <)_") + + # Test https get with proxy + mock_http_resp.read = Mock(return_value="_(:3| <)_") resp = restutil._http_request("GET", "foo", "bar", proxy_host="foo.bar", proxy_port=23333, secure=True) self.assertNotEquals(None, resp) self.assertEquals("_(:3| <)_", resp.read()) - + @patch("time.sleep") @patch("azurelinuxagent.common.utils.restutil._http_request") def test_http_request_with_retry(self, _http_request, sleep): - mock_httpresp = MagicMock() - mock_httpresp.read = Mock(return_value="hehe") - _http_request.return_value = mock_httpresp + mock_http_resp = MagicMock() + mock_http_resp.read = Mock(return_value="hehe") + _http_request.return_value = mock_http_resp - #Test http get - resp = restutil.http_get("http://foo.bar") + # Test http get + resp = restutil.http_get("http://foo.bar") self.assertEquals("hehe", resp.read()) - #Test https get - resp = restutil.http_get("https://foo.bar") + # Test https get + resp = restutil.http_get("https://foo.bar") self.assertEquals("hehe", resp.read()) - - #Test http failure + + # Test http failure _http_request.side_effect = httpclient.HTTPException("Http failure") - self.assertRaises(restutil.HttpError, restutil.http_get, "http://foo.bar") + self.assertRaises(restutil.HttpError, restutil.http_get, + "http://foo.bar") - #Test http failure + # Test http failure _http_request.side_effect = IOError("IO failure") - self.assertRaises(restutil.HttpError, restutil.http_get, "http://foo.bar") - -if __name__ == '__main__': - unittest.main() + self.assertRaises(restutil.HttpError, restutil.http_get, + "http://foo.bar") + + if __name__ == '__main__': unittest.main() diff --git a/tests/utils/test_text_util.py b/tests/utils/test_text_util.py index 9ac0707..dc3de85 100644 --- a/tests/utils/test_text_util.py +++ b/tests/utils/test_text_util.py @@ -35,12 +35,31 @@ class TestTextUtil(AgentTestCase): data = ustr(b'\xef\xbb\xbfhehe', encoding='utf-8') data = textutil.remove_bom(data) self.assertNotEquals(0xbb, data[0]) - + + #bom is comprised of a sequence of three bytes and ff length of the input is shorter + # than three bytes, remove_bom should not do anything + data = u"\xa7" + data = textutil.remove_bom(data) + self.assertEquals(data, data[0]) + + data = u"\xa7\xef" + data = textutil.remove_bom(data) + self.assertEquals(u"\xa7", data[0]) + self.assertEquals(u"\xef", data[1]) + #Test string without BOM is not affected data = u"hehe" data = textutil.remove_bom(data) self.assertEquals(u"h", data[0]) + data = u"" + data = textutil.remove_bom(data) + self.assertEquals(u"", data) + + data = u" " + data = textutil.remove_bom(data) + self.assertEquals(u" ", data) + def test_version_compare(self): self.assertTrue(Version("1.0") < Version("1.1")) self.assertTrue(Version("1.9") < Version("1.10")) |