| # Copyright 2015 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import argparse |
| import os |
| import sys |
| import unittest |
| |
| DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data') |
| ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| |
| sys.path.insert(0, ROOT_DIR) |
| sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party')) |
| |
| from glucose import install |
| from glucose import main_ |
| from glucose import pack |
| from glucose import util |
| |
| |
| class ParserTest(unittest.TestCase): |
| def test_install_zero_packages(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| options = parser.parse_args(['install']) |
| self.assertTrue(hasattr(options, 'command')) |
| self.assertTrue(callable(options.command)) |
| |
| def test_install_one_package(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| options = parser.parse_args(['install', 'package_name.whl', |
| '-i', 'nice-dir']) |
| self.assertTrue(hasattr(options, 'command')) |
| self.assertTrue(callable(options.command)) |
| |
| def test_install_two_packages(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| options = parser.parse_args(['install', 'package_name.whl', |
| 'package2_name.whl', '-i', 'nice-dir2']) |
| self.assertTrue(hasattr(options, 'command')) |
| self.assertTrue(callable(options.command)) |
| |
| |
| class CheckSha1Test(unittest.TestCase): |
| def test_good_hash(self): |
| filename = os.path.join( |
| DATA_DIR, 'wheels', 'source_package_valid_sha1-0.0.1-0_' |
| + '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| self.assertTrue(install.has_valid_sha1(filename, verbose=False)) |
| |
| def test_bad_hash(self): |
| filename = os.path.join( |
| DATA_DIR, 'wheels', 'source_package_invalid_sha1-0.0.1-0_' |
| + 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| self.assertFalse(install.has_valid_sha1(filename, verbose=False)) |
| |
| def test_no_build_number(self): |
| # The file doesn't have to exist. |
| filename = ('source_package_invalid_sha1-0.0.1-py2-none-any.whl') |
| self.assertFalse(install.has_valid_sha1(filename, verbose=False)) |
| |
| def test_invalid_filename(self): |
| # The file doesn't have to exist. |
| filename = ('source_package.txt') |
| self.assertFalse(install.has_valid_sha1(filename, verbose=False)) |
| |
| |
| class InstallLocalPackagesTest(unittest.TestCase): |
| def test_install_one_package(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| |
| with util.temporary_directory(prefix='glyco-install-test-') as tempdir: |
| options = parser.parse_args(['pack', |
| os.path.join(DATA_DIR, 'source_package'), |
| '--output-dir', tempdir]) |
| self.assertFalse(pack.pack(options)) |
| wheel_paths = [os.path.join(tempdir, whl) for whl in os.listdir(tempdir)] |
| |
| options = parser.parse_args( |
| ['install', wheel_paths[0], |
| '--install-dir', os.path.join(tempdir, 'local')]) |
| |
| install.install(options) |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'source_package'))) |
| self.assertTrue( |
| len(os.listdir(os.path.join(tempdir, 'local', 'source_package'))) > 0) |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'source_package-0.0.1.dist-info'))) |
| |
| def test_install_two_packages(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| |
| with util.temporary_directory(prefix='glyco-install-test-') as tempdir: |
| options = parser.parse_args(['pack', |
| os.path.join(DATA_DIR, 'source_package'), |
| os.path.join(DATA_DIR, 'installed_package'), |
| '--output-dir', tempdir]) |
| self.assertFalse(pack.pack(options)) |
| wheel_paths = [os.path.join(tempdir, whl) for whl in os.listdir(tempdir)] |
| |
| options = parser.parse_args( |
| ['install', wheel_paths[0], wheel_paths[1], |
| '--install-dir', os.path.join(tempdir, 'local')]) |
| |
| install.install(options) |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'source_package'))) |
| # Make sure the directory is not empty. |
| self.assertTrue( |
| len(os.listdir(os.path.join(tempdir, 'local', 'source_package'))) > 0) |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'source_package-0.0.1.dist-info'))) |
| |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'installed_package'))) |
| self.assertTrue( |
| len(os.listdir( |
| os.path.join(tempdir, 'local', 'installed_package'))) > 0) |
| self.assertTrue(os.path.isdir( |
| os.path.join(tempdir, 'local', 'installed_package-0.0.1.dist-info'))) |
| |
| def test_install_invalid_hash(self): |
| parser = argparse.ArgumentParser() |
| main_.add_argparse_options(parser) |
| with util.temporary_directory(prefix='glyco-install-test-') as tempdir: |
| options = parser.parse_args([ |
| 'install', os.path.join( |
| DATA_DIR, 'wheels', 'source_package_invalid_sha1-0.0.1-0_' |
| + 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'), |
| '--install-dir', os.path.join(tempdir, 'local')]) |
| |
| result = install.install(options) |
| |
| self.assertNotEqual(result, 0) |
| |
| |
| class FakeHttp(object): |
| """Replacement object for httplib2.Http""" |
| |
| def __init__(self): |
| # Map url to header and filename containing response. |
| # File names are relative to DATA_DIR |
| self.filenames = {} |
| filenames = self.filenames # saving a few characters. |
| |
| # Good file |
| filenames['https://a.random.host.com/' |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'] = [ |
| {'status': '200'}, |
| os.path.join('wheels', |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ] |
| |
| # Bad files |
| filenames['https://a.random.host.com/' |
| 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'] = [ |
| {'status': '200'}, |
| os.path.join('wheels', |
| 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ] |
| |
| filenames['https://a.random.host.com/invalid_name.whl'] = [ |
| {'status': '200'}, |
| os.path.join('wheels', |
| 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ] |
| |
| # And simulating an error. |
| filenames['https://a.broken.host.com/' |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'] = [ |
| {'status': '403'}, None |
| ] |
| |
| def request(self, url, verb): |
| if verb != 'GET': |
| raise NotImplementedError('Only GET is implementd in this mock.') |
| resp, filename = self.filenames[url] |
| content = '' |
| if filename: |
| with open(os.path.join(DATA_DIR, filename), 'rb') as f: |
| content = f.read() |
| return resp, content |
| |
| |
| class FetchPackagesTest(unittest.TestCase): |
| def test_fetch_one_file(self): |
| requester = None |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = os.path.join( |
| DATA_DIR, |
| 'wheels', |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| |
| install_list = [{'location': 'file://' + location, |
| 'location_type': 'file'}] |
| paths = install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| self.assertEqual(len(paths), 1) |
| self.assertEqual(paths[0], location) |
| # The file is readily available, nothing should be copied to the cache. |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| def test_fetch_two_files(self): |
| requester = None |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| locations = [ |
| os.path.join( |
| DATA_DIR, |
| 'wheels', |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl' |
| ), |
| os.path.join( |
| DATA_DIR, |
| 'wheels', |
| 'installed_package-0.0.1-0_' |
| '58a752f45f35a07c7d94149511b3af04bab11740-py2-none-any.whl' |
| ) |
| ] |
| |
| install_list = [{'location': 'file://' + location, |
| 'location_type': 'file'} |
| for location in locations] |
| paths = install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| self.assertEqual(len(paths), 2) |
| self.assertEqual(paths[0], locations[0]) |
| self.assertEqual(paths[1], locations[1]) |
| |
| # The file is readily available, nothing should be copied to the cache. |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| def test_fetch_one_invalid_file(self): |
| requester = None |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = os.path.join( |
| DATA_DIR, |
| 'wheels', |
| 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl' |
| ) |
| install_list = [{'location': 'file://' + location, |
| 'location_type': 'file'}] |
| with self.assertRaises(ValueError): |
| install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| # The file is readily available, nothing should be copied to the cache. |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| def test_fetch_one_http_file_good(self): |
| requester = FakeHttp() |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = ('https://a.random.host.com/' |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| install_list = [{'location': location, |
| 'location_type': 'http'}] |
| paths = install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| cache_files = [os.path.join(cache, filename) |
| for filename in os.listdir(cache)] |
| self.assertEqual(len(cache_files), 1) |
| self.assertEqual(paths[0], cache_files[0]) |
| |
| def test_fetch_one_http_file_invalid_sha1(self): |
| requester = FakeHttp() |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = ('https://a.random.host.com/' |
| 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| install_list = [{'location': location, |
| 'location_type': 'http'}] |
| with self.assertRaises(ValueError): |
| install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| # An invalid file should be deleted from the cache. |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| def test_fetch_one_http_file_invalid_name(self): |
| requester = FakeHttp() |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = 'https://a.random.host.com/invalid_name.whl' |
| install_list = [{'location': location, |
| 'location_type': 'http'}] |
| with self.assertRaises(ValueError): |
| install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| # Still nothing in the cache |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| def test_fetch_one_http_file_http_error(self): |
| requester = FakeHttp() |
| with util.temporary_directory(prefix='glyco-fetch-test-') as tempdir: |
| cache = os.path.join(tempdir, 'cache') |
| location = ('https://a.broken.host.com/' |
| 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| install_list = [{'location': location, |
| 'location_type': 'http'}] |
| with self.assertRaises(ValueError): |
| install.fetch_packages( |
| install_list, requester=requester, cache=cache, verbose=False) |
| |
| # Still nothing in the cache |
| self.assertEqual(len(os.listdir(cache)), 0) |
| |
| |
| class GetInstallListTest(unittest.TestCase): |
| def test_one_file(self): |
| install_list = install.get_install_list([ |
| os.path.join(DATA_DIR, 'wheels', 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ]) |
| |
| self.assertEqual(len(install_list), 1) |
| self.assertEqual(install_list[0]['location_type'], 'file') |
| self.assertTrue(install_list[0]['location'].startswith('file://')) |
| |
| def test_two_files(self): |
| install_list = install.get_install_list([ |
| os.path.join(DATA_DIR, 'wheels', 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'), |
| os.path.join(DATA_DIR, 'wheels', 'source_package_invalid_sha1-0.0.1-0_' |
| 'deadbeef3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ]) |
| |
| self.assertEqual(len(install_list), 2) |
| for item in install_list: |
| self.assertEqual(item['location_type'], 'file') |
| self.assertTrue(item['location'].startswith('file://')) |
| |
| def test_existing_and_non_existing_files(self): |
| install_list = install.get_install_list([ |
| os.path.join(DATA_DIR, 'wheels', 'source_package_valid_sha1-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl'), |
| os.path.join( |
| DATA_DIR, 'wheels', 'non_existing_file-0.0.1-0_' |
| '41128a9c3767ced3ec53b89b297e0268f1338b2b-py2-none-any.whl') |
| ]) |
| self.assertTrue(len(install_list), 2) |
| self.assertFalse(install_list[0].get('error')) |
| self.assertTrue(install_list[1].get('error')) |
| |
| def test_http_uri(self): |
| install_list = install.get_install_list( |
| ['http://some.host.com/some_package.whl']) |
| self.assertTrue(len(install_list), 1) |
| self.assertTrue(install_list[0].get('error')) |
| |
| def test_https_uri(self): |
| install_list = install.get_install_list( |
| ['https://some.host.com/some_package.whl']) |
| |
| self.assertEqual(len(install_list), 1) |
| self.assertEqual(install_list[0]['location_type'], 'http') |
| self.assertTrue(install_list[0]['location'].startswith('https://')) |
| |
| def test_gs_uri(self): |
| install_list = install.get_install_list( |
| ['gs://some_bucket/some_package.whl']) |
| |
| self.assertEqual(len(install_list), 1) |
| self.assertEqual(install_list[0]['location_type'], 'http') |
| self.assertEqual( |
| install_list[0]['location'], |
| 'https://storage.googleapis.com/some_bucket/some_package.whl') |
| |
| def test_unknown_uri(self): |
| install_list = install.get_install_list( |
| ['weird://some.host.com/some_package.whl']) |
| self.assertTrue(len(install_list), 1) |
| self.assertTrue(install_list[0].get('error')) |
| |
| if __name__ == '__main__': |
| unittest.main() |
