<s> try : <EOL> from builtins import range <EOL> except ImportError : <EOL> pass <EOL> import unittest <EOL> import tempfile <EOL> import os <EOL> from flask_blogging . sqlastorage import SQLAStorage <EOL> from sqlalchemy import create_engine <EOL> from test import FlaskBloggingTestCase <EOL> import sqlalchemy as sqla <EOL> from flask . ext . sqlalchemy import SQLAlchemy <EOL> import time <EOL> try : <EOL> import _mysql <EOL> HAS_MYSQL = True <EOL> except ImportError : <EOL> HAS_MYSQL = False <EOL> try : <EOL> import psycopg2 <EOL> HAS_POSTGRES = True <EOL> except ImportError : <EOL> HAS_POSTGRES = False <EOL> class TestSQLiteStorage ( FlaskBloggingTestCase ) : <EOL> def _create_storage ( self ) : <EOL> temp_dir = tempfile . gettempdir ( ) <EOL> self . _dbfile = os . path . join ( temp_dir , "<STR_LIT>" ) <EOL> self . _engine = create_engine ( '<STR_LIT>' + self . _dbfile ) <EOL> self . _meta = sqla . MetaData ( ) <EOL> self . storage = SQLAStorage ( self . _engine , metadata = self . _meta ) <EOL> self . _meta . create_all ( bind = self . _engine ) <EOL> def setUp ( self ) : <EOL> FlaskBloggingTestCase . setUp ( self ) <EOL> self . _create_storage ( ) <EOL> def tearDown ( self ) : <EOL> os . remove ( self . _dbfile ) <EOL> def test_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT:id>' , '<STR_LIT:title>' , '<STR_LIT:text>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_tag_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT:id>' , '<STR_LIT:text>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_tag_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_user_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_user_post_table_consistency ( self ) : <EOL> user_id = <NUM_LIT:1> <EOL> post_id = <NUM_LIT:5> <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT:user>" , tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> posts = self . storage . get_posts ( ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:1> ) <EOL> self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] , <EOL> post_id = pid ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:1> ) <EOL> return <EOL> def test_tags_uniqueness ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> with self . _engine . begin ( ) as conn : <EOL> statement = table . insert ( ) . values ( text = "<STR_LIT>" ) <EOL> conn . execute ( statement ) <EOL> with self . _engine . begin ( ) as conn : <EOL> statement = table . insert ( ) . values ( text = "<STR_LIT>" ) <EOL> self . assertRaises ( sqla . exc . IntegrityError , conn . execute , statement ) <EOL> def test_tags_consistency ( self ) : <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT:user>" , tags = tags ) <EOL> post = self . storage . get_post_by_id ( pid ) <EOL> self . assertEqual ( len ( post [ "<STR_LIT>" ] ) , <NUM_LIT:2> ) <EOL> tags . pop ( ) <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT:user>" , tags = tags , post_id = pid ) <EOL> post = self . storage . get_post_by_id ( pid ) <EOL> self . assertEqual ( len ( post [ "<STR_LIT>" ] ) , <NUM_LIT:1> ) <EOL> def test_tag_post_uniqueness ( self ) : <EOL> self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT:user>" , tags = [ "<STR_LIT>" ] ) <EOL> table_name = "<STR_LIT>" <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> with self . _engine . begin ( ) as conn : <EOL> statement = table . insert ( ) . values ( tag_id = <NUM_LIT:1> , post_id = <NUM_LIT:1> ) <EOL> self . assertRaises ( sqla . exc . IntegrityError , conn . execute , statement ) <EOL> def test_user_post_uniqueness ( self ) : <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> table_name = "<STR_LIT>" <EOL> metadata = sqla . MetaData ( ) <EOL> metadata . reflect ( bind = self . _engine ) <EOL> table = metadata . tables [ table_name ] <EOL> with self . _engine . begin ( ) as conn : <EOL> statement = table . insert ( ) . values ( user_id = "<STR_LIT>" , <EOL> post_id = pid ) <EOL> self . assertRaises ( sqla . exc . IntegrityError , conn . execute , statement ) <EOL> def test_bind_database ( self ) : <EOL> self . test_post_table_exists ( ) <EOL> self . test_tag_table_exists ( ) <EOL> self . test_tag_post_table_exists ( ) <EOL> self . test_user_post_table_exists ( ) <EOL> def test_save_post ( self ) : <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] , post_id = <NUM_LIT:1> ) <EOL> p = self . storage . get_post_by_id ( <NUM_LIT:2> ) <EOL> self . assertIsNone ( p ) <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] , <EOL> post_id = <NUM_LIT:5> ) <EOL> self . assertNotEqual ( pid , <NUM_LIT:5> ) <EOL> self . assertEqual ( pid , <NUM_LIT:2> ) <EOL> p = self . storage . get_post_by_id ( <NUM_LIT:2> ) <EOL> self . assertIsNotNone ( p ) <EOL> def test_delete_post ( self ) : <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> p = self . storage . get_post_by_id ( pid ) <EOL> self . assertIsNotNone ( p ) <EOL> self . storage . delete_post ( pid ) <EOL> p = self . storage . get_post_by_id ( pid ) <EOL> self . assertIsNone ( p ) <EOL> pid = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] , <EOL> post_id = <NUM_LIT:1> ) <EOL> p = self . storage . get_post_by_id ( pid ) <EOL> self . assertIsNotNone ( p ) <EOL> def test_get_post_by_id ( self ) : <EOL> pid1 = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> pid2 = self . storage . save_post ( title = "<STR_LIT>" , text = "<STR_LIT>" , <EOL> user_id = "<STR_LIT>" , <EOL> tags = [ "<STR_LIT:hello>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> post = self . storage . get_post_by_id ( pid1 ) <EOL> self . _assert_post ( post , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> post = self . storage . get_post_by_id ( pid2 ) <EOL> self . _assert_post ( post , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def _assert_post ( self , post , title , text , user_id , tags ) : <EOL> tags = set ( [ t . upper ( ) for t in tags ] ) <EOL> self . assertSetEqual ( set ( post [ "<STR_LIT>" ] ) , tags ) <EOL> self . assertEqual ( post [ "<STR_LIT:title>" ] , title ) <EOL> self . assertEqual ( post [ "<STR_LIT:text>" ] , text ) <EOL> self . assertEqual ( post [ "<STR_LIT>" ] , user_id ) <EOL> def test_get_posts ( self ) : <EOL> self . _create_dummy_data ( ) <EOL> posts = self . storage . get_posts ( ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctr -= <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( recent = False ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT:0> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT:hello>" ] ) <EOL> ctr += <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( count = <NUM_LIT:5> , offset = <NUM_LIT:5> , recent = False ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:5> ) <EOL> ctr = <NUM_LIT:5> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT:hello>" ] ) <EOL> ctr += <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( tag = "<STR_LIT:hello>" , recent = False ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT:0> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT:hello>" ] ) <EOL> ctr += <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( tag = "<STR_LIT>" , recent = False ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT:10> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctr += <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( user_id = "<STR_LIT>" , recent = True ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> ctr -= <NUM_LIT:1> <EOL> posts = self . storage . get_posts ( user_id = "<STR_LIT>" , recent = True ) <EOL> self . assertEqual ( len ( posts ) , <NUM_LIT:10> ) <EOL> ctr = <NUM_LIT:9> <EOL> for post in posts : <EOL> self . _assert_post ( post , "<STR_LIT>" % ctr , <EOL> "<STR_LIT>" % ctr , "<STR_LIT>" , [ "<STR_LIT:hello>" ] ) <EOL> ctr -= <NUM_LIT:1> <EOL> return <EOL> def test_count_posts ( self ) : <EOL> self . _create_dummy_data ( ) <EOL> count = self . storage . count_posts ( ) <EOL> self . assertEqual ( count , <NUM_LIT:20> ) <EOL> count = self . storage . count_posts ( user_id = "<STR_LIT>" ) <EOL> self . assertEqual ( count , <NUM_LIT:10> ) <EOL> count = self . storage . count_posts ( user_id = "<STR_LIT>" ) <EOL> self . assertEqual ( count , <NUM_LIT:10> ) <EOL> count = self . storage . count_posts ( user_id = "<STR_LIT>" ) <EOL> self . assertEqual ( count , <NUM_LIT:10> ) <EOL> count = self . storage . count_posts ( tag = "<STR_LIT:hello>" ) <EOL> self . assertEqual ( count , <NUM_LIT:10> ) <EOL> count = self . storage . count_posts ( tag = "<STR_LIT>" ) <EOL> self . assertEqual ( count , <NUM_LIT:10> ) <EOL> count = self . storage . count_posts ( user_id = "<STR_LIT>" , tag = "<STR_LIT>" ) <EOL> self . assertEqual ( count , <NUM_LIT:0> ) <EOL> def _create_dummy_data ( self ) : <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> tags = [ "<STR_LIT:hello>" ] if i < <NUM_LIT:10> else [ "<STR_LIT>" ] <EOL> user = "<STR_LIT>" if i < <NUM_LIT:10> else "<STR_LIT>" <EOL> self . storage . save_post ( title = "<STR_LIT>" % i , <EOL> text = "<STR_LIT>" % i , <EOL> user_id = user , tags = tags ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> @ unittest . skipUnless ( HAS_MYSQL , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class TestMySQLStorage ( TestSQLiteStorage ) : <EOL> def _create_storage ( self ) : <EOL> self . _engine = create_engine ( <EOL> "<STR_LIT>" ) <EOL> self . _meta = sqla . MetaData ( ) <EOL> self . storage = SQLAStorage ( self . _engine , metadata = self . _meta ) <EOL> self . _meta . create_all ( bind = self . _engine ) <EOL> def tearDown ( self ) : <EOL> metadata = sqla . MetaData ( ) <EOL> metadata . reflect ( bind = self . _engine ) <EOL> metadata . drop_all ( bind = self . _engine ) <EOL> @ unittest . skipUnless ( HAS_POSTGRES , "<STR_LIT>" ) <EOL> class TestPostgresStorage ( TestSQLiteStorage ) : <EOL> def _create_storage ( self ) : <EOL> self . _engine = create_engine ( <EOL> "<STR_LIT>" ) <EOL> self . _meta = sqla . MetaData ( ) <EOL> self . storage = SQLAStorage ( self . _engine , metadata = self . _meta ) <EOL> self . _meta . create_all ( bind = self . _engine ) <EOL> def tearDown ( self ) : <EOL> metadata = sqla . MetaData ( ) <EOL> metadata . reflect ( bind = self . _engine ) <EOL> metadata . drop_all ( bind = self . _engine ) <EOL> class TestSQLiteBinds ( FlaskBloggingTestCase ) : <EOL> def _conn_string ( self , dbfile ) : <EOL> return '<STR_LIT>' + dbfile <EOL> def setUp ( self ) : <EOL> FlaskBloggingTestCase . setUp ( self ) <EOL> temp_dir = tempfile . gettempdir ( ) <EOL> self . _dbfile = os . path . join ( temp_dir , "<STR_LIT>" ) <EOL> conn_string = self . _conn_string ( self . _dbfile ) <EOL> self . app . config [ "<STR_LIT>" ] = { <EOL> '<STR_LIT>' : conn_string <EOL> } <EOL> self . _db = SQLAlchemy ( self . app ) <EOL> self . storage = SQLAStorage ( db = self . _db , bind = "<STR_LIT>" ) <EOL> self . _engine = self . _db . get_engine ( self . app , bind = "<STR_LIT>" ) <EOL> self . _meta = self . _db . metadata <EOL> self . _db . create_all ( bind = [ "<STR_LIT>" ] ) <EOL> def tearDown ( self ) : <EOL> os . remove ( self . _dbfile ) <EOL> def test_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT:id>' , '<STR_LIT:title>' , '<STR_LIT:text>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_tag_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT:id>' , '<STR_LIT:text>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_tag_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> def test_user_post_table_exists ( self ) : <EOL> table_name = "<STR_LIT>" <EOL> with self . _engine . begin ( ) as conn : <EOL> self . assertTrue ( conn . dialect . has_table ( conn , table_name ) ) <EOL> metadata = self . _meta <EOL> table = metadata . tables [ table_name ] <EOL> columns = [ t . name for t in table . columns ] <EOL> expected_columns = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertListEqual ( columns , expected_columns ) <EOL> @ unittest . skipUnless ( HAS_MYSQL , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class TestMySQLBinds ( TestSQLiteBinds ) : <EOL> def _conn_string ( self , dbfile ) : <EOL> return "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> pass <EOL> @ unittest . skipUnless ( HAS_POSTGRES , "<STR_LIT>" ) <EOL> class TestPostgresBinds ( TestSQLiteBinds ) : <EOL> def _conn_string ( self , dbfile ) : <EOL> return "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> pass </s>
