<s> from __future__ import absolute_import <EOL> import os <EOL> import logging <EOL> import tarfile <EOL> import hashlib <EOL> from collections import namedtuple <EOL> from urlparse import urlparse <EOL> import toml <EOL> import json <EOL> import zinc . helpers as helpers <EOL> import zinc . utils as utils <EOL> from . catalog import ZincCatalog <EOL> from . defaults import defaults <EOL> from . coordinators import coordinator_for_url <EOL> from . formats import Formats <EOL> from . models import ZincModel , ZincIndex , ZincCatalogConfig <EOL> from . storages import storage_for_url <EOL> from . tasks . bundle_update import ZincBundleUpdateTask <EOL> from . utils import enum , memoized <EOL> log = logging . getLogger ( __name__ ) <EOL> SymbolicBundleVersions = utils . enum ( <EOL> ALL = '<STR_LIT>' , <EOL> UNREFERENCED = '<STR_LIT>' , <EOL> LATEST = '<STR_LIT>' ) <EOL> SymbolicSingleBundleVersions = utils . enum ( <EOL> LATEST = '<STR_LIT>' ) <EOL> BundleVersionDistroPrefix = '<STR_LIT:@>' <EOL> class ZincClientConfig ( ZincModel ) : <EOL> VARS = '<STR_LIT>' <EOL> ENV = '<STR_LIT>' <EOL> def __init__ ( self , d = None , ** kwargs ) : <EOL> super ( ZincClientConfig , self ) . __init__ ( ** kwargs ) <EOL> self . _d = d <EOL> @ classmethod <EOL> def from_bytes ( cls , b , mutable = True ) : <EOL> d = toml . loads ( b ) <EOL> return cls . from_dict ( d , mutable = mutable ) <EOL> @ classmethod <EOL> def from_dict ( cls , d , mutable = True ) : <EOL> replaced = cls . _replace_vars ( d , d . get ( cls . VARS ) or dict ( ) ) <EOL> zincConfig = cls ( replaced , mutable = mutable ) <EOL> return zincConfig <EOL> @ classmethod <EOL> def _replace_vars ( cls , indict , vars ) : <EOL> outdict = dict ( ) <EOL> for key , value in indict . iteritems ( ) : <EOL> if isinstance ( value , dict ) : <EOL> outdict [ key ] = cls . _replace_vars ( value , vars ) <EOL> elif isinstance ( value , basestring ) and ( value . startswith ( cls . VARS + '<STR_LIT::>' ) <EOL> or value . startswith ( cls . ENV + '<STR_LIT::>' ) ) : <EOL> if value . startswith ( cls . VARS + '<STR_LIT::>' ) : <EOL> varname = value [ len ( cls . VARS ) + <NUM_LIT:1> : ] <EOL> var = vars [ varname ] <EOL> elif value . startswith ( cls . ENV + '<STR_LIT::>' ) : <EOL> varname = value [ len ( cls . ENV ) + <NUM_LIT:1> : ] <EOL> var = os . environ [ varname ] <EOL> outdict [ key ] = var <EOL> else : <EOL> outdict [ key ] = value <EOL> return outdict <EOL> @ property <EOL> def vars ( self ) : <EOL> return self . _d . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def bookmarks ( self ) : <EOL> return self . _d . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def coordinators ( self ) : <EOL> return self . _d . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def storages ( self ) : <EOL> return self . _d . get ( '<STR_LIT>' ) <EOL> OutputType = enum ( PRETTY = '<STR_LIT>' , JSON = '<STR_LIT>' ) <EOL> class _Result ( object ) : <EOL> def __init__ ( self , pretty = None ) : <EOL> self . _pretty = pretty or str <EOL> def to_dict ( self ) : <EOL> raise NotImplementedError ( ) <EOL> def format ( self , fmt ) : <EOL> if fmt == OutputType . JSON : <EOL> return json . dumps ( self . to_dict ( ) ) <EOL> elif fmt == OutputType . PRETTY : <EOL> return self . _pretty ( self ) <EOL> else : <EOL> raise NotImplementedError ( ) <EOL> class DictResult ( _Result ) : <EOL> def __init__ ( self , d , ** kwargs ) : <EOL> super ( DictResult , self ) . __init__ ( ** kwargs ) <EOL> self . _dict = d <EOL> def to_dict ( self ) : <EOL> return self . _dict <EOL> def __str__ ( self ) : <EOL> return str ( self . _dict ) <EOL> def __getitem__ ( self , k ) : <EOL> return self . _dict [ k ] <EOL> MessageTypes = enum ( <EOL> INFO = '<STR_LIT:info>' , <EOL> WARNING = '<STR_LIT>' , <EOL> ERROR = '<STR_LIT:error>' ) <EOL> class Message ( _Result ) : <EOL> def __init__ ( self , type , text , ** kwargs ) : <EOL> super ( Message , self ) . __init__ ( ** kwargs ) <EOL> self . _type = type <EOL> self . _text = text <EOL> @ classmethod <EOL> def info ( cls , s ) : <EOL> return cls ( MessageTypes . INFO , s ) <EOL> @ classmethod <EOL> def warn ( cls , s ) : <EOL> return cls ( MessageTypes . WARNING , s ) <EOL> @ classmethod <EOL> def error ( cls , s ) : <EOL> return cls ( MessageTypes . ERROR , s ) <EOL> @ property <EOL> def text ( self ) : <EOL> return self . _text <EOL> @ property <EOL> def type ( self ) : <EOL> return self . _type <EOL> @ type . setter <EOL> def type ( self , val ) : <EOL> assert val in MessageTypes <EOL> self . _type = val <EOL> def to_dict ( self ) : <EOL> return { <EOL> '<STR_LIT:message>' : { <EOL> '<STR_LIT:type>' : self . type , <EOL> '<STR_LIT:text>' : self . text , <EOL> } <EOL> } <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . _type , self . _text ) <EOL> class ResultSet ( object ) : <EOL> def __init__ ( self , items , pretty = None ) : <EOL> self . _items = items <EOL> self . pretty = pretty or str <EOL> @ property <EOL> @ memoized <EOL> def items ( self ) : <EOL> return self . _items ( ) <EOL> def __iter__ ( self ) : <EOL> return iter ( self . items ) <EOL> def __str__ ( self ) : <EOL> return str ( self . items ) <EOL> def errors ( self ) : <EOL> return [ i for i in self . items if isinstance ( i , Message ) and i . type == MessageTypes . ERROR ] <EOL> def catalog_list ( catalog , distro = None , print_versions = True , ** kwargs ) : <EOL> index = catalog . get_index ( ) <EOL> def pretty_without_versions ( result ) : <EOL> return "<STR_LIT:%s>" % ( result [ '<STR_LIT>' ] ) <EOL> def pretty_with_versions ( result ) : <EOL> distros = index . distributions_for_bundle_by_version ( result [ '<STR_LIT>' ] ) <EOL> versions = index . versions_for_bundle ( result [ '<STR_LIT>' ] ) <EOL> version_strings = list ( ) <EOL> for version in versions : <EOL> version_string = str ( version ) <EOL> if distros . get ( version ) is not None : <EOL> distro_string = "<STR_LIT>" % ( "<STR_LIT:U+002CU+0020>" . join ( sorted ( distros . get ( version ) ) ) ) <EOL> version_string += '<STR_LIT:=>' + distro_string <EOL> version_strings . append ( version_string ) <EOL> final_version_string = "<STR_LIT>" % ( "<STR_LIT:U+002CU+0020>" . join ( version_strings ) ) <EOL> return "<STR_LIT>" % ( result [ '<STR_LIT>' ] , final_version_string ) <EOL> pretty = pretty_with_versions if print_versions else pretty_without_versions <EOL> def results ( ) : <EOL> bundle_names = sorted ( index . bundle_names ( ) ) <EOL> for bundle_name in bundle_names : <EOL> d = dict ( ) <EOL> if distro and distro not in index . distributions_for_bundle ( bundle_name ) : <EOL> continue <EOL> d [ '<STR_LIT>' ] = bundle_name <EOL> d [ '<STR_LIT>' ] = index . versions_for_bundle ( bundle_name ) <EOL> d [ '<STR_LIT>' ] = index . distributions_for_bundle ( bundle_name ) <EOL> yield DictResult ( d , pretty = pretty ) <EOL> return ResultSet ( results ) <EOL> def bundle_list ( catalog , bundle_name , version_ish , print_sha = False , flavor_name = None ) : <EOL> version = _resolve_single_bundle_version ( catalog , bundle_name , version_ish ) <EOL> manifest = catalog . manifest_for_bundle ( bundle_name , version = version ) <EOL> def pretty ( r ) : <EOL> if print_sha : <EOL> return "<STR_LIT>" % ( r [ '<STR_LIT:file>' ] , r [ '<STR_LIT>' ] ) <EOL> else : <EOL> return "<STR_LIT:%s>" % ( r [ '<STR_LIT:file>' ] ) <EOL> def results ( ) : <EOL> all_files = sorted ( manifest . get_all_files ( flavor = flavor_name ) ) <EOL> for f in all_files : <EOL> d = { <EOL> '<STR_LIT:file>' : f , <EOL> '<STR_LIT>' : manifest . sha_for_file ( f ) <EOL> } <EOL> yield DictResult ( d , pretty = pretty ) <EOL> return ResultSet ( results ) <EOL> def bundle_verify ( catalog , bundle_name , version_ish , check_shas = True , <EOL> should_lock = False , ** kwargs ) : <EOL> version = _resolve_single_bundle_version ( catalog , bundle_name , version_ish ) <EOL> manifest = catalog . get_manifest ( bundle_name , version ) <EOL> def results ( ) : <EOL> for result in _verify_bundle_with_manifest ( catalog , manifest , <EOL> check_shas = check_shas , <EOL> should_lock = should_lock , <EOL> ** kwargs ) : <EOL> yield result <EOL> return ResultSet ( results ) <EOL> def verify_catalog ( catalog , should_lock = False , ** kwargs ) : <EOL> index = catalog . get_index ( ) <EOL> manifests = list ( ) <EOL> ph = catalog . path_helper <EOL> def results ( ) : <EOL> for ( bundle_name , bundle_info ) in index . _bundle_info_by_name . iteritems ( ) : <EOL> for version in bundle_info [ '<STR_LIT>' ] : <EOL> manifest_name = ph . manifest_name ( bundle_name , version ) <EOL> yield Message . info ( "<STR_LIT>" % ( manifest_name ) ) <EOL> manifest = catalog . get_manifest ( bundle_name , version ) <EOL> if manifest is None : <EOL> yield Message . error ( "<STR_LIT>" % ( manifest_name ) ) <EOL> continue <EOL> manifests . append ( manifest ) <EOL> verified_files = set ( ) <EOL> for manifest in manifests : <EOL> yield Message . info ( "<STR_LIT>" % ( manifest . bundle_name , <EOL> manifest . version ) ) <EOL> for result in _verify_bundle_with_manifest ( catalog , manifest , <EOL> verified_files = verified_files ) : <EOL> yield result <EOL> return ResultSet ( results ) <EOL> def create_bundle_version ( catalog , bundle_name , src_dir , flavor_spec = None , <EOL> force = False , skip_master_archive = False ) : <EOL> task = ZincBundleUpdateTask ( ) <EOL> task . catalog = catalog <EOL> task . bundle_name = bundle_name <EOL> task . src_dir = src_dir <EOL> task . flavor_spec = flavor_spec <EOL> task . skip_master_archive = skip_master_archive <EOL> task . force = force <EOL> return task . run ( ) <EOL> def delete_bundle_versions ( catalog , bundle_name , version_ish ) : <EOL> version_list = _resolve_multiple_bundle_versions ( catalog , bundle_name , version_ish ) <EOL> with catalog . lock ( ) : <EOL> for version in version_list : <EOL> catalog . delete_bundle_version ( bundle_name , version ) <EOL> def update_distribution ( catalog , distro_name , bundle_name , version , <EOL> save_previous = True ) : <EOL> catalog . update_distribution ( distro_name , bundle_name , version , <EOL> save_previous = save_previous ) <EOL> def delete_distribution ( catalog , distribution_name , bundle_name , <EOL> delete_previous = True ) : <EOL> catalog . delete_distribution ( distribution_name , bundle_name , <EOL> delete_previous = delete_previous ) <EOL> def clone_bundle ( catalog , bundle_name , version , root_path = None , bundle_dir_name = None , flavor = None ) : <EOL> assert catalog <EOL> assert bundle_name <EOL> assert version <EOL> if root_path is None : <EOL> root_path = '<STR_LIT:.>' <EOL> if bundle_dir_name is None : <EOL> bundle_id = helpers . make_bundle_id ( catalog . id , bundle_name ) <EOL> bundle_dir_name = helpers . make_bundle_descriptor ( bundle_id , version , <EOL> flavor = flavor ) <EOL> manifest = catalog . manifest_for_bundle ( bundle_name , version ) <EOL> if manifest is None : <EOL> raise Exception ( "<STR_LIT>" % ( bundle_name , version ) ) <EOL> if flavor is not None and flavor not in manifest . flavors : <EOL> raise Exception ( "<STR_LIT>" % ( flavor ) ) <EOL> all_files = manifest . get_all_files ( flavor = flavor ) <EOL> root_dir = os . path . join ( root_path , bundle_dir_name ) <EOL> utils . makedirs ( root_dir ) <EOL> for file in all_files : <EOL> dst_path = os . path . join ( root_dir , file ) <EOL> formats = manifest . formats_for_file ( file ) <EOL> sha = manifest . sha_for_file ( file ) <EOL> utils . makedirs ( os . path . dirname ( dst_path ) ) <EOL> if formats . get ( Formats . RAW ) is not None : <EOL> format = Formats . RAW <EOL> elif formats . get ( Formats . GZ ) is not None : <EOL> format = Formats . GZ <EOL> else : <EOL> format = None <EOL> ext = helpers . file_extension_for_format ( format ) <EOL> with catalog . _read_file ( sha , ext = ext ) as infile : <EOL> b = infile . read ( ) <EOL> if format == Formats . GZ : <EOL> b = utils . gunzip_bytes ( b ) <EOL> with open ( dst_path , '<STR_LIT>' ) as outfile : <EOL> outfile . write ( b ) <EOL> log . info ( "<STR_LIT>" % ( sha , dst_path ) ) <EOL> log . info ( "<STR_LIT>" % ( len ( all_files ) , root_dir ) ) <EOL> def _catalog_connection_get_api_version ( url ) : <EOL> import requests <EOL> ZINC_VERSION_HEADER = '<STR_LIT>' <EOL> resp = requests . head ( url , allow_redirects = False ) <EOL> api_version = resp . headers . get ( ZINC_VERSION_HEADER ) <EOL> if api_version is None : <EOL> raise Exception ( "<STR_LIT>" % <EOL> ( ZINC_VERSION_HEADER ) ) <EOL> return api_version <EOL> def _catalog_connection_get_http ( url ) : <EOL> ZINC_SUPPORTED_API_VERSIONS = ( '<STR_LIT:1.0>' ) <EOL> api_version = _catalog_connection_get_api_version ( url ) <EOL> if api_version not in ZINC_SUPPORTED_API_VERSIONS : <EOL> raise Exception ( "<STR_LIT>" % ( api_version ) ) <EOL> else : <EOL> log . debug ( "<STR_LIT>" % ( api_version ) ) <EOL> def catalog_ref_split ( catalog_ref ) : <EOL> CatalogRefSplitResult = namedtuple ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> CatalogInfo = namedtuple ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> urlcomps = urlparse ( catalog_ref ) <EOL> if urlcomps . scheme in ( '<STR_LIT:http>' , '<STR_LIT>' ) : <EOL> catalog_id = os . path . split ( urlcomps . path ) [ - <NUM_LIT:1> ] <EOL> service = catalog_ref [ : - len ( catalog_id ) ] <EOL> return CatalogRefSplitResult ( service , CatalogInfo ( catalog_id , None ) ) <EOL> elif urlcomps . scheme in ( '<STR_LIT:file>' , '<STR_LIT>' ) : <EOL> return CatalogRefSplitResult ( catalog_ref , CatalogInfo ( None , '<STR_LIT:.>' ) ) <EOL> def create_catalog ( catalog_id = None , storage_info = None ) : <EOL> assert catalog_id <EOL> assert storage_info <EOL> storage_class = storage_for_url ( storage_info [ '<STR_LIT:url>' ] ) <EOL> storage = storage_class ( ** storage_info ) <EOL> catalog_storage = storage . bind_to_catalog ( id = catalog_id ) <EOL> catalog_storage . puts ( defaults [ '<STR_LIT>' ] , <EOL> ZincCatalogConfig ( ) . to_bytes ( ) ) <EOL> catalog_storage . puts ( defaults [ '<STR_LIT>' ] , <EOL> ZincIndex ( catalog_id ) . to_bytes ( ) ) <EOL> catalog = ZincCatalog ( storage = catalog_storage ) <EOL> catalog . save ( ) <EOL> def get_service ( service_url = None , coordinator_info = None , storage_info = None , ** kwargs ) : <EOL> if service_url is not None : <EOL> urlcomps = urlparse ( service_url ) <EOL> if urlcomps . scheme in ( '<STR_LIT:http>' , '<STR_LIT>' ) : <EOL> _catalog_connection_get_http ( service_url ) <EOL> from zinc . services . web import WebServiceConsumer <EOL> return WebServiceConsumer ( service_url ) <EOL> elif urlcomps . scheme in ( '<STR_LIT:file>' , '<STR_LIT>' ) : <EOL> if urlcomps . scheme == '<STR_LIT>' : <EOL> file_url = '<STR_LIT>' % ( utils . canonical_path ( service_url ) ) <EOL> else : <EOL> file_url = service_url <EOL> from zinc . services . simple import SimpleServiceConsumer <EOL> return SimpleServiceConsumer ( file_url ) <EOL> elif coordinator_info is not None and storage_info is not None : <EOL> coord_class = coordinator_for_url ( coordinator_info [ '<STR_LIT:url>' ] ) <EOL> coord = coord_class ( ** coordinator_info ) <EOL> storage_class = storage_for_url ( storage_info [ '<STR_LIT:url>' ] ) <EOL> storage = storage_class ( ** storage_info ) <EOL> from zinc . services import CustomServiceConsumer <EOL> return CustomServiceConsumer ( coordinator = coord , storage = storage ) <EOL> raise NotImplementedError ( ) <EOL> def connect ( service_url = None , coordinator_info = None , storage_info = None , ** kwargs ) : <EOL> return get_service ( service_url = service_url , <EOL> coordinator_info = coordinator_info , <EOL> storage_info = storage_info , ** kwargs ) <EOL> def _verify_bundle_with_manifest ( catalog , manifest , check_shas = True , <EOL> should_lock = False , ** kwargs ) : <EOL> if not check_shas : <EOL> yield Message . warn ( '<STR_LIT>' ) <EOL> for path , info in manifest . files . iteritems ( ) : <EOL> sha = manifest . sha_for_file ( path ) <EOL> if kwargs . get ( '<STR_LIT>' ) is not None : <EOL> if sha in kwargs [ '<STR_LIT>' ] : <EOL> log . debug ( '<STR_LIT>' % ( sha ) ) <EOL> continue <EOL> else : <EOL> kwargs [ '<STR_LIT>' ] . add ( sha ) <EOL> meta = catalog . _get_file_info ( sha ) <EOL> if meta is None : <EOL> yield Message . error ( '<STR_LIT>' % ( sha ) ) <EOL> continue <EOL> format = meta [ '<STR_LIT>' ] <EOL> meta_size = meta [ '<STR_LIT:size>' ] <EOL> manifest_size = info [ '<STR_LIT>' ] [ format ] [ '<STR_LIT:size>' ] <EOL> log . debug ( "<STR_LIT>" % ( sha , format , meta_size , manifest_size ) ) <EOL> if meta_size != manifest_size : <EOL> yield Message . error ( '<STR_LIT>' % ( sha ) ) <EOL> continue <EOL> if check_shas : <EOL> ext = helpers . file_extension_for_format ( format ) <EOL> with catalog . _read_file ( sha , ext = ext ) as f : <EOL> b = f . read ( ) <EOL> if format == Formats . GZ : <EOL> b = utils . gunzip_bytes ( b ) <EOL> digest = hashlib . sha1 ( b ) . hexdigest ( ) <EOL> if digest != sha : <EOL> yield Message . error ( '<STR_LIT>' % ( sha ) ) <EOL> continue <EOL> yield Message . info ( '<STR_LIT>' % ( sha ) ) <EOL> flavors = list ( manifest . flavors ) <EOL> flavors . append ( None ) <EOL> for flavor in flavors : <EOL> archive_name = catalog . path_helper . archive_name ( manifest . bundle_name , manifest . version , flavor = flavor ) <EOL> meta = catalog . _get_archive_info ( manifest . bundle_name , manifest . version , flavor = flavor ) <EOL> if meta is None : <EOL> if len ( manifest . get_all_files ( flavor = flavor ) ) == <NUM_LIT:1> : <EOL> continue <EOL> elif flavor is None and len ( flavors ) > <NUM_LIT:1> : <EOL> Message . warn ( '<STR_LIT>' % ( archive_name ) ) <EOL> continue <EOL> else : <EOL> yield Message . error ( '<STR_LIT>' % ( archive_name ) ) <EOL> for result in _verify_archive ( catalog , manifest , flavor = flavor , check_shas = check_shas ) : <EOL> yield result <EOL> def _verify_archive ( catalog , manifest , flavor = None , check_shas = True ) : <EOL> if not check_shas : <EOL> yield Message . warn ( '<STR_LIT>' ) <EOL> archive_name = catalog . path_helper . archive_name ( manifest . bundle_name , manifest . version , flavor = flavor ) <EOL> all_files = manifest . get_all_files ( flavor = flavor ) <EOL> with catalog . _read_archive ( manifest . bundle_name , manifest . version , flavor = flavor ) as fileobj : <EOL> tar = tarfile . open ( fileobj = fileobj ) <EOL> members = tar . getmembers ( ) <EOL> member_names = tar . getnames ( ) <EOL> found_error = False <EOL> for file in all_files : <EOL> sha = manifest . sha_for_file ( file ) <EOL> format , info = manifest . get_format_info_for_file ( file , preferred_formats = defaults [ '<STR_LIT>' ] ) <EOL> target_member_name = helpers . append_file_extension_for_format ( sha , format ) <EOL> if target_member_name not in member_names : <EOL> found_error = True <EOL> yield Message . error ( '<STR_LIT>' % ( target_member_name , helpers . make_bundle_descriptor ( manifest . bundle_name , manifest . version , flavor = flavor ) ) ) <EOL> else : <EOL> member = members [ member_names . index ( target_member_name ) ] <EOL> if check_shas : <EOL> f = tar . extractfile ( member ) <EOL> b = f . read ( ) <EOL> f . close ( ) <EOL> if format == Formats . GZ : <EOL> b = utils . gunzip_bytes ( b ) <EOL> digest = hashlib . sha1 ( b ) . hexdigest ( ) <EOL> if digest != sha : <EOL> found_error = True <EOL> yield Message . error ( '<STR_LIT>' % ( target_member_name , digest ) ) <EOL> else : <EOL> if info [ '<STR_LIT:size>' ] != member . size : <EOL> found_error = True <EOL> yield Message . error ( '<STR_LIT>' % ( target_member_name , info [ '<STR_LIT:size>' ] , member . size ) ) <EOL> tar . close ( ) <EOL> if not found_error : <EOL> yield Message . info ( '<STR_LIT>' % ( archive_name ) ) <EOL> def _resolve_single_bundle_version ( catalog , bundle_name , version_ish ) : <EOL> if isinstance ( version_ish , int ) : <EOL> version = version_ish <EOL> elif version_ish == SymbolicBundleVersions . LATEST : <EOL> index = catalog . get_index ( ) <EOL> version = index . versions_for_bundle ( bundle_name ) [ - <NUM_LIT:1> ] <EOL> elif version_ish . startswith ( '<STR_LIT:@>' ) : <EOL> source_distro = version_ish [ <NUM_LIT:1> : ] <EOL> version = catalog . index . version_for_bundle ( bundle_name , source_distro ) <EOL> return version <EOL> def _resolve_multiple_bundle_versions ( catalog , bundle_name , version_ish ) : <EOL> if version_ish == SymbolicBundleVersions . ALL : <EOL> index = catalog . get_index ( ) <EOL> versions = index . versions_for_bundle ( bundle_name ) <EOL> elif version_ish == SymbolicBundleVersions . UNREFERENCED : <EOL> index = catalog . get_index ( ) <EOL> all_versions = index . versions_for_bundle ( bundle_name ) <EOL> referenced_versions = catalog . index . distributions_for_bundle_by_version ( bundle_name ) . keys ( ) <EOL> versions = [ v for v in all_versions if v not in referenced_versions ] <EOL> single_version = _resolve_single_bundle_version ( catalog , bundle_name , version_ish ) <EOL> if single_version is not None : <EOL> versions = [ single_version ] <EOL> return versions </s>
