<s> import datetime <EOL> import logging <EOL> import itertools <EOL> import time <EOL> import mock <EOL> import contextlib2 as contextlib <EOL> import pytest <EOL> from sqlalchemy import create_engine <EOL> from sqlalchemy . orm import sessionmaker <EOL> from vcr . cassette import Cassette , CassetteContextDecorator <EOL> from . import util <EOL> from okcupyd import db , settings <EOL> from okcupyd import util as okcupyd_util <EOL> from okcupyd . db import model , adapters <EOL> log = logging . getLogger ( __name__ ) <EOL> BREAK_ON_EXCEPTION = False <EOL> def pytest_addoption ( parser ) : <EOL> okcupyd_util . add_command_line_options ( <EOL> parser . addoption , use_short_options = False <EOL> ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> default = False , help = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" , default = False ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> action = '<STR_LIT:store_true>' , default = False , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store>' , <EOL> default = '<STR_LIT>' , help = "<STR_LIT>" ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> default = False , help = "<STR_LIT>" ) <EOL> parser . addoption ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> default = False , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def patch ( option , * patches , ** kwargs ) : <EOL> negate = kwargs . get ( '<STR_LIT>' , False ) <EOL> @ pytest . yield_fixture ( autouse = True , scope = '<STR_LIT>' ) <EOL> def patch_conditionally ( request ) : <EOL> condition = bool ( request . config . getoption ( option ) ) <EOL> if negate : <EOL> condition = not condition <EOL> if condition : <EOL> with contextlib . ExitStack ( ) as exit_stack : <EOL> for patch in patches : <EOL> exit_stack . enter_context ( patch ) <EOL> yield <EOL> else : <EOL> yield <EOL> return patch_conditionally <EOL> patch_settings = patch ( '<STR_LIT>' , <EOL> mock . patch . object ( settings , '<STR_LIT>' , '<STR_LIT:username>' ) , <EOL> mock . patch . object ( settings , '<STR_LIT>' , '<STR_LIT:password>' ) , <EOL> mock . patch . object ( settings , '<STR_LIT>' , '<STR_LIT:username>' ) , <EOL> mock . patch . object ( settings , '<STR_LIT>' , '<STR_LIT:password>' ) , <EOL> negate = True ) <EOL> patch_save = patch ( <EOL> '<STR_LIT>' , mock . patch . object ( <EOL> Cassette , '<STR_LIT>' , okcupyd_util . curry ( Cassette . _save ) ( force = True ) <EOL> ) <EOL> ) <EOL> def reraise_exception ( obj , error_type , error , traceback ) : <EOL> if error is not None : <EOL> raise <EOL> patch_use_cassette_enabled = patch ( <EOL> '<STR_LIT>' , <EOL> mock . patch . object ( CassetteContextDecorator , '<STR_LIT>' , <EOL> lambda stuff : log . debug ( "<STR_LIT>" ) ) , <EOL> mock . patch . object ( CassetteContextDecorator , '<STR_LIT>' , <EOL> reraise_exception ) <EOL> ) <EOL> patch_vcrpy_filters = patch ( <EOL> '<STR_LIT>' , mock . patch . object ( util , '<STR_LIT>' , True ) <EOL> ) <EOL> patch_break_on_exception = patch ( <EOL> '<STR_LIT>' , <EOL> mock . patch ( '<STR_LIT>' , True ) <EOL> ) <EOL> original_load = Cassette . _load <EOL> def new_load ( self ) : <EOL> self . dirty = False <EOL> self . rewound = True <EOL> patch_record = patch ( <EOL> '<STR_LIT>' , mock . patch . object ( <EOL> Cassette , '<STR_LIT>' , new_load <EOL> ) <EOL> ) <EOL> @ pytest . fixture ( autouse = True , scope = '<STR_LIT>' ) <EOL> def set_vcr_options ( request ) : <EOL> util . okcupyd_vcr . record_mode = request . config . getoption ( '<STR_LIT>' ) <EOL> @ pytest . fixture ( autouse = True , scope = '<STR_LIT>' ) <EOL> def process_command_line_args ( request ) : <EOL> okcupyd_util . handle_command_line_options ( request . config . option ) <EOL> @ pytest . fixture ( scope = '<STR_LIT>' ) <EOL> def engine ( request ) : <EOL> return create_engine ( '<STR_LIT>' , echo = request . config . getoption ( '<STR_LIT>' ) ) <EOL> @ pytest . fixture ( scope = '<STR_LIT>' ) <EOL> def session ( engine ) : <EOL> new_kwargs = db . Session . kw . copy ( ) <EOL> new_kwargs [ '<STR_LIT>' ] = engine <EOL> return sessionmaker ( ** new_kwargs ) <EOL> @ pytest . yield_fixture ( autouse = True , scope = '<STR_LIT>' ) <EOL> def setup_db ( engine , session ) : <EOL> with mock . patch . object ( db , '<STR_LIT>' , session ) : <EOL> orig_engine = db . Base . metadata . bind <EOL> db . Base . metadata . bind = engine <EOL> db . Base . metadata . drop_all ( ) <EOL> db . Base . metadata . create_all ( ) <EOL> yield <EOL> db . Base . metadata . bind = orig_engine <EOL> @ pytest . fixture <EOL> def T ( mock_profile_builder , mock_message_thread_builder , mock_message_builder ) : <EOL> class testing ( object ) : <EOL> pass <EOL> testing . ensure = mock . Mock ( ) <EOL> testing . build_mock = mock . Mock ( ) <EOL> testing . factory = mock . Mock ( ) <EOL> def ensure_thread_model_resembles_okcupyd_thread ( <EOL> thread_model , okcupyd_thread <EOL> ) : <EOL> assert thread_model . okc_id == okcupyd_thread . id <EOL> ensure_user_model_resembles_okcupyd_profile ( <EOL> thread_model . initiator , okcupyd_thread . initiator <EOL> ) <EOL> ensure_user_model_resembles_okcupyd_profile ( <EOL> thread_model . respondent , okcupyd_thread . respondent <EOL> ) <EOL> for pair in zip ( thread_model . messages , okcupyd_thread . messages ) : <EOL> ensure_message_model_resembles_okcupyd_message ( * pair ) <EOL> assert len ( thread_model . messages ) == len ( okcupyd_thread . messages ) <EOL> def ensure_message_model_resembles_okcupyd_message ( <EOL> message_model , okcupyd_message <EOL> ) : <EOL> assert message_model . okc_id == okcupyd_message . id <EOL> assert message_model . sender . handle == okcupyd_message . sender . username <EOL> assert message_model . recipient . handle == okcupyd_message . recipient . username <EOL> assert message_model . text == okcupyd_message . content <EOL> def ensure_user_model_resembles_okcupyd_profile ( user_model , <EOL> okcupyd_profile ) : <EOL> assert user_model . handle == okcupyd_profile . username <EOL> testing . ensure . user_model_resembles_okcupyd_profile = ensure_user_model_resembles_okcupyd_profile <EOL> testing . ensure . message_model_resembles_okcupyd_message = ensure_message_model_resembles_okcupyd_message <EOL> testing . ensure . thread_model_resembles_okcupyd_thread = ensure_thread_model_resembles_okcupyd_thread <EOL> testing . build_mock . thread = mock_message_thread_builder <EOL> testing . build_mock . message = mock_message_builder <EOL> testing . build_mock . profile = mock_profile_builder <EOL> def build_message_thread ( ) : <EOL> message_thread = testing . build_mock . thread ( ) <EOL> return adapters . ThreadAdapter ( message_thread ) . get_thread ( ) [ <NUM_LIT:0> ] <EOL> def build_user ( username ) : <EOL> profile = testing . build_mock . profile ( username ) <EOL> return adapters . UserAdapter ( profile ) . get ( ) <EOL> def build_okcupyd_user ( user ) : <EOL> user_model = model . User . from_profile ( user . profile ) <EOL> user_model . upsert_model ( id_key = '<STR_LIT>' ) <EOL> okcupyd_user = model . OKCupydUser ( user_id = user_model . id ) <EOL> okcupyd_user . upsert_model ( id_key = '<STR_LIT>' ) <EOL> return okcupyd_user <EOL> testing . factory . message_thread = build_message_thread <EOL> testing . factory . user = build_user <EOL> testing . factory . okcupyd_user = build_okcupyd_user <EOL> return testing <EOL> @ pytest . fixture <EOL> def mock_profile_builder ( ) : <EOL> counter = itertools . count ( ) <EOL> next ( counter ) <EOL> username_to_profile = { } <EOL> def _build_mock_profile ( username = '<STR_LIT:username>' , age = <NUM_LIT:30> , id = None , <EOL> location = '<STR_LIT>' , ** kwargs ) : <EOL> if username in username_to_profile : <EOL> return username_to_profile [ username ] <EOL> if id is None : <EOL> id = next ( counter ) <EOL> mock_profile = mock . MagicMock ( id = id or next ( counter ) , location = location , <EOL> age = age , username = username , ** kwargs ) <EOL> username_to_profile [ username ] = mock_profile <EOL> return mock_profile <EOL> return _build_mock_profile <EOL> @ pytest . fixture <EOL> def mock_message_builder ( ) : <EOL> counter = itertools . count ( ) <EOL> next ( counter ) <EOL> def _build_mock_message ( id = None , sender = '<STR_LIT>' , recipient = '<STR_LIT>' , <EOL> content = '<STR_LIT:content>' , ** kwargs ) : <EOL> kwargs . setdefault ( '<STR_LIT>' , <EOL> datetime . datetime ( year = <NUM_LIT> , day = <NUM_LIT:2> , month = <NUM_LIT:4> ) ) <EOL> if id is None : <EOL> id = next ( counter ) <EOL> assert isinstance ( sender , str ) <EOL> assert isinstance ( recipient , str ) <EOL> return mock . MagicMock ( id = id , sender = mock . Mock ( username = sender ) , <EOL> recipient = mock . Mock ( username = recipient ) , <EOL> content = content , ** kwargs ) <EOL> return _build_mock_message <EOL> @ pytest . fixture <EOL> def mock_message_thread_builder ( mock_message_builder , mock_profile_builder ) : <EOL> counter = itertools . count ( ) <EOL> next ( counter ) <EOL> def _build_mock_message_thread ( id = None , message_count = <NUM_LIT:2> , <EOL> initiator = '<STR_LIT>' , <EOL> respondent = '<STR_LIT>' , ** kwargs ) : <EOL> if id is None : <EOL> id = next ( counter ) <EOL> messages = [ mock_message_builder ( content = '<STR_LIT>' . format ( i ) , <EOL> sender = respondent <EOL> if i % <NUM_LIT:2> else initiator , <EOL> recipient = initiator <EOL> if i % <NUM_LIT:2> else respondent ) <EOL> for i in range ( message_count ) ] <EOL> kwargs . setdefault ( '<STR_LIT>' , <EOL> datetime . datetime ( year = <NUM_LIT> , day = <NUM_LIT:2> , month = <NUM_LIT:4> ) ) <EOL> message_thread = mock . MagicMock ( id = id , messages = messages , ** kwargs ) <EOL> message_thread . initiator = mock_profile_builder ( initiator ) <EOL> message_thread . respondent = mock_profile_builder ( respondent ) <EOL> return message_thread <EOL> return _build_mock_message_thread <EOL> def pytest_exception_interact ( ) : <EOL> if BREAK_ON_EXCEPTION : <EOL> import ipdb ; ipdb . set_trace ( ) <EOL> try : <EOL> models = [ ] <EOL> for key in dir ( model ) : <EOL> prospect = getattr ( model , key ) <EOL> if isinstance ( prospect , type ) and issubclass ( prospect , db . Base ) : <EOL> models . append ( prospect ) <EOL> model_class_to_instances = { } <EOL> for model_class in models : <EOL> model_class_to_instances [ model_class ] = { <EOL> instance . okc_id : instance for instance in model_class . query ( ) <EOL> } <EOL> log . info ( model_class_to_instances ) <EOL> except : <EOL> pass <EOL> @ pytest . fixture <EOL> def vcr_live_sleep ( request ) : <EOL> return ( time . sleep <EOL> if ( request . config . getoption ( '<STR_LIT>' ) or <EOL> request . config . getoption ( '<STR_LIT>' ) ) <EOL> else mock . Mock ( ) ) </s>
