<s> import os <EOL> import logging <EOL> import json <EOL> from zinc . models import ZincIndex , ZincManifest , ZincFlavorSpec <EOL> from zinc . catalog import ZincCatalogPathHelper <EOL> from zinc . defaults import defaults <EOL> from zinc . catalog import ZincCatalog <EOL> from zinc . storages import StorageBackend <EOL> from zinc . client import connect , create_bundle_version <EOL> import zinc . helpers as helpers <EOL> import zinc . utils as utils <EOL> from tests import * <EOL> class StorageBackendTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . storage = StorageBackend ( ) <EOL> def test_get_raises ( self ) : <EOL> self . assertRaises ( NotImplementedError , self . storage . get , '<STR_LIT:foo>' ) <EOL> def test_get_meta_raises ( self ) : <EOL> self . assertRaises ( NotImplementedError , self . storage . get_meta , '<STR_LIT:foo>' ) <EOL> def test_put_raises ( self ) : <EOL> self . assertRaises ( NotImplementedError , self . storage . put , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> def create_catalog_at_path ( path , id ) : <EOL> service = connect ( '<STR_LIT:/>' ) <EOL> service . create_catalog ( id = id , loc = path ) <EOL> catalog = service . get_catalog ( loc = path ) <EOL> return catalog <EOL> class ZincCatalogPathHelperTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . pathHelper = ZincCatalogPathHelper ( ) <EOL> def test_config_dir ( self ) : <EOL> self . assertEquals ( self . pathHelper . config_dir , "<STR_LIT>" ) <EOL> def test_config_flavors_spec_dir ( self ) : <EOL> self . assertEquals ( self . pathHelper . config_flavorspec_dir , "<STR_LIT>" ) <EOL> def test_path_for_flavor_spec_name ( self ) : <EOL> flavor_spec_name = "<STR_LIT>" <EOL> expected_path = "<STR_LIT>" % flavor_spec_name <EOL> actual_path = self . pathHelper . path_for_flavorspec_name ( flavor_spec_name ) <EOL> self . assertEquals ( expected_path , actual_path ) <EOL> class ZincCatalogTestCase ( TempDirTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ZincCatalogTestCase , self ) . setUp ( ) <EOL> self . catalog_dir = os . path . join ( self . dir , "<STR_LIT>" ) <EOL> os . mkdir ( self . catalog_dir ) <EOL> self . scratch_dir = os . path . join ( self . dir , "<STR_LIT>" ) <EOL> os . mkdir ( self . scratch_dir ) <EOL> logging . info ( "<STR_LIT>" % self . catalog_dir ) <EOL> logging . info ( "<STR_LIT>" % self . scratch_dir ) <EOL> def path_exists_in_catalog ( self , subpath ) : <EOL> fullpath = os . path . join ( self . catalog_dir , subpath ) <EOL> return os . path . exists ( fullpath ) <EOL> def test_catalog_create ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> self . assertTrue ( catalog is not None ) <EOL> self . assertTrue ( len ( catalog . index . bundle_names ( ) ) == <NUM_LIT:0> ) <EOL> self . assertTrue ( catalog . format ( ) == defaults [ '<STR_LIT>' ] ) <EOL> def test_catalog_read_invalid_format ( self ) : <EOL> create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> index_path = os . path . join ( self . catalog_dir , defaults [ '<STR_LIT>' ] ) <EOL> index = ZincIndex . from_path ( index_path ) <EOL> index . _format = <NUM_LIT:2> <EOL> index . write ( index_path ) <EOL> self . assertRaises ( Exception , ZincCatalog , ( self . catalog_dir ) ) <EOL> def test_catalog_import_file ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> f1 = create_random_file ( self . scratch_dir ) <EOL> catalog . import_path ( f1 ) <EOL> def test_bundle_names_with_no_bundles ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> self . assertTrue ( len ( catalog . index . bundle_names ( ) ) == <NUM_LIT:0> ) <EOL> def test_versions_for_nonexistant_bundle ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> versions = catalog . index . versions_for_bundle ( "<STR_LIT>" ) <EOL> self . assertTrue ( len ( versions ) == <NUM_LIT:0> ) <EOL> def _build_test_catalog ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> catalog . _reload ( ) <EOL> return catalog <EOL> def test_create_bundle_version ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> self . assertTrue ( "<STR_LIT>" in catalog . get_index ( ) . bundle_names ( ) ) <EOL> self . assertTrue ( <NUM_LIT:1> in catalog . get_index ( ) . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> manifest = catalog . manifest_for_bundle ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> self . assertTrue ( manifest is not None ) <EOL> for ( file , props ) in manifest . files . items ( ) : <EOL> sha = props [ '<STR_LIT>' ] <EOL> formats = props [ '<STR_LIT>' ] <EOL> for format in formats . keys ( ) : <EOL> ext = None <EOL> if format == '<STR_LIT>' : <EOL> ext = '<STR_LIT>' <EOL> object_path = ZincCatalogPathHelper ( ) . path_for_file_with_sha ( sha , ext ) <EOL> self . assertTrue ( self . path_exists_in_catalog ( object_path ) ) <EOL> def test_bundle_name_in_manifest ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> bundle_name = "<STR_LIT>" <EOL> manifest = catalog . manifest_for_bundle ( bundle_name , <NUM_LIT:1> ) <EOL> self . assertTrue ( manifest . bundle_name == bundle_name ) <EOL> def test_create_bundle_with_subdirs ( self ) : <EOL> create_random_file ( self . scratch_dir ) <EOL> one_dir = os . mkdir ( os . path . join ( self . scratch_dir , "<STR_LIT>" ) ) <EOL> create_random_file ( one_dir ) <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> def test_create_second_bundle_version ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> self . assertTrue ( <NUM_LIT:2> in catalog . get_index ( ) . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> new_index = ZincIndex . from_path ( os . path . join ( catalog . path , defaults [ '<STR_LIT>' ] ) ) <EOL> self . assertTrue ( <NUM_LIT:1> in new_index . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( <NUM_LIT:2> in new_index . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> def test_create_duplicate_bundle_version_no_force ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> create_random_file ( self . scratch_dir ) <EOL> manifest1 = create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> self . assertTrue ( <NUM_LIT:2> in catalog . get_index ( ) . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> manifest2 = create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> self . assertEquals ( manifest1 . version , manifest2 . version ) <EOL> def test_create_duplicate_bundle_version_with_force ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> create_random_file ( self . scratch_dir ) <EOL> manifest1 = create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> self . assertTrue ( <NUM_LIT:2> in catalog . get_index ( ) . versions_for_bundle ( "<STR_LIT>" ) ) <EOL> manifest2 = create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir , <EOL> force = True ) <EOL> self . assertNotEquals ( manifest1 . version , manifest2 . version ) <EOL> def test_create_identical_bundle_version ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> self . assertEquals ( len ( catalog . get_index ( ) . versions_for_bundle ( "<STR_LIT>" ) ) , <NUM_LIT:1> ) <EOL> def test_path_for_manifest_with_name_version ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> manifest = ZincManifest ( catalog . index . id , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> path = ZincCatalogPathHelper ( ) . path_for_manifest ( manifest ) <EOL> filename = os . path . split ( path ) [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( filename , '<STR_LIT>' ) <EOL> def test_single_file_bundle_does_not_create_archive ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> archive_path = ZincCatalogPathHelper ( ) . path_for_archive_for_bundle_version ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> self . assertFalse ( self . path_exists_in_catalog ( archive_path ) ) <EOL> def test_more_than_one_file_bundle_does_create_archive ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir ) <EOL> archive_path = ZincCatalogPathHelper ( ) . path_for_archive_for_bundle_version ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> self . assertTrue ( self . path_exists_in_catalog ( archive_path ) ) <EOL> def test_single_file_flavor_does_not_create_archive ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> flavor_spec = ZincFlavorSpec . from_dict ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir , <EOL> flavor_spec = flavor_spec ) <EOL> archive_path = ZincCatalogPathHelper ( ) . path_for_archive_for_bundle_version ( "<STR_LIT>" , <NUM_LIT:1> , <EOL> flavor = '<STR_LIT>' ) <EOL> self . assertFalse ( self . path_exists_in_catalog ( archive_path ) ) <EOL> def test_skip_master_archive_and_no_flavor_specified ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir , <EOL> skip_master_archive = True ) <EOL> archive_path = ZincCatalogPathHelper ( ) . path_for_archive_for_bundle_version ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> self . assertTrue ( self . path_exists_in_catalog ( archive_path ) ) <EOL> def test_skip_master_archive_and_flavor_specified ( self ) : <EOL> catalog = create_catalog_at_path ( self . catalog_dir , '<STR_LIT>' ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_random_file ( self . scratch_dir ) <EOL> flavor_spec = ZincFlavorSpec . from_dict ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> create_bundle_version ( catalog , "<STR_LIT>" , self . scratch_dir , <EOL> flavor_spec = flavor_spec , skip_master_archive = True ) <EOL> archive_path = ZincCatalogPathHelper ( ) . path_for_archive_for_bundle_version ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> self . assertFalse ( self . path_exists_in_catalog ( archive_path ) ) <EOL> def test_update_distro_basic ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> bundle_name , distro = "<STR_LIT>" , "<STR_LIT>" <EOL> catalog . update_distribution ( distro , bundle_name , <NUM_LIT:1> ) <EOL> version = catalog . index . version_for_bundle ( bundle_name , distro ) <EOL> self . assertEquals ( version , <NUM_LIT:1> ) <EOL> def test_save_prev_distro_if_no_previous ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> bundle_name , distro = "<STR_LIT>" , "<STR_LIT>" <EOL> catalog . update_distribution ( distro , bundle_name , <NUM_LIT:1> ) <EOL> prev_distro = helpers . distro_previous_name ( distro ) <EOL> prev_version = catalog . index . version_for_bundle ( bundle_name , prev_distro ) <EOL> self . assertTrue ( prev_version is None ) <EOL> def test_save_prev_distro_if_prev_exists ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> bundle_name , distro = "<STR_LIT>" , "<STR_LIT>" <EOL> catalog . update_distribution ( distro , bundle_name , <NUM_LIT:1> ) <EOL> create_random_file ( self . scratch_dir ) <EOL> create_bundle_version ( catalog , bundle_name , self . scratch_dir ) <EOL> catalog . update_distribution ( distro , bundle_name , <NUM_LIT:2> ) <EOL> prev_distro = helpers . distro_previous_name ( distro ) <EOL> prev_version = catalog . index . version_for_bundle ( bundle_name , prev_distro ) <EOL> self . assertEquals ( prev_version , <NUM_LIT:1> ) <EOL> def add_dummy_flavorspec ( self , catalog , flavorspec_name ) : <EOL> flavorspec_string = json . dumps ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> subpath = catalog . path_helper . path_for_flavorspec_name ( flavorspec_name ) <EOL> catalog . _storage . puts ( subpath , flavorspec_string ) <EOL> def test_update_flavorspec ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> flavorspec_string = json . dumps ( { '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> catalog . update_flavorspec_from_json_string ( "<STR_LIT>" , flavorspec_string ) <EOL> expected_path = catalog . _ph . path_for_flavorspec_name ( "<STR_LIT>" ) <EOL> self . path_exists_in_catalog ( expected_path ) <EOL> def test_list_flavorspec ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> self . add_dummy_flavorspec ( catalog , "<STR_LIT:test>" ) <EOL> actual_names = catalog . get_flavorspec_names ( ) <EOL> self . assertEquals ( [ "<STR_LIT:test>" ] , actual_names ) <EOL> def test_delete_flavorspec ( self ) : <EOL> catalog = self . _build_test_catalog ( ) <EOL> self . add_dummy_flavorspec ( catalog , "<STR_LIT:test>" ) <EOL> catalog . delete_flavorspec ( "<STR_LIT:test>" ) <EOL> subpath = catalog . path_helper . path_for_flavorspec_name ( "<STR_LIT:test>" ) <EOL> self . assertFalse ( os . path . exists ( os . path . join ( self . dir , subpath ) ) ) </s>
