<s> import os <EOL> import pytest <EOL> import pandas as pd <EOL> import ibis . compat as compat <EOL> from . test_comms import double_ex <EOL> from ibis . tasks import IbisTaskMessage , IbisTaskExecutor <EOL> from ibis . util import guid <EOL> from ibis . wire import BytesIO <EOL> import ibis . wire as wire <EOL> from ibis . compat import unittest <EOL> from ibis . tests . test_server import WorkerTestFixture <EOL> try : <EOL> from ibis . comms import SharedMmap , IPCLock , IbisTableWriter <EOL> SKIP_TESTS = False <EOL> except ImportError : <EOL> SKIP_TESTS = True <EOL> pytestmark = pytest . mark . skipif ( SKIP_TESTS or compat . PY3 , <EOL> reason = '<STR_LIT>' ) <EOL> class TestTasks ( unittest . TestCase ) : <EOL> def test_message_encode_decode ( self ) : <EOL> task = IbisTaskMessage ( <NUM_LIT> , '<STR_LIT:foo>' , <NUM_LIT:12> , <NUM_LIT:1000> ) <EOL> encoded = task . encode ( ) <EOL> decoded = IbisTaskMessage . decode ( encoded ) <EOL> encoded2 = decoded . encode ( ) <EOL> self . assertEqual ( encoded , encoded2 ) <EOL> attrs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for attr in attrs : <EOL> self . assertEqual ( getattr ( task , attr ) , getattr ( decoded , attr ) ) <EOL> class TestPingPongTask ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . paths_to_delete = [ ] <EOL> path , size , offset = '<STR_LIT>' % guid ( ) , <NUM_LIT> , <NUM_LIT:0> <EOL> self . paths_to_delete . append ( path ) <EOL> self . lock = IPCLock ( is_slave = <NUM_LIT:0> ) <EOL> self . task = IbisTaskMessage ( self . lock . semaphore_id , path , <EOL> offset , size ) <EOL> self . mm = SharedMmap ( path , size , create = True ) <EOL> wire . PackedMessageWriter ( self . mm ) . string ( '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> for path in self . paths_to_delete : <EOL> try : <EOL> os . remove ( path ) <EOL> except os . error : <EOL> pass <EOL> def test_execute_task ( self ) : <EOL> _execute_task ( self . task , self . lock ) <EOL> self . mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( self . mm ) <EOL> assert reader . uint8 ( ) <EOL> assert reader . string ( ) == '<STR_LIT>' <EOL> def _execute_task ( task , master_lock ) : <EOL> executor = IbisTaskExecutor ( task ) <EOL> try : <EOL> executor . execute ( ) <EOL> except : <EOL> executor . lock . release ( ) <EOL> raise <EOL> master_lock . acquire ( ) <EOL> class TestTaskE2E ( TestPingPongTask , WorkerTestFixture ) : <EOL> def setUp ( self ) : <EOL> TestPingPongTask . setUp ( self ) <EOL> WorkerTestFixture . setUp ( self ) <EOL> def tearDown ( self ) : <EOL> TestPingPongTask . tearDown ( self ) <EOL> WorkerTestFixture . tearDown ( self ) <EOL> def test_task_end_to_end ( self ) : <EOL> msg = self . _run_task ( self . task ) <EOL> assert msg == '<STR_LIT>' <EOL> self . mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( self . mm ) <EOL> assert reader . uint8 ( ) <EOL> assert reader . string ( ) == '<STR_LIT>' <EOL> def test_task_return_exception ( self ) : <EOL> self . mm . seek ( <NUM_LIT:0> ) <EOL> wire . PackedMessageWriter ( self . mm ) . string ( '<STR_LIT>' ) <EOL> msg = self . _run_task ( self . task ) <EOL> assert msg == '<STR_LIT>' <EOL> self . mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( self . mm ) <EOL> assert reader . uint8 ( ) == <NUM_LIT:0> <EOL> error_msg = reader . string ( ) <EOL> assert '<STR_LIT>' in error_msg <EOL> def _run_task ( self , task ) : <EOL> encoded_task = task . encode ( ) <EOL> worker_port , worker_pid = self . _spawn_worker ( ) <EOL> sock = self . _connect ( worker_port ) <EOL> sock . send ( encoded_task ) <EOL> msg = sock . recv ( <NUM_LIT> ) <EOL> self . lock . acquire ( ) <EOL> return msg <EOL> def delete_all_guid_files ( ) : <EOL> import glob <EOL> import os <EOL> [ os . remove ( x ) for x in glob . glob ( '<STR_LIT:*>' ) if len ( x ) == <NUM_LIT:32> ] <EOL> class NRows ( object ) : <EOL> def __init__ ( self ) : <EOL> self . total = <NUM_LIT:0> <EOL> def update ( self , values ) : <EOL> self . total += len ( values ) <EOL> def merge ( self , other ) : <EOL> self . total += other . total <EOL> return self <EOL> def finalize ( self ) : <EOL> return self . total <EOL> class Summ ( object ) : <EOL> def __init__ ( self ) : <EOL> self . total = <NUM_LIT:0> <EOL> def update ( self , values ) : <EOL> import pandas as pd <EOL> self . total += pd . Series ( values ) . sum ( ) <EOL> def merge ( self , other ) : <EOL> self . total += other . total <EOL> return self <EOL> def finalize ( self ) : <EOL> return self . total <EOL> class TestAggregateTasks ( unittest . TestCase ) : <EOL> def _get_mean_uda ( self ) : <EOL> class Mean ( object ) : <EOL> def __init__ ( self ) : <EOL> self . total = <NUM_LIT:0> <EOL> self . count = <NUM_LIT:0> <EOL> def update ( self , values ) : <EOL> values = pd . Series ( values ) <EOL> self . total += values . sum ( ) <EOL> self . count += values . count ( ) <EOL> def merge ( self , other ) : <EOL> self . total += other . total <EOL> self . count += other . count <EOL> return self <EOL> def finalize ( self ) : <EOL> return self . total / float ( self . count ) <EOL> return Mean <EOL> def setUp ( self ) : <EOL> self . paths_to_delete = [ ] <EOL> self . col_fragments = [ double_ex ( <NUM_LIT:1000> ) for _ in range ( <NUM_LIT:10> ) ] <EOL> self . lock = IPCLock ( is_slave = <NUM_LIT:0> ) <EOL> def test_update ( self ) : <EOL> klass = self . _get_mean_uda ( ) <EOL> col = self . col_fragments [ <NUM_LIT:0> ] <EOL> task , mm = self . _make_update_task ( klass , [ col ] ) <EOL> _execute_task ( task , self . lock ) <EOL> mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( mm ) <EOL> if not reader . uint8 ( ) : <EOL> raise Exception ( reader . string ( ) ) <EOL> result = compat . pickle_load ( reader . string ( ) ) <EOL> ex_total = pd . Series ( col . to_numpy_for_pandas ( ) ) . sum ( ) <EOL> assert result . total == ex_total <EOL> col = self . col_fragments [ <NUM_LIT:1> ] <EOL> task , mm = self . _make_update_task ( klass , [ col ] , prior_state = result ) <EOL> self . lock . release ( ) <EOL> _execute_task ( task , self . lock ) <EOL> mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( mm ) <EOL> if not reader . uint8 ( ) : <EOL> raise Exception ( reader . string ( ) ) <EOL> result = compat . pickle_load ( reader . string ( ) ) <EOL> ex_total += pd . Series ( col . to_numpy_for_pandas ( ) ) . sum ( ) <EOL> assert ex_total != <NUM_LIT:0> <EOL> assert result . total == ex_total <EOL> def test_merge ( self ) : <EOL> klass = self . _get_mean_uda ( ) <EOL> lcol = self . col_fragments [ <NUM_LIT:0> ] <EOL> rcol = self . col_fragments [ <NUM_LIT:1> ] <EOL> left = self . _update ( klass , [ lcol ] ) <EOL> right = self . _update ( klass , [ rcol ] ) <EOL> task , mm = self . _make_merge_task ( left , right ) <EOL> _execute_task ( task , self . lock ) <EOL> mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( mm ) <EOL> if not reader . uint8 ( ) : <EOL> raise Exception ( reader . string ( ) ) <EOL> result = compat . pickle_load ( reader . string ( ) ) <EOL> larr = lcol . to_numpy_for_pandas ( ) <EOL> rarr = rcol . to_numpy_for_pandas ( ) <EOL> assert larr is not None <EOL> ex_total = ( pd . Series ( larr ) . sum ( ) + pd . Series ( rarr ) . sum ( ) ) <EOL> assert result . total == ex_total <EOL> def test_finalize ( self ) : <EOL> klass = self . _get_mean_uda ( ) <EOL> col = self . col_fragments [ <NUM_LIT:0> ] <EOL> result = self . _update ( klass , [ col ] ) <EOL> task , mm = self . _make_finalize_task ( result ) <EOL> _execute_task ( task , self . lock ) <EOL> mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( mm ) <EOL> if not reader . uint8 ( ) : <EOL> raise Exception ( reader . string ( ) ) <EOL> result = compat . pickle_load ( reader . string ( ) ) <EOL> arr = col . to_numpy_for_pandas ( ) <EOL> ex_result = pd . Series ( arr ) . mean ( ) <EOL> assert result == ex_result <EOL> def _update ( self , klass , args ) : <EOL> task , mm = self . _make_update_task ( klass , args ) <EOL> _execute_task ( task , self . lock ) <EOL> self . lock . release ( ) <EOL> mm . seek ( <NUM_LIT:0> ) <EOL> reader = wire . PackedMessageReader ( mm ) <EOL> if not reader . uint8 ( ) : <EOL> raise Exception ( reader . string ( ) ) <EOL> return reader . string ( ) <EOL> def _make_update_task ( self , uda_class , cols , prior_state = None ) : <EOL> payload = BytesIO ( ) <EOL> msg_writer = wire . PackedMessageWriter ( payload ) <EOL> msg_writer . string ( '<STR_LIT>' ) <EOL> msg_writer . string ( compat . pickle_dump ( uda_class ) ) <EOL> if prior_state is not None : <EOL> msg_writer . uint8 ( <NUM_LIT:1> ) <EOL> msg_writer . string ( compat . pickle_dump ( prior_state ) ) <EOL> else : <EOL> msg_writer . uint8 ( <NUM_LIT:0> ) <EOL> writer = IbisTableWriter ( cols ) <EOL> path = '<STR_LIT>' % guid ( ) <EOL> size = writer . total_size ( ) + payload . tell ( ) <EOL> offset = <NUM_LIT:0> <EOL> mm = SharedMmap ( path , size , create = True ) <EOL> self . paths_to_delete . append ( path ) <EOL> mm . write ( payload . getvalue ( ) ) <EOL> writer . write ( mm ) <EOL> task = IbisTaskMessage ( self . lock . semaphore_id , path , offset , size ) <EOL> return task , mm <EOL> def _make_merge_task ( self , left_pickled , right_pickled ) : <EOL> payload = BytesIO ( ) <EOL> msg_writer = wire . PackedMessageWriter ( payload ) <EOL> msg_writer . string ( '<STR_LIT>' ) <EOL> msg_writer . string ( left_pickled ) <EOL> msg_writer . string ( right_pickled ) <EOL> path = '<STR_LIT>' % guid ( ) <EOL> size = payload . tell ( ) <EOL> offset = <NUM_LIT:0> <EOL> mm = SharedMmap ( path , size , create = True ) <EOL> self . paths_to_delete . append ( path ) <EOL> mm . write ( payload . getvalue ( ) ) <EOL> task = IbisTaskMessage ( self . lock . semaphore_id , path , offset , size ) <EOL> return task , mm <EOL> def _make_finalize_task ( self , pickled ) : <EOL> payload = BytesIO ( ) <EOL> msg_writer = wire . PackedMessageWriter ( payload ) <EOL> msg_writer . string ( '<STR_LIT>' ) <EOL> msg_writer . string ( pickled ) <EOL> path = '<STR_LIT>' % guid ( ) <EOL> size = payload . tell ( ) <EOL> offset = <NUM_LIT:0> <EOL> mm = SharedMmap ( path , size , create = True ) <EOL> self . paths_to_delete . append ( path ) <EOL> mm . write ( payload . getvalue ( ) ) <EOL> task = IbisTaskMessage ( self . lock . semaphore_id , path , offset , size ) <EOL> return task , mm <EOL> def tearDown ( self ) : <EOL> for path in self . paths_to_delete : <EOL> try : <EOL> os . remove ( path ) <EOL> except os . error : <EOL> pass </s>
