| #!/usr/bin/python |
| # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for the cb_url_lib module.""" |
| |
| import cb_url_lib |
| import formatter |
| import logging |
| import mox |
| import os |
| import unittest |
| import urllib |
| import tempfile |
| |
| from cb_constants import BundlingError, IMAGE_GSD_BUCKET, IMAGE_GSD_PREFIX |
| from cb_util import CommandResult |
| |
| from mox import IsA |
| |
| _LINK_NAME = 'href_atrribute/of_a_link' |
| |
| |
| class UrlListerTest(unittest.TestCase): |
| """Unit tests for the UrlLister class.""" |
| |
| def setUp(self): |
| self.givenformatter = formatter.NullFormatter() |
| |
| def testReset(self): |
| """Tests the reset method of class UrlLister.""" |
| with open('testdata/test_page_one_link.html', 'r') as page: |
| parser = cb_url_lib.UrlLister(self.givenformatter) |
| parser.urls = ['a_fake_link', 'another_fake_link'] |
| parser.reset() |
| parser.feed(page.read()) |
| parser.close() |
| self.assertEqual([_LINK_NAME], parser.urls) |
| |
| def testStart_aNoLinks(self): |
| """Verify behavior when no links exist on given page.""" |
| with open('testdata/test_page_no_links.html', 'r') as page: |
| parser = cb_url_lib.UrlLister(self.givenformatter) |
| parser.feed(page.read()) |
| parser.close() |
| self.assertEqual([], parser.urls) |
| |
| def testStart_aOneLink(self): |
| """Verify parsing when one link present.""" |
| with open('testdata/test_page_one_link.html', 'r') as page: |
| parser = cb_url_lib.UrlLister(self.givenformatter) |
| parser.feed(page.read()) |
| parser.close() |
| self.assertEqual([_LINK_NAME], parser.urls) |
| |
| def testStart_aManyLinks(self): |
| """Verify parsing when many links present.""" |
| with open('testdata/test_page_many_links.html', 'r') as page: |
| parser = cb_url_lib.UrlLister(self.givenformatter) |
| parser.feed(page.read()) |
| parser.close() |
| # sample page links taken from real chromeos-images index page |
| expected = ['/stable-channel/x86-alex/', |
| 'ChromeOS-0.12.433.269-r72d7eaa2-b198-x86-alex.zip', |
| 'ChromeOS-factory-0.12.433.269-r72d7eaa2-b198-x86-alex.zip'] |
| actual = parser.urls |
| self.assertEqual(expected, actual) |
| |
| |
| class TestDetermineUrl(mox.MoxTestBase): |
| """Unit tests related to DetermineUrl.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(urllib, 'urlopen') |
| self.mox.StubOutWithMock(cb_url_lib, 'RunCommand') |
| self.url = 'test_url' |
| self.pattern = ['chromeos', '.zip'] |
| self.http_url = IMAGE_GSD_PREFIX + '/test_url' |
| self.gsd_url = IMAGE_GSD_BUCKET + '/test_url/*' |
| self.gsd_pattern = ['chromeos', '.bin'] |
| |
| def testHttpUrlGood(self): |
| """Verify URL returned when page opens properly.""" |
| with open('testdata/test_page_many_links.html', 'r') as test_page: |
| urllib.urlopen('test_url').AndReturn(test_page) |
| self.mox.ReplayAll() |
| expected = 'test_url/ChromeOS-0.12.433.269-r72d7eaa2-b198-x86-alex.zip' |
| actual = cb_url_lib.DetermineUrl(self.url, self.pattern) |
| self.assertEqual(expected, actual) |
| |
| def testHttpUrlBad(self): |
| """Verify None returned when page fails to open properly.""" |
| urllib.urlopen('test_url').AndRaise(IOError) |
| self.mox.ReplayAll() |
| actual = cb_url_lib.DetermineUrl(self.url, self.pattern) |
| self.assertEqual(None, actual) |
| |
| def testHttpUrlNoMatch(self): |
| """Verify None returned when MatchUrl() returns None.""" |
| with open('testdata/test_page_no_links.html', 'r') as test_page: |
| urllib.urlopen('test_url').AndReturn(test_page) |
| self.mox.ReplayAll() |
| actual = cb_url_lib.DetermineUrl(self.url, self.pattern) |
| self.assertEqual(None, actual) |
| |
| def testGsdUrlGood(self): |
| """Verify GSD URL returned when page opens properly.""" |
| with open('testdata/test_page_gsd_links.html', 'r') as test_page: |
| test_result = CommandResult() |
| test_result.output = test_page.read() |
| test_result.returncode = 0 |
| |
| cb_url_lib.RunCommand(['gsutil', 'ls', self.gsd_url], redirect_stdout=True, |
| redirect_stderr=True).AndReturn(test_result) |
| self.mox.ReplayAll() |
| expected = ( |
| 'gs://chromeos-releases/stable-channel/x86-alex/0.12.433.269/' |
| 'chromeos_0.12.433.269_x86-alex_recovery_stable-channel_mp-v3.bin') |
| actual = cb_url_lib.DetermineUrl(self.http_url, self.gsd_pattern) |
| self.assertEqual(expected, actual) |
| |
| def testGsdUrlRunCommandError(self): |
| """Verify None returned when RunCommand returns non-zero code.""" |
| test_result = CommandResult() |
| test_result.returncode = -1 |
| |
| cb_url_lib.RunCommand(['gsutil', 'ls', self.gsd_url], redirect_stdout=True, |
| redirect_stderr=True).AndReturn(test_result) |
| self.mox.ReplayAll() |
| actual = cb_url_lib.DetermineUrl(self.http_url, self.gsd_pattern) |
| self.assertEqual(None, actual) |
| |
| def testGsdUrlNoMatch(self): |
| """Verify None returned when MatchUrl() returns None.""" |
| with open('testdata/test_page_many_links.html', 'r') as test_page: |
| test_result = CommandResult() |
| test_result.output = test_page.read() |
| test_result.returncode = 0 |
| |
| cb_url_lib.RunCommand(['gsutil', 'ls', self.gsd_url], redirect_stdout=True, |
| redirect_stderr=True).AndReturn(test_result) |
| self.mox.ReplayAll() |
| actual = cb_url_lib.DetermineUrl(self.http_url, self.gsd_pattern) |
| self.assertEqual(None, actual) |
| |
| |
| class TestMatchUrl(unittest.TestCase): |
| """Unit tests related to MatchUrl.""" |
| |
| def setUp(self): |
| self.token_list = ['abc', '.bin'] |
| self.filename = 'abcdefghi.bin' |
| |
| def testNoMatch(self): |
| """Verify behavior when no match exists.""" |
| url_list = ['abd.bin', 'acd.bin'] |
| actual = cb_url_lib.MatchUrl(url_list, self.token_list) |
| self.assertEqual(None, actual) |
| |
| def testOneMatch(self): |
| """Verify string returned upon a single good match.""" |
| url_list = ['abd.bin', 'acd.bin', self.filename] |
| actual = cb_url_lib.MatchUrl(url_list, self.token_list) |
| self.assertEqual(self.filename, actual) |
| |
| def testManyMatches(self): |
| """verify string returned upon multiple good matches.""" |
| url_list = [self.filename, 'abc.bin', 'bdfhabc.bin'] |
| actual = cb_url_lib.MatchUrl(url_list, self.token_list) |
| self.assertEqual(self.filename, actual) |
| |
| def testMatchUrlEmptyUrl(self): |
| """Verify behavior when input url_list is empty.""" |
| self.assertFalse(cb_url_lib.MatchUrl([], self.token_list)) |
| |
| def testMatchUrlEmptyTokenList(self): |
| """Verify behavior when input token_list is empty.""" |
| self.assertFalse(cb_url_lib.MatchUrl(['url'], [])) |
| |
| def testMatchUrlWithFullUrl(self): |
| """Verify good match when input url is a full path.""" |
| test_url = os.path.join('http://domain.com', self.filename) |
| self.assertTrue(cb_url_lib.MatchUrl([test_url], self.token_list)) |
| |
| def testMatchUrlWithFilenameOnly(self): |
| """Verify good match when input url is a filename (no path info).""" |
| self.assertTrue(cb_url_lib.MatchUrl([self.filename], self.token_list)) |
| |
| def testMatchUrlWithFilenameOnlyMixedCase(self): |
| """Verify good match when input url is a filename (no path info).""" |
| self.filename = 'aBcdEfgHI.biN' |
| self.assertTrue(cb_url_lib.MatchUrl([self.filename], self.token_list)) |
| |
| def testMatchUrlMissingStartToken(self): |
| """Verify no match when token_list lacks beginning of filename.""" |
| self.token_list[0] = 'ghi' |
| self.assertFalse(cb_url_lib.MatchUrl([self.filename], self.token_list)) |
| |
| def testMatchUrlMissingEndToken(self): |
| """Verify no match when token_list lacks end of filename.""" |
| self.token_list[1] = '.bi' |
| self.assertFalse(cb_url_lib.MatchUrl([self.filename], self.token_list)) |
| |
| |
| class TestDownload(mox.MoxTestBase): |
| """Unit tests realted to Download.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(urllib, 'urlopen') |
| self.mox.StubOutWithMock(cb_url_lib, 'RunCommand') |
| self.url = 'test_url' |
| self.gsd_url = IMAGE_GSD_BUCKET + self.url |
| self.named_file = tempfile.NamedTemporaryFile() |
| self.test_file = tempfile.TemporaryFile() |
| self.test_file.write('Some sample content for testing.') |
| self.test_file.seek(0) # must rewind file handle for read |
| # Stub out os.path AFTER creating temp files |
| self.mox.StubOutWithMock(os.path, 'join') |
| |
| def testUrlGoodLocalFileOpenSucceeds(self): |
| """Verify return value when page opens properly.""" |
| os.path.join(IsA(str), IsA(str)).AndReturn(self.named_file.name) |
| urllib.urlopen('test_url').AndReturn(self.test_file) |
| self.mox.ReplayAll() |
| self.assertTrue(cb_url_lib.Download(self.url)) |
| |
| def testUrlBad(self): |
| """Verify clean return value when page does not open properly.""" |
| os.path.join(IsA(str), IsA(str)).AndReturn('') |
| urllib.urlopen('test_url').AndRaise(IOError) |
| self.mox.ReplayAll() |
| expected = False |
| actual = cb_url_lib.Download(self.url) |
| self.assertEqual(expected, actual) |
| |
| def testLocalFileOpenFails(self): |
| """Verify clean return value when local file fails to open.""" |
| os.path.join(IsA(str), IsA(str)).AndReturn('') |
| urllib.urlopen('test_url').AndReturn(self.test_file) |
| self.mox.ReplayAll() |
| self.assertFalse(cb_url_lib.Download(self.url)) |
| |
| def testGsdUrlGoodLocalFileOpenSucceeds(self): |
| """Verify return value when GSD URL opens properly.""" |
| test_result = CommandResult() |
| test_result.returncode = 0 |
| |
| os.path.join(IsA(str), IsA(str)).AndReturn(self.named_file.name) |
| cb_url_lib.RunCommand( |
| ['gsutil', 'cp', self.gsd_url, self.named_file.name], |
| redirect_stdout=True, redirect_stderr=True).AndReturn(test_result) |
| self.mox.ReplayAll() |
| self.assertTrue(cb_url_lib.Download(self.gsd_url)) |
| |
| def testGsdUrlFileCopyFails(self): |
| """Verify return value when gsutil copy fails.""" |
| test_result = CommandResult() |
| test_result.returncode = -1 |
| |
| os.path.join(IsA(str), IsA(str)).AndReturn(self.named_file.name) |
| cb_url_lib.RunCommand( |
| ['gsutil', 'cp', self.gsd_url, self.named_file.name], |
| redirect_stdout=True, redirect_stderr=True).AndReturn(test_result) |
| self.mox.ReplayAll() |
| self.assertFalse(cb_url_lib.Download(self.gsd_url)) |
| |
| |
| class TestDetermineThenDownloadCheckMd5(mox.MoxTestBase): |
| """Unit tests related to DetermineThenDownloadCheckMd5.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(cb_url_lib, 'DetermineUrl') |
| self.url = 'url' |
| self.pattern = ['pattern'] |
| self.path = 'path' |
| self.desc = 'desc' |
| |
| def testUrlGood(self): |
| """Verify return value when determining URL succeeds.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'DownloadCheckMd5') |
| cb_url_lib.DetermineUrl(self.url, self.pattern).AndReturn(self.url) |
| cb_url_lib.DownloadCheckMd5(self.url, |
| self.path, |
| self.desc).AndReturn('name') |
| self.mox.ReplayAll() |
| expected = 'name' |
| actual = cb_url_lib.DetermineThenDownloadCheckMd5(self.url, |
| self.pattern, |
| self.path, |
| self.desc) |
| self.assertEqual(expected, actual) |
| |
| def testUrlBad(self): |
| """Verify clean return value when determining URL fails.""" |
| cb_url_lib.DetermineUrl(self.url, self.pattern).AndReturn(None) |
| self.mox.ReplayAll() |
| self.assertRaises(cb_url_lib.NameResolutionError, |
| cb_url_lib.DetermineThenDownloadCheckMd5, |
| self.url, |
| self.pattern, |
| self.path, |
| self.desc) |
| |
| |
| class TestCheckResourceExistsWithMd5(mox.MoxTestBase): |
| """Unit tests related to CheckResourceExistsWithMd5.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(os.path, 'exists') |
| self.filename = 'file' |
| self.md5filename = 'file.md5' |
| |
| def testFileExistsMd5FileExistsMd5ChecksOut(self): |
| """Verify return value when all is well.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'CheckMd5') |
| os.path.exists('file').AndReturn(True) |
| os.path.exists('file.md5').AndReturn(True) |
| cb_url_lib.CheckMd5('file', 'file.md5').AndReturn(True) |
| self.mox.ReplayAll() |
| self.assertTrue(cb_url_lib.CheckResourceExistsWithMd5(self.filename, |
| self.md5filename)) |
| |
| def testFileDoesNotExist(self): |
| """Verify return value when local file does not exist.""" |
| os.path.exists('file').AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertFalse(cb_url_lib.CheckResourceExistsWithMd5(self.filename, |
| self.md5filename)) |
| |
| def testMd5FileDoesNotExist(self): |
| """Verify return value when nonexistent md5 file name is given.""" |
| os.path.exists('file').AndReturn(True) |
| os.path.exists('file.md5').AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertFalse(cb_url_lib.CheckResourceExistsWithMd5(self.filename, |
| self.md5filename)) |
| |
| def testMd5DoesNotChecksOut(self): |
| """Verify return value when MD5 checksum is bad.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'CheckMd5') |
| os.path.exists('file').AndReturn(True) |
| os.path.exists('file.md5').AndReturn(True) |
| cb_url_lib.CheckMd5('file', 'file.md5').AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertFalse(cb_url_lib.CheckResourceExistsWithMd5(self.filename, |
| self.md5filename)) |
| |
| |
| class TestDownloadCheckMd5(mox.MoxTestBase): |
| """Unit tests related to DownloadCheckMd5.""" |
| |
| def setUp(self): |
| self.mox = mox.Mox() |
| self.mox.StubOutWithMock(cb_url_lib, 'CheckResourceExistsWithMd5') |
| self.url = 'url/file' |
| self.md5url = 'url/file.md5' |
| self.path = 'path' |
| self.desc = 'desc' |
| self.name = 'path/file' |
| self.md5name = 'path/file.md5' |
| |
| def testResourceExists(self): |
| """Test behavior when resource exists with good MD5.""" |
| cb_url_lib.CheckResourceExistsWithMd5(self.name, |
| self.md5name).AndReturn(True) |
| self.mox.ReplayAll() |
| expected = self.name |
| actual = cb_url_lib.DownloadCheckMd5(self.url, self.path, self.desc) |
| self.assertEqual(expected, actual) |
| |
| def testDownloadsCheckMd5Succeed(self): |
| """Test behavior when fetch is entirely successful.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'Download') |
| self.mox.StubOutWithMock(cb_url_lib, 'CheckMd5') |
| cb_url_lib.CheckResourceExistsWithMd5(self.name, |
| self.md5name).AndReturn(False) |
| cb_url_lib.Download(self.url).AndReturn(True) |
| cb_url_lib.Download(self.md5url).AndReturn(True) |
| cb_url_lib.CheckMd5(self.name, self.md5name).AndReturn(True) |
| self.mox.ReplayAll() |
| expected = self.name |
| actual = cb_url_lib.DownloadCheckMd5(self.url, self.path, self.desc) |
| self.assertEqual(expected, actual) |
| |
| def testFileDownloadFails(self): |
| """Test behavior when file fetch fails.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'Download') |
| cb_url_lib.CheckResourceExistsWithMd5(self.name, |
| self.md5name).AndReturn(False) |
| cb_url_lib.Download(self.url).AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertRaises(BundlingError, |
| cb_url_lib.DownloadCheckMd5, |
| self.url, |
| self.path, |
| self.desc) |
| |
| def testMd5FileDownloadFails(self): |
| """Test behavior when MD5 file fetch fails.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'Download') |
| cb_url_lib.CheckResourceExistsWithMd5(self.name, |
| self.md5name).AndReturn(False) |
| cb_url_lib.Download(self.url).AndReturn(True) |
| cb_url_lib.Download(self.md5url).AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertRaises(BundlingError, |
| cb_url_lib.DownloadCheckMd5, |
| self.url, |
| self.path, |
| self.desc) |
| |
| def testMd5CheckFails(self): |
| """Test behavior when MD5 check fails.""" |
| self.mox.StubOutWithMock(cb_url_lib, 'Download') |
| self.mox.StubOutWithMock(cb_url_lib, 'CheckMd5') |
| cb_url_lib.CheckResourceExistsWithMd5(self.name, |
| self.md5name).AndReturn(False) |
| cb_url_lib.Download(self.url).AndReturn(True) |
| cb_url_lib.Download(self.md5url).AndReturn(True) |
| cb_url_lib.CheckMd5(self.name, self.md5name).AndReturn(False) |
| self.mox.ReplayAll() |
| self.assertRaises(BundlingError, |
| cb_url_lib.DownloadCheckMd5, |
| self.url, |
| self.path, |
| self.desc) |
| |
| |
| class TestConvertHttpToGsUrl(unittest.TestCase): |
| """Unit tests for _ConvertHttpToGsUrl method.""" |
| |
| def testConvertHttpToGsUrl(self): |
| expected = 'gs://chromeos-releases/stable-channel/x86-alex/0.12.433.269/*' |
| http_url = ('https://sandbox.google.com/storage/chromeos-releases/' |
| 'stable-channel/x86-alex/0.12.433.269') |
| actual = cb_url_lib._ConvertHttpToGsUrl(http_url) |
| self.assertEqual(expected, actual) |
| |
| |
| if __name__ == "__main__": |
| logging.basicConfig(level=logging.CRITICAL) |
| unittest.main() |
