<s> from couchdbkit . ext . django . loading import get_db <EOL> from django . test import TestCase , SimpleTestCase <EOL> from couchexport . export import SCALAR_NEVER_WAS , get_formatted_rows , scalar_never_was <EOL> from couchexport . models import ExportSchema , SavedExportSchema , SplitColumn <EOL> from datetime import datetime , timedelta <EOL> from couchexport . util import SerializableFunction <EOL> from dimagi . utils . couch . database import get_safe_write_kwargs <EOL> import json <EOL> from couchexport . models import Format <EOL> class ExportSchemaTest ( TestCase ) : <EOL> def testSaveAndLoad ( self ) : <EOL> index = [ "<STR_LIT:foo>" , <NUM_LIT:2> ] <EOL> schema = ExportSchema ( index = index , timestamp = datetime . utcnow ( ) ) <EOL> inner = { "<STR_LIT>" : { "<STR_LIT:bar>" : <NUM_LIT:1> , "<STR_LIT>" : [ <NUM_LIT:2> , <NUM_LIT:3> ] } , <EOL> "<STR_LIT:list>" : [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT>" : "<STR_LIT>" } ] , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> schema . schema = inner <EOL> schema . save ( ) <EOL> back = ExportSchema . get ( schema . get_id ) <EOL> self . assertEqual ( inner , back . schema ) <EOL> self . assertEqual ( index , back . index ) <EOL> def testGetLast ( self ) : <EOL> indices = [ "<STR_LIT>" , [ "<STR_LIT:a>" , "<STR_LIT:list>" ] ] <EOL> save_args = get_safe_write_kwargs ( ) <EOL> for index in indices : <EOL> self . assertEqual ( None , ExportSchema . last ( index ) ) <EOL> dt = datetime . utcnow ( ) <EOL> schema1 = ExportSchema ( index = index , timestamp = dt ) <EOL> schema1 . save ( ** save_args ) <EOL> self . assertEqual ( schema1 . _id , ExportSchema . last ( index ) . _id ) <EOL> schema2 = ExportSchema ( index = index , timestamp = dt + timedelta ( seconds = <NUM_LIT:1> ) ) <EOL> schema2 . save ( ** save_args ) <EOL> self . assertEqual ( schema2 . _id , ExportSchema . last ( index ) . _id ) <EOL> schema3 = ExportSchema ( index = index , timestamp = dt - timedelta ( seconds = <NUM_LIT:1> ) ) <EOL> schema3 . save ( ** save_args ) <EOL> self . assertEqual ( schema2 . _id , ExportSchema . last ( index ) . _id ) <EOL> class SavedSchemaTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . db = get_db ( '<STR_LIT>' ) <EOL> self . custom_export = SavedExportSchema . wrap ( { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : Format . JSON , <EOL> '<STR_LIT:index>' : json . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT:index>' : '<STR_LIT:#>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:index>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT:d>' ] } <EOL> ] , <EOL> } ] <EOL> } ) <EOL> self . custom_export . filter_function = SerializableFunction ( ) <EOL> self . schema = [ { '<STR_LIT>' : [ u'<STR_LIT:string>' ] , '<STR_LIT>' : u'<STR_LIT:string>' , '<STR_LIT>' : u'<STR_LIT:string>' } ] <EOL> def tearDown ( self ) : <EOL> for doc in self . db . all_docs ( ) : <EOL> if not doc [ '<STR_LIT:id>' ] . startswith ( '<STR_LIT>' ) : <EOL> self . db . delete_doc ( doc [ '<STR_LIT:id>' ] ) <EOL> def post_it ( self , ** kwargs ) : <EOL> doc = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> doc . update ( kwargs ) <EOL> self . db . save_doc ( doc , ** get_safe_write_kwargs ( ) ) <EOL> def _test_split_column ( self , rows , ignore_extras = False ) : <EOL> self . custom_export . tables_by_index [ '<STR_LIT:#>' ] . columns [ <NUM_LIT:0> ] . ignore_extras = ignore_extras <EOL> files = self . custom_export . get_export_files ( ) <EOL> data = json . loads ( files . file . payload ) <EOL> expected_headers = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> if not ignore_extras : <EOL> expected_headers += [ '<STR_LIT>' ] <EOL> self . assertEqual ( data [ '<STR_LIT>' ] [ '<STR_LIT>' ] , expected_headers ) <EOL> export_rows = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( len ( export_rows ) , len ( rows ) ) <EOL> tuple_rows = { tuple ( r ) for r in export_rows } <EOL> if len ( tuple_rows ) != len ( export_rows ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> for row in rows : <EOL> self . assertIn ( tuple ( row ) , tuple_rows ) <EOL> def test_split_column ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . _test_split_column ( [ [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , None ] ] ) <EOL> def test_split_column_order ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . _test_split_column ( [ [ <NUM_LIT:1> , None , <NUM_LIT:1> , <NUM_LIT:1> , None ] ] ) <EOL> def test_split_column_empty ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . _test_split_column ( [ [ None , None , None , None , None ] ] ) <EOL> def test_split_column_None ( self ) : <EOL> self . post_it ( multi = None ) <EOL> self . _test_split_column ( [ [ None , None , None , None , None ] ] ) <EOL> def test_split_column_ignore_extras ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . _test_split_column ( [ [ None , None , None , <NUM_LIT:1> ] ] , ignore_extras = True ) <EOL> def test_split_column_missing ( self ) : <EOL> self . post_it ( ) <EOL> self . _test_split_column ( [ [ SCALAR_NEVER_WAS ] * <NUM_LIT:5> ] ) <EOL> def test_split_column_missing_ignore_extras ( self ) : <EOL> self . post_it ( ) <EOL> self . _test_split_column ( [ [ SCALAR_NEVER_WAS ] * <NUM_LIT:4> ] , ignore_extras = True ) <EOL> def test_split_column_None_ignore_extras ( self ) : <EOL> self . post_it ( multi = None ) <EOL> self . _test_split_column ( [ [ None , None , None , None ] ] , ignore_extras = True ) <EOL> def test_split_column_fit_to_schema_None ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . post_it ( multi = None ) <EOL> self . _test_split_column ( <EOL> [ <EOL> [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , None ] , <EOL> [ None , None , None , None ] , <EOL> ] , <EOL> ignore_extras = True <EOL> ) <EOL> def test_split_column_fit_to_schema_missing ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . post_it ( ) <EOL> self . _test_split_column ( <EOL> [ <EOL> [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , None ] , <EOL> [ SCALAR_NEVER_WAS ] * <NUM_LIT:4> , <EOL> ] , <EOL> ignore_extras = True <EOL> ) <EOL> def test_split_column_remainder ( self ) : <EOL> self . post_it ( multi = '<STR_LIT>' ) <EOL> self . _test_split_column ( [ [ None , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , '<STR_LIT>' ] ] ) <EOL> def test_split_column_header_format ( self ) : <EOL> col = SplitColumn ( display = '<STR_LIT>' , options = [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> self . assertEqual ( <EOL> list ( col . get_headers ( ) ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_split_column_not_string ( self ) : <EOL> col = SplitColumn ( display = '<STR_LIT>' , options = [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) <EOL> self . assertEqual ( <EOL> col . get_data ( <NUM_LIT:1> ) , <EOL> [ None , None , <NUM_LIT:1> ] <EOL> ) <EOL> class GetFormattedRowsTests ( SimpleTestCase ) : <EOL> def test ( self ) : <EOL> doc = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> schema = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:string>' , <EOL> '<STR_LIT>' : '<STR_LIT:string>' <EOL> } <EOL> } <EOL> formatted_rows = get_formatted_rows ( doc , schema , '<STR_LIT:.>' ) <EOL> headers = formatted_rows [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] . get_data ( ) <EOL> values = formatted_rows [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] . get_data ( ) <EOL> row_dict = dict ( zip ( list ( headers ) , list ( values ) ) ) <EOL> self . assertEqual ( <EOL> row_dict , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : scalar_never_was , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ) <EOL> class ExportSchemaWrapTest ( SimpleTestCase ) : <EOL> def test_wrap_datetime_hippy ( self ) : <EOL> schema1 = ExportSchema ( <EOL> schema = { } , <EOL> timestamp = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> ) , <EOL> index = '<STR_LIT:index>' , <EOL> ) <EOL> schema2 = ExportSchema . wrap ( schema1 . to_json ( ) ) <EOL> self . assertEqual ( schema2 . timestamp , datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> def test_wrap_datetime_min ( self ) : <EOL> schema_bad = ExportSchema ( <EOL> schema = { } , <EOL> timestamp = datetime . min , <EOL> index = '<STR_LIT:index>' , <EOL> ) <EOL> schema_good = ExportSchema . wrap ( schema_bad . to_json ( ) ) <EOL> self . assertEqual ( schema_good . timestamp , datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) </s>
