<s> from __future__ import absolute_import , print_function , division <EOL> import sqlite3 <EOL> from tempfile import NamedTemporaryFile <EOL> from petl . compat import next <EOL> from petl . test . helpers import ieq , eq_ <EOL> from petl . io . db import fromdb , todb , appenddb <EOL> def test_fromdb ( ) : <EOL> data = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> connection = sqlite3 . connect ( '<STR_LIT>' ) <EOL> c = connection . cursor ( ) <EOL> c . execute ( '<STR_LIT>' ) <EOL> for row in data : <EOL> c . execute ( '<STR_LIT>' , row ) <EOL> connection . commit ( ) <EOL> c . close ( ) <EOL> actual = fromdb ( connection , '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> ieq ( expect , actual ) <EOL> ieq ( expect , actual ) <EOL> i1 = iter ( actual ) <EOL> i2 = iter ( actual ) <EOL> eq_ ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , next ( i1 ) ) <EOL> eq_ ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , next ( i1 ) ) <EOL> eq_ ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , next ( i2 ) ) <EOL> eq_ ( ( '<STR_LIT:b>' , <NUM_LIT:2> ) , next ( i1 ) ) <EOL> def test_fromdb_mkcursor ( ) : <EOL> data = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> connection = sqlite3 . connect ( '<STR_LIT>' ) <EOL> c = connection . cursor ( ) <EOL> c . execute ( '<STR_LIT>' ) <EOL> for row in data : <EOL> c . execute ( '<STR_LIT>' , row ) <EOL> connection . commit ( ) <EOL> c . close ( ) <EOL> mkcursor = lambda : connection . cursor ( ) <EOL> actual = fromdb ( mkcursor , '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> ieq ( expect , actual ) <EOL> ieq ( expect , actual ) <EOL> i1 = iter ( actual ) <EOL> i2 = iter ( actual ) <EOL> eq_ ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , next ( i1 ) ) <EOL> eq_ ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , next ( i1 ) ) <EOL> eq_ ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , next ( i2 ) ) <EOL> eq_ ( ( '<STR_LIT:b>' , <NUM_LIT:2> ) , next ( i1 ) ) <EOL> def test_fromdb_withargs ( ) : <EOL> data = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> connection = sqlite3 . connect ( '<STR_LIT>' ) <EOL> c = connection . cursor ( ) <EOL> c . execute ( '<STR_LIT>' ) <EOL> for row in data : <EOL> c . execute ( '<STR_LIT>' , row ) <EOL> connection . commit ( ) <EOL> c . close ( ) <EOL> actual = fromdb ( <EOL> connection , <EOL> '<STR_LIT>' , <EOL> ( <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> ) <EOL> expect = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT> ) ) <EOL> ieq ( expect , actual ) <EOL> ieq ( expect , actual ) <EOL> def test_todb_appenddb ( ) : <EOL> f = NamedTemporaryFile ( delete = False ) <EOL> conn = sqlite3 . connect ( f . name ) <EOL> conn . execute ( '<STR_LIT>' ) <EOL> conn . commit ( ) <EOL> table = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) ) <EOL> todb ( table , conn , '<STR_LIT>' ) <EOL> actual = conn . execute ( '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) ) <EOL> ieq ( expect , actual ) <EOL> table2 = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:d>' , <NUM_LIT:7> ) , <EOL> ( '<STR_LIT:e>' , <NUM_LIT:9> ) , <EOL> ( '<STR_LIT:f>' , <NUM_LIT:1> ) ) <EOL> appenddb ( table2 , conn , '<STR_LIT>' ) <EOL> actual = conn . execute ( '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:d>' , <NUM_LIT:7> ) , <EOL> ( '<STR_LIT:e>' , <NUM_LIT:9> ) , <EOL> ( '<STR_LIT:f>' , <NUM_LIT:1> ) ) <EOL> ieq ( expect , actual ) <EOL> def test_todb_appenddb_cursor ( ) : <EOL> f = NamedTemporaryFile ( delete = False ) <EOL> conn = sqlite3 . connect ( f . name ) <EOL> conn . execute ( '<STR_LIT>' ) <EOL> conn . commit ( ) <EOL> table = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) ) <EOL> cursor = conn . cursor ( ) <EOL> todb ( table , cursor , '<STR_LIT>' ) <EOL> actual = conn . execute ( '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) ) <EOL> ieq ( expect , actual ) <EOL> table2 = ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , <EOL> ( '<STR_LIT:d>' , <NUM_LIT:7> ) , <EOL> ( '<STR_LIT:e>' , <NUM_LIT:9> ) , <EOL> ( '<STR_LIT:f>' , <NUM_LIT:1> ) ) <EOL> appenddb ( table2 , cursor , '<STR_LIT>' ) <EOL> actual = conn . execute ( '<STR_LIT>' ) <EOL> expect = ( ( '<STR_LIT:a>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT:b>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:c>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT:d>' , <NUM_LIT:7> ) , <EOL> ( '<STR_LIT:e>' , <NUM_LIT:9> ) , <EOL> ( '<STR_LIT:f>' , <NUM_LIT:1> ) ) <EOL> ieq ( expect , actual ) </s>
