<s> from datetime import date , datetime <EOL> import pickle <EOL> from nose . tools import eq_ <EOL> from elasticutils import S , DefaultMappingType , NoModelError , MappingType <EOL> from elasticutils . tests import ESTestCase <EOL> model_cache = [ ] <EOL> def reset_model_cache ( ) : <EOL> del model_cache [ <NUM_LIT:0> : ] <EOL> class Meta ( object ) : <EOL> def __init__ ( self , db_table ) : <EOL> self . db_table = db_table <EOL> class Manager ( object ) : <EOL> def filter ( self , id__in = None ) : <EOL> return [ m for m in model_cache if m . id in id__in ] <EOL> class FakeModel ( object ) : <EOL> _meta = Meta ( '<STR_LIT>' ) <EOL> objects = Manager ( ) <EOL> def __init__ ( self , ** kw ) : <EOL> for key in kw : <EOL> setattr ( self , key , kw [ key ] ) <EOL> model_cache . append ( self ) <EOL> @ classmethod <EOL> def get ( cls , id ) : <EOL> id = int ( id ) <EOL> return [ m for m in model_cache if m . id == id ] [ <NUM_LIT:0> ] <EOL> class FakeMappingType ( MappingType ) : <EOL> @ classmethod <EOL> def get_index ( cls ) : <EOL> return '<STR_LIT>' <EOL> @ classmethod <EOL> def get_mapping_type_name ( cls ) : <EOL> return '<STR_LIT>' <EOL> def get_model ( self ) : <EOL> return FakeModel <EOL> class TestResultsWithData ( ESTestCase ) : <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:foo>' : '<STR_LIT:bar>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:width>' : '<STR_LIT:2>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:width>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:3> , '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:width>' : '<STR_LIT:5>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:4> , '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:width>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:5> , '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:width>' : '<STR_LIT>' } <EOL> ] <EOL> @ classmethod <EOL> def teardown_class ( cls ) : <EOL> super ( TestResultsWithData , cls ) . teardown_class ( ) <EOL> reset_model_cache ( ) <EOL> def test_default_results_are_default_mapping_type ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) . query ( foo = '<STR_LIT:bar>' ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , DefaultMappingType ) <EOL> def test_typed_s_returns_type ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( FakeMappingType ) . query ( foo = '<STR_LIT:bar>' ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , FakeMappingType ) <EOL> def test_values_dict_no_fields ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) . query ( foo = '<STR_LIT:bar>' ) . values_dict ( ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , dict ) <EOL> def test_values_dict_results ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) <EOL> . query ( foo = '<STR_LIT:bar>' ) <EOL> . values_dict ( '<STR_LIT:foo>' , '<STR_LIT:width>' ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , dict ) <EOL> eq_ ( <EOL> sorted ( searcher [ <NUM_LIT:0> ] . items ( ) ) , <EOL> sorted ( [ ( u'<STR_LIT:foo>' , [ u'<STR_LIT:bar>' ] ) , ( u'<STR_LIT:width>' , [ u'<STR_LIT:2>' ] ) ] ) <EOL> ) <EOL> def test_values_list_no_fields ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) <EOL> . query ( foo = '<STR_LIT:bar>' ) <EOL> . values_list ( ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , tuple ) <EOL> eq_ ( <EOL> sorted ( searcher [ <NUM_LIT:0> ] ) , <EOL> sorted ( [ [ u'<STR_LIT:1>' ] , [ u'<STR_LIT>' ] ] ) <EOL> ) <EOL> def test_values_list_results ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) <EOL> . query ( foo = '<STR_LIT:bar>' ) <EOL> . values_list ( '<STR_LIT:foo>' , '<STR_LIT:width>' ) ) <EOL> assert isinstance ( searcher [ <NUM_LIT:0> ] , tuple ) <EOL> eq_ ( <EOL> sorted ( searcher [ <NUM_LIT:0> ] ) , <EOL> sorted ( ( [ u'<STR_LIT:2>' ] , [ u'<STR_LIT:bar>' ] ) ) <EOL> ) <EOL> def test_default_results_form_has_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) . query ( foo = '<STR_LIT:bar>' ) ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:id>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:source>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:type>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> def test_values_list_form_has_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) . query ( foo = '<STR_LIT:bar>' ) . values_list ( '<STR_LIT:id>' ) ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:id>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:source>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:type>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> def test_values_dict_form_has_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> searcher = list ( self . get_s ( ) . query ( foo = '<STR_LIT:bar>' ) . values_dict ( ) ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:id>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:source>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT:type>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> assert hasattr ( searcher [ <NUM_LIT:0> ] . es_meta , '<STR_LIT>' ) <EOL> def test_values_dict_no_args ( self ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( S ( ) . query ( fld1 = <NUM_LIT:2> ) <EOL> . values_dict ( ) <EOL> . build_search ( ) , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:2> } } , '<STR_LIT>' : [ '<STR_LIT:*>' ] } ) <EOL> def test_values_list_no_args ( self ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( S ( ) . query ( fld1 = <NUM_LIT:2> ) <EOL> . values_list ( ) <EOL> . build_search ( ) , <EOL> { '<STR_LIT>' : { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:2> } } , '<STR_LIT>' : [ '<STR_LIT:*>' ] } ) <EOL> class TestResultsWithStoredFields ( ESTestCase ) : <EOL> def test_values_list_no_args_no_stored_fields ( self ) : <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> '<STR_LIT:name>' : { '<STR_LIT:type>' : '<STR_LIT:string>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_list ( ) ) <EOL> eq_ ( sorted ( results [ <NUM_LIT:0> ] , key = repr ) , <EOL> sorted ( ( [ u'<STR_LIT:1>' ] , [ u'<STR_LIT>' ] ) , key = repr ) <EOL> ) <EOL> def test_values_list_no_args_with_stored_fields ( self ) : <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:store>' : True } , <EOL> '<STR_LIT:name>' : { '<STR_LIT:type>' : '<STR_LIT:string>' , '<STR_LIT:store>' : True } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_list ( ) ) <EOL> eq_ ( sorted ( results [ <NUM_LIT:0> ] , key = repr ) , <EOL> sorted ( ( [ <NUM_LIT:1> ] , [ u'<STR_LIT>' ] ) , key = repr ) <EOL> ) <EOL> def test_values_dict_no_args_no_stored_fields ( self ) : <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> '<STR_LIT:name>' : { '<STR_LIT:type>' : '<STR_LIT:string>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_dict ( ) ) <EOL> eq_ ( sorted ( results [ <NUM_LIT:0> ] . items ( ) ) , <EOL> sorted ( [ ( '<STR_LIT>' , [ u'<STR_LIT:1>' ] ) , ( '<STR_LIT>' , [ u'<STR_LIT>' ] ) ] ) <EOL> ) <EOL> def test_values_dict_no_args_with_stored_fields ( self ) : <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:store>' : True } , <EOL> '<STR_LIT:name>' : { '<STR_LIT:type>' : '<STR_LIT:string>' , '<STR_LIT:store>' : True } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_dict ( ) ) <EOL> eq_ ( sorted ( results [ <NUM_LIT:0> ] . items ( ) ) , <EOL> sorted ( [ ( u'<STR_LIT:id>' , [ <NUM_LIT:1> ] ) , ( u'<STR_LIT:name>' , [ u'<STR_LIT>' ] ) ] ) <EOL> ) <EOL> class TestFakeMappingType ( ESTestCase ) : <EOL> index_name = FakeMappingType . get_index ( ) <EOL> mapping_type_name = FakeMappingType . get_mapping_type_name ( ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> ] <EOL> @ classmethod <EOL> def setup_class ( cls ) : <EOL> super ( TestFakeMappingType , cls ) . setup_class ( ) <EOL> for doc in cls . data : <EOL> FakeModel ( ** doc ) <EOL> @ classmethod <EOL> def teardown_class ( cls ) : <EOL> super ( TestFakeMappingType , cls ) . setup_class ( ) <EOL> reset_model_cache ( ) <EOL> def test_object ( self ) : <EOL> s = S ( FakeMappingType ) . query ( name__prefix = '<STR_LIT>' ) <EOL> eq_ ( len ( s ) , <NUM_LIT:1> ) <EOL> eq_ ( s [ <NUM_LIT:0> ] . object . id , <NUM_LIT:1> ) <EOL> class TestResultsWithDates ( ESTestCase ) : <EOL> def test_dates ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT:date>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT:date>' } <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : date ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT> ) } , <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_dict ( '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> eq_ ( results , <EOL> [ { u'<STR_LIT>' : [ datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) ] , <EOL> u'<STR_LIT>' : [ datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:0> ) ] , <EOL> u'<STR_LIT:id>' : [ <NUM_LIT:1> ] } ] <EOL> ) <EOL> def test_dates_lookalikes ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup_index ( ) <EOL> self . create_index ( <EOL> mappings = { <EOL> self . mapping_type_name : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:id>' : { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT:string>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> } <EOL> } <EOL> } <EOL> ) <EOL> data = [ <EOL> { '<STR_LIT:id>' : [ <NUM_LIT:1> ] , '<STR_LIT>' : [ '<STR_LIT>' ] } <EOL> ] <EOL> self . index_data ( data ) <EOL> self . refresh ( ) <EOL> results = list ( self . get_s ( ) . values_dict ( '<STR_LIT:id>' , '<STR_LIT>' ) ) <EOL> eq_ ( results , <EOL> [ { u'<STR_LIT:id>' : [ <NUM_LIT:1> ] , u'<STR_LIT>' : [ u'<STR_LIT>' ] } ] <EOL> ) <EOL> class TestMappingType ( ESTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestMappingType , self ) . setUp ( ) <EOL> self . cleanup_index ( ) <EOL> self . create_index ( ) <EOL> def tearDown ( self ) : <EOL> self . cleanup_index ( ) <EOL> super ( TestMappingType , self ) . tearDown ( ) <EOL> def test_default_mapping_type ( self ) : <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> ] <EOL> self . index_data ( data ) <EOL> s = self . get_s ( DefaultMappingType ) <EOL> result = list ( s ) [ <NUM_LIT:0> ] <EOL> assert isinstance ( result , DefaultMappingType ) <EOL> eq_ ( result . id , <NUM_LIT:1> ) <EOL> self . assertRaises ( NoModelError , lambda : result . object ) <EOL> def test_mapping_type_attribute_override ( self ) : <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : '<STR_LIT:foo>' } <EOL> ] <EOL> self . index_data ( data ) <EOL> s = self . get_s ( DefaultMappingType ) <EOL> result = list ( s ) [ <NUM_LIT:0> ] <EOL> eq_ ( result . _object , None ) <EOL> eq_ ( result [ '<STR_LIT>' ] , '<STR_LIT:foo>' ) <EOL> eq_ ( result . id , <NUM_LIT:1> ) <EOL> result . id = '<STR_LIT:foo>' <EOL> eq_ ( result . id , '<STR_LIT:foo>' ) <EOL> eq_ ( result [ '<STR_LIT:id>' ] , <NUM_LIT:1> ) <EOL> self . assertRaises ( AttributeError , lambda : result . doesnt_exist ) <EOL> self . assertRaises ( KeyError , lambda : result [ '<STR_LIT>' ] ) <EOL> def test_pickleable ( self ) : <EOL> data = [ <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : '<STR_LIT:foo>' } <EOL> ] <EOL> self . index_data ( data ) <EOL> s = self . get_s ( DefaultMappingType ) <EOL> result = list ( s ) [ <NUM_LIT:0> ] <EOL> pickled_mt = pickle . dumps ( result , <NUM_LIT:2> ) <EOL> unpickled = pickle . loads ( pickled_mt ) <EOL> eq_ ( unpickled . id , <NUM_LIT:1> ) </s>
