<s> import os <EOL> import sqlite3 <EOL> import tempfile <EOL> import time <EOL> from datetime import datetime <EOL> from datetime import timedelta <EOL> from pushmanager . core import db <EOL> def create_temp_db_file ( ) : <EOL> fd , db_file_path = tempfile . mkstemp ( suffix = "<STR_LIT>" ) <EOL> os . close ( fd ) <EOL> return db_file_path <EOL> def get_temp_db_uri ( dbfile = None ) : <EOL> if not dbfile : <EOL> dbfile = create_temp_db_file ( ) <EOL> return "<STR_LIT>" + dbfile <EOL> def make_test_db ( dbfile = None ) : <EOL> if not dbfile : <EOL> dbfile = create_temp_db_file ( ) <EOL> testsql = open ( <EOL> os . path . join ( <EOL> os . path . dirname ( __file__ ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> ) . read ( ) <EOL> test_db = sqlite3 . connect ( dbfile ) <EOL> test_db . cursor ( ) . executescript ( testsql ) <EOL> test_db . commit ( ) <EOL> test_db . close ( ) <EOL> return dbfile <EOL> class FakeDataMixin ( object ) : <EOL> now = time . time ( ) <EOL> yesterday = time . mktime ( ( datetime . now ( ) - timedelta ( days = <NUM_LIT:1> ) ) . timetuple ( ) ) <EOL> push_data = [ <EOL> [ <NUM_LIT:10> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:abc>' , '<STR_LIT>' , yesterday , now , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ <NUM_LIT:11> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , now , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ <NUM_LIT:12> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , now , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , now , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> ] <EOL> push_keys = [ <EOL> '<STR_LIT:id>' , '<STR_LIT:title>' , '<STR_LIT:user>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:state>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> fake_revision = "<STR_LIT:0>" * <NUM_LIT> <EOL> request_data = [ <EOL> [ <NUM_LIT:10> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> now , now , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' , fake_revision , '<STR_LIT>' ] , <EOL> [ <NUM_LIT:11> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , now , <EOL> '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' , fake_revision , '<STR_LIT>' ] , <EOL> [ <NUM_LIT:12> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , <EOL> now , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' , fake_revision , '<STR_LIT>' ] , <EOL> [ <NUM_LIT> , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , now , <EOL> now , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' , fake_revision , '<STR_LIT>' ] , <EOL> ] <EOL> request_keys = [ <EOL> '<STR_LIT:id>' , '<STR_LIT:user>' , '<STR_LIT:state>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:title>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:description>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> def on_db_return ( self , success , db_results ) : <EOL> assert success <EOL> def make_push_dict ( self , data ) : <EOL> return dict ( zip ( self . push_keys , data ) ) <EOL> def make_request_dict ( self , data ) : <EOL> return dict ( zip ( self . request_keys , data ) ) <EOL> def insert_pushes ( self ) : <EOL> push_queries = [ ] <EOL> for pd in self . push_data : <EOL> push_queries . append ( db . push_pushes . insert ( self . make_push_dict ( pd ) ) ) <EOL> db . execute_transaction_cb ( push_queries , self . on_db_return ) <EOL> def insert_requests ( self ) : <EOL> request_queries = [ ] <EOL> for rd in self . request_data : <EOL> request_queries . append ( db . push_requests . insert ( self . make_request_dict ( rd ) ) ) <EOL> db . execute_transaction_cb ( request_queries , self . on_db_return ) <EOL> def insert_pushcontent ( self , requestid , pushid ) : <EOL> db . execute_cb ( <EOL> db . push_pushcontents . insert ( { '<STR_LIT>' : requestid , '<STR_LIT>' : pushid } ) , <EOL> self . on_db_return <EOL> ) <EOL> def get_push_for_request ( self , requestid ) : <EOL> pushid = [ None ] <EOL> def on_select_return ( success , db_results ) : <EOL> assert success <EOL> _ , pushid [ <NUM_LIT:0> ] = db_results . fetchone ( ) <EOL> first_pushcontent_query = db . push_pushcontents . select ( <EOL> db . push_pushcontents . c . request == requestid <EOL> ) <EOL> db . execute_cb ( first_pushcontent_query , on_select_return ) <EOL> return pushid [ <NUM_LIT:0> ] <EOL> def get_pushes ( self ) : <EOL> pushes = [ None ] <EOL> def on_select_return ( success , db_results ) : <EOL> assert success <EOL> pushes [ <NUM_LIT:0> ] = db_results . fetchall ( ) <EOL> db . execute_cb ( db . push_pushes . select ( ) , on_select_return ) <EOL> return pushes [ <NUM_LIT:0> ] <EOL> def get_requests ( self ) : <EOL> requests = [ None ] <EOL> def on_select_return ( success , db_results ) : <EOL> assert success <EOL> requests [ <NUM_LIT:0> ] = db_results . fetchall ( ) <EOL> db . execute_cb ( db . push_requests . select ( ) , on_select_return ) <EOL> return requests [ <NUM_LIT:0> ] <EOL> def get_requests_by_user ( self , user ) : <EOL> return [ req for req in self . get_requests ( ) if req [ '<STR_LIT:user>' ] == user ] </s>
