<s> from datetime import datetime <EOL> from nose . plugins . attrib import attr <EOL> from manager_rest import storage_manager , models <EOL> from manager_rest . test import base_test <EOL> @ attr ( client_min_version = <NUM_LIT:1> , client_max_version = base_test . LATEST_API_VERSION ) <EOL> class StorageManagerTests ( base_test . BaseServerTestCase ) : <EOL> def test_store_load_delete_blueprint ( self ) : <EOL> now = str ( datetime . now ( ) ) <EOL> sm = storage_manager . _get_instance ( ) <EOL> blueprint = models . BlueprintState ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> description = None , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> source = '<STR_LIT>' , <EOL> main_file_name = '<STR_LIT>' ) <EOL> sm . put_blueprint ( '<STR_LIT>' , blueprint ) <EOL> blueprint_from_list = sm . blueprints_list ( ) . items [ <NUM_LIT:0> ] <EOL> blueprint_restored = sm . get_blueprint ( '<STR_LIT>' ) <EOL> bp_from_delete = sm . delete_blueprint ( '<STR_LIT>' ) <EOL> self . assertEquals ( blueprint . to_dict ( ) , blueprint_from_list . to_dict ( ) ) <EOL> self . assertEquals ( blueprint . to_dict ( ) , blueprint_restored . to_dict ( ) ) <EOL> self . assertEquals ( blueprint . id , bp_from_delete . id ) <EOL> blueprints_list = sm . blueprints_list ( ) . items <EOL> self . assertEquals ( <NUM_LIT:0> , len ( blueprints_list ) ) <EOL> def test_get_blueprint_deployments ( self ) : <EOL> now = str ( datetime . now ( ) ) <EOL> sm = storage_manager . _get_instance ( ) <EOL> blueprint = models . BlueprintState ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> description = None , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> source = '<STR_LIT>' , <EOL> main_file_name = '<STR_LIT>' ) <EOL> sm . put_blueprint ( '<STR_LIT>' , blueprint ) <EOL> deployment1 = models . Deployment ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> blueprint_id = '<STR_LIT>' , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> permalink = None , <EOL> workflows = { } , <EOL> inputs = { } , <EOL> policy_types = { } , <EOL> policy_triggers = { } , <EOL> groups = { } , <EOL> outputs = { } ) <EOL> sm . put_deployment ( '<STR_LIT>' , deployment1 ) <EOL> deployment2 = models . Deployment ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> blueprint_id = '<STR_LIT>' , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> permalink = None , <EOL> workflows = { } , <EOL> inputs = { } , <EOL> policy_types = { } , <EOL> policy_triggers = { } , <EOL> groups = { } , <EOL> outputs = { } ) <EOL> sm . put_deployment ( '<STR_LIT>' , deployment2 ) <EOL> deployment3 = models . Deployment ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> blueprint_id = '<STR_LIT>' , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> permalink = None , <EOL> workflows = { } , <EOL> inputs = { } , <EOL> policy_types = { } , <EOL> policy_triggers = { } , <EOL> groups = { } , <EOL> outputs = { } ) <EOL> sm . put_deployment ( '<STR_LIT>' , deployment3 ) <EOL> blueprint_deployments = sm . get_blueprint_deployments ( '<STR_LIT>' ) <EOL> blueprint_deployments = sm . get_blueprint_deployments ( <EOL> '<STR_LIT>' ) . items <EOL> self . assertEquals ( <NUM_LIT:2> , len ( blueprint_deployments ) ) <EOL> if blueprint_deployments [ <NUM_LIT:0> ] . id != deployment1 . id : <EOL> blueprint_deployments [ <NUM_LIT:0> ] , blueprint_deployments [ <NUM_LIT:1> ] = blueprint_deployments [ <NUM_LIT:1> ] , blueprint_deployments [ <NUM_LIT:0> ] <EOL> self . assertEquals ( deployment1 . to_dict ( ) , <EOL> blueprint_deployments [ <NUM_LIT:0> ] . to_dict ( ) ) <EOL> self . assertEquals ( deployment2 . to_dict ( ) , <EOL> blueprint_deployments [ <NUM_LIT:1> ] . to_dict ( ) ) <EOL> def test_model_serialization ( self ) : <EOL> dep = models . Deployment ( id = '<STR_LIT>' , <EOL> created_at = '<STR_LIT>' , <EOL> updated_at = '<STR_LIT>' , <EOL> blueprint_id = '<STR_LIT>' , <EOL> plan = { '<STR_LIT>' : '<STR_LIT:value>' } , <EOL> permalink = None , <EOL> workflows = { } , <EOL> inputs = { } , <EOL> policy_types = { } , <EOL> policy_triggers = { } , <EOL> groups = { } , <EOL> outputs = { } ) <EOL> serialized_dep = dep . to_dict ( ) <EOL> self . assertEquals ( <NUM_LIT:11> , len ( serialized_dep ) ) <EOL> self . assertEquals ( dep . id , serialized_dep [ '<STR_LIT:id>' ] ) <EOL> self . assertEquals ( dep . created_at , serialized_dep [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( dep . updated_at , serialized_dep [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( dep . blueprint_id , serialized_dep [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( dep . permalink , serialized_dep [ '<STR_LIT>' ] ) <EOL> deserialized_dep = models . Deployment ( ** serialized_dep ) <EOL> self . assertEquals ( dep . id , deserialized_dep . id ) <EOL> self . assertEquals ( dep . created_at , deserialized_dep . created_at ) <EOL> self . assertEquals ( dep . updated_at , deserialized_dep . updated_at ) <EOL> self . assertEquals ( dep . blueprint_id , deserialized_dep . blueprint_id ) <EOL> self . assertEquals ( dep . permalink , deserialized_dep . permalink ) <EOL> def test_fields_query ( self ) : <EOL> now = str ( datetime . now ( ) ) <EOL> blueprint = models . BlueprintState ( id = '<STR_LIT>' , <EOL> created_at = now , <EOL> updated_at = now , <EOL> description = None , <EOL> plan = { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> source = '<STR_LIT>' , <EOL> main_file_name = '<STR_LIT>' ) <EOL> sm = storage_manager . _get_instance ( ) <EOL> sm . put_blueprint ( '<STR_LIT>' , blueprint ) <EOL> blueprint_restored = sm . get_blueprint ( '<STR_LIT>' , <EOL> { '<STR_LIT:id>' , '<STR_LIT>' } ) <EOL> self . assertEquals ( '<STR_LIT>' , blueprint_restored . id ) <EOL> self . assertEquals ( now , blueprint_restored . created_at ) <EOL> self . assertEquals ( None , blueprint_restored . updated_at ) <EOL> self . assertEquals ( None , blueprint_restored . plan ) <EOL> self . assertEquals ( None , blueprint_restored . main_file_name ) </s>
