<s> """<STR_LIT>""" <EOL> import sys <EOL> import os <EOL> import tempfile <EOL> from sqlalchemy import exc <EOL> import pytest <EOL> sys . path . insert ( <NUM_LIT:0> , os . path . dirname ( os . path . dirname ( os . path . dirname ( os . path . abspath ( __file__ ) ) ) ) ) <EOL> import gristle . metadata as mod <EOL> class TestSchema ( object ) : <EOL> def setup_method ( self , method ) : <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> self . md = mod . GristleMetaData ( self . tempdir ) <EOL> self . schema_id , self . collection_id = create_basic_metadata ( self . md ) <EOL> def teardown_method ( self , method ) : <EOL> os . remove ( os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> os . rmdir ( self . tempdir ) <EOL> def test_schema_contents ( self ) : <EOL> row = self . md . schema_tools . getter ( schema_name = '<STR_LIT>' ) <EOL> assert len ( row ) == <NUM_LIT:3> <EOL> assert row . schema_name == '<STR_LIT>' <EOL> assert row . schema_desc == '<STR_LIT>' <EOL> def test_schema_selecting_nonexistent_row ( self ) : <EOL> assert self . md . schema_tools . getter ( schema_name = '<STR_LIT>' ) is None <EOL> def test_schema_upsert ( self ) : <EOL> assert self . md . schema_tools . setter ( schema_name = '<STR_LIT>' , schema_desc = '<STR_LIT>' ) != <NUM_LIT:0> <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:2> <EOL> assert self . md . schema_tools . getter ( schema_name = '<STR_LIT>' ) . schema_desc == '<STR_LIT>' <EOL> assert self . md . schema_tools . setter ( schema_name = '<STR_LIT>' , schema_desc = '<STR_LIT>' ) == <NUM_LIT:0> <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:2> <EOL> assert self . md . schema_tools . getter ( schema_name = '<STR_LIT>' ) . schema_desc == '<STR_LIT>' <EOL> new_schema_rows = self . md . schema_tools . lister ( ) <EOL> def test_schema_list_and_delete ( self ) : <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:1> <EOL> old_schema_rows = self . md . schema_tools . lister ( ) <EOL> assert self . md . schema_tools . deleter ( schema_name = '<STR_LIT>' ) == <NUM_LIT:1> <EOL> assert self . md . schema_tools . deleter ( schema_name = '<STR_LIT>' ) == <NUM_LIT:0> <EOL> for field in self . md . field_tools . lister ( ) : <EOL> if field . collection_id == self . collection_id : <EOL> self . md . field_tools . deleter ( collection_id = self . collection_id , field_id = field . field_id ) <EOL> for collection in self . md . collection_tools . lister ( ) : <EOL> if collection . schema_id == self . schema_id : <EOL> assert self . md . collection_tools . deleter ( schema_id = self . schema_id , <EOL> collection_id = collection . collection_id ) == <NUM_LIT:1> <EOL> for element in self . md . element_tools . lister ( ) : <EOL> self . md . element_tools . deleter ( element_name = element . element_name ) <EOL> self . md . schema_tools . deleter ( schema_id = self . schema_id ) <EOL> assert len ( self . md . element_tools . lister ( ) ) == <NUM_LIT:0> <EOL> assert len ( self . md . field_tools . lister ( ) ) == <NUM_LIT:0> <EOL> assert len ( self . md . collection_tools . lister ( ) ) == <NUM_LIT:0> <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:0> <EOL> def test_schema_bad_delete ( self ) : <EOL> old_schema_rows = self . md . schema_tools . lister ( ) <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:1> <EOL> assert self . md . schema_tools . deleter ( schema_name = '<STR_LIT>' ) == <NUM_LIT:0> <EOL> assert len ( self . md . schema_tools . lister ( ) ) == <NUM_LIT:1> <EOL> new_schema_rows = self . md . schema_tools . lister ( ) <EOL> if old_schema_rows != new_schema_rows : <EOL> pytest . fail ( '<STR_LIT>' ) <EOL> class TestCollection ( object ) : <EOL> def setup_method ( self , method ) : <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> self . md = mod . GristleMetaData ( self . tempdir ) <EOL> self . schema_id , self . collection_id = create_basic_metadata ( self . md ) <EOL> def teardown_method ( self , method ) : <EOL> os . remove ( os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> os . rmdir ( self . tempdir ) <EOL> def test_collection_row_count ( self ) : <EOL> assert len ( self . md . collection_tools . lister ( ) ) == <NUM_LIT:1> <EOL> def test_collection_get ( self ) : <EOL> assert len ( self . md . collection_tools . getter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT>' ) ) == <NUM_LIT:4> <EOL> def test_collection_select_nonexisting_row ( self ) : <EOL> assert self . md . collection_tools . getter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT>' ) is None <EOL> def test_collection_deletion_by_pk ( self ) : <EOL> kva = { '<STR_LIT>' : self . schema_id , <EOL> '<STR_LIT>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> collection_id = self . md . collection_tools . setter ( ** kva ) <EOL> assert collection_id != <NUM_LIT:0> <EOL> assert self . md . collection_tools . deleter ( collection_id = collection_id ) == <NUM_LIT:1> <EOL> def test_collection_deletion_by_uk ( self ) : <EOL> kva = { '<STR_LIT>' : self . schema_id , <EOL> '<STR_LIT>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> collection_id = self . md . collection_tools . setter ( ** kva ) <EOL> assert collection_id != <NUM_LIT:0> <EOL> assert self . md . collection_tools . deleter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT:a>' ) == <NUM_LIT:1> <EOL> def test_collection_deletion_by_partial_uk ( self ) : <EOL> kva = { '<STR_LIT>' : self . schema_id , <EOL> '<STR_LIT>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> collection_id = self . md . collection_tools . setter ( ** kva ) <EOL> assert collection_id != <NUM_LIT:0> <EOL> assert self . md . collection_tools . deleter ( collection_name = '<STR_LIT:a>' ) == <NUM_LIT:0> <EOL> def test_collection_insert_update ( self ) : <EOL> old_rows = self . md . collection_tools . lister ( ) <EOL> kva = { '<STR_LIT>' : self . schema_id , <EOL> '<STR_LIT>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> kvb = { '<STR_LIT>' : self . schema_id , <EOL> '<STR_LIT>' : '<STR_LIT:b>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> assert self . md . collection_tools . setter ( ** kva ) != <NUM_LIT:0> <EOL> assert self . md . collection_tools . setter ( ** kvb ) != <NUM_LIT:0> <EOL> assert len ( self . md . collection_tools . lister ( ) ) == <NUM_LIT:3> <EOL> assert self . md . collection_tools . getter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT:b>' ) . collection_desc == '<STR_LIT>' <EOL> assert self . md . collection_tools . getter ( ** kva ) . collection_desc == '<STR_LIT>' <EOL> kvb [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> assert self . md . collection_tools . setter ( ** kvb ) == <NUM_LIT:0> <EOL> assert len ( self . md . collection_tools . lister ( ) ) == <NUM_LIT:3> <EOL> assert self . md . collection_tools . getter ( ** kvb ) . collection_desc == '<STR_LIT>' <EOL> kvb [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> assert self . md . collection_tools . setter ( ** kvb ) == <NUM_LIT:0> <EOL> assert len ( self . md . collection_tools . lister ( ) ) == <NUM_LIT:3> <EOL> assert self . md . collection_tools . getter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT:b>' ) . collection_desc == '<STR_LIT>' <EOL> assert self . md . collection_tools . getter ( schema_id = self . schema_id , <EOL> collection_name = '<STR_LIT:a>' ) . collection_desc == '<STR_LIT>' <EOL> def test_collection_insert_rejects ( self ) : <EOL> kv = { '<STR_LIT>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . schema_id } <EOL> orig_rowcount = len ( self . md . collection_tools . lister ( ) ) <EOL> old_rows = self . md . collection_tools . lister ( ) <EOL> kv2 = kv . copy ( ) <EOL> kv2 . pop ( '<STR_LIT>' ) <EOL> with pytest . raises ( KeyError ) : <EOL> self . md . collection_tools . setter ( ** kv2 ) <EOL> assert len ( self . md . collection_tools . lister ( ) ) == orig_rowcount <EOL> new_rows = self . md . collection_tools . lister ( ) <EOL> assert rowproxy_diff ( old_rows , new_rows , expected_add_cnt = <NUM_LIT:0> , expected_del_cnt = <NUM_LIT:0> ) <EOL> class TestField ( object ) : <EOL> def setup_method ( self , method ) : <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> self . md = mod . GristleMetaData ( self . tempdir ) <EOL> self . schema_id , self . collection_id = create_basic_metadata ( self . md ) <EOL> def teardown_method ( self , method ) : <EOL> os . remove ( os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> os . rmdir ( self . tempdir ) <EOL> def test_field_row_count ( self ) : <EOL> assert len ( self . md . field_tools . lister ( ) ) == <NUM_LIT:6> <EOL> def test_field_get ( self ) : <EOL> assert len ( self . md . field_tools . getter ( schema_id = self . schema_id , <EOL> collection_id = self . collection_id , <EOL> field_name = '<STR_LIT>' ) ) == <NUM_LIT:8> <EOL> def test_field_select_nonexisting_row ( self ) : <EOL> print '<STR_LIT>' <EOL> assert self . md . field_tools . getter ( schema_id = self . schema_id , <EOL> collection_id = self . collection_id , <EOL> field_name = '<STR_LIT>' ) is None <EOL> def test_field_select_missing_collection_id ( self ) : <EOL> with pytest . raises ( KeyError ) : <EOL> self . md . field_tools . getter ( schema_id = self . schema_id , field_name = '<STR_LIT>' ) <EOL> def test_field_update_with_invalid_element ( self ) : <EOL> field_keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> val_list = [ self . collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' , <NUM_LIT:15> , None ] <EOL> kv = dict ( zip ( field_keys , val_list ) ) <EOL> assert self . md . field_tools . setter ( ** kv ) > <NUM_LIT:0> <EOL> assert self . md . field_tools . setter ( ** kv ) == <NUM_LIT:0> <EOL> val_list = [ self . collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' , <NUM_LIT:15> , '<STR_LIT>' ] <EOL> kv = dict ( zip ( field_keys , val_list ) ) <EOL> with pytest . raises ( exc . IntegrityError ) : <EOL> self . md . field_tools . setter ( ** kv ) <EOL> class TestElement ( object ) : <EOL> def setup_method ( self , method ) : <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> self . md = mod . GristleMetaData ( self . tempdir ) <EOL> create_basic_metadata ( self . md ) <EOL> def teardown_method ( self , method ) : <EOL> os . remove ( os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> os . rmdir ( self . tempdir ) <EOL> def test_element_misc ( self ) : <EOL> kv = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:string>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> assert self . md . element_tools . setter ( ** kv ) != <NUM_LIT:0> <EOL> assert len ( self . md . element_tools . lister ( ) ) == <NUM_LIT:2> <EOL> assert self . md . element_tools . getter ( element_name = '<STR_LIT>' ) . element_len == <NUM_LIT> <EOL> kv [ '<STR_LIT>' ] = <NUM_LIT:100> <EOL> assert self . md . element_tools . setter ( ** kv ) == <NUM_LIT:0> <EOL> assert len ( self . md . element_tools . lister ( ) ) == <NUM_LIT:2> <EOL> assert self . md . element_tools . getter ( element_name = '<STR_LIT>' ) . element_len == <NUM_LIT:100> <EOL> class TestReports ( object ) : <EOL> def setup_method ( self , method ) : <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> self . md = mod . GristleMetaData ( self . tempdir ) <EOL> create_basic_metadata ( self . md ) <EOL> def teardown_method ( self , method ) : <EOL> os . remove ( os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> os . rmdir ( self . tempdir ) <EOL> def test_get_data_dictionary ( self ) : <EOL> results = self . md . get_data_dictionary ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> for row in results : <EOL> assert row . schema_name == '<STR_LIT>' <EOL> assert row . schema_desc == '<STR_LIT>' <EOL> assert row . collection_name == '<STR_LIT>' <EOL> def create_basic_metadata ( md ) : <EOL> """<STR_LIT>""" <EOL> schema_id = md . schema_tools . setter ( schema_name = '<STR_LIT>' , <EOL> schema_desc = '<STR_LIT>' ) <EOL> kv = { '<STR_LIT>' : schema_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:string>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> } <EOL> md . element_tools . setter ( ** kv ) <EOL> assert md . element_tools . getter ( element_name = '<STR_LIT>' ) . element_len == <NUM_LIT:2> <EOL> keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> v = [ schema_id , '<STR_LIT>' , '<STR_LIT>' ] <EOL> kv = dict ( zip ( keys , v ) ) <EOL> collection_id = md . collection_tools . setter ( ** kv ) <EOL> assert collection_id > <NUM_LIT:0> <EOL> field_keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> val_list = [ [ collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' , <NUM_LIT:15> , None ] , <EOL> [ collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' , <NUM_LIT:15> , None ] , <EOL> [ collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:int>' , None , None ] , <EOL> [ collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:int>' , None , None ] , <EOL> [ collection_id , '<STR_LIT>' , '<STR_LIT>' , None , None , '<STR_LIT>' ] , <EOL> [ collection_id , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' , <NUM_LIT> , None ] ] <EOL> for row in val_list : <EOL> kv = dict ( zip ( field_keys , row ) ) <EOL> assert md . field_tools . setter ( ** kv ) > <NUM_LIT:0> <EOL> return schema_id , collection_id <EOL> def rowproxy_diff ( old , new , expected_add_cnt = <NUM_LIT:0> , expected_del_cnt = <NUM_LIT:0> ) : <EOL> if expected_add_cnt != len ( [ x for x in new if x not in old ] ) : <EOL> return False <EOL> elif expected_del_cnt != len ( [ x for x in old if x not in new ] ) : <EOL> return False <EOL> return True <EOL> def content_rpt ( md ) : <EOL> """<STR_LIT>""" <EOL> rpt = '''<STR_LIT>''' <EOL> result = md . engine . execute ( rpt ) <EOL> print <EOL> print <EOL> print '<STR_LIT>' % ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for row in result : <EOL> print '<STR_LIT>' % ( row [ <NUM_LIT:0> ] , row [ <NUM_LIT:1> ] , row [ <NUM_LIT:2> ] , row [ <NUM_LIT:3> ] , row [ <NUM_LIT:4> ] , row [ <NUM_LIT:5> ] , row [ <NUM_LIT:6> ] ) <EOL> print </s>
