<s> from __future__ import print_function , unicode_literals <EOL> import os <EOL> import unittest <EOL> import time <EOL> import mocker <EOL> from test . util . wlcontext import wlcontext <EOL> from test . util . module_disposer import case_uses_module <EOL> import test . unit . configuration as configuration_module <EOL> import weblab . core . server as UserProcessingServer <EOL> import weblab . core . server as core_api <EOL> import weblab . core . user_processor as UserProcessor <EOL> import weblab . core . reservations as Reservation <EOL> import weblab . core . coordinator . confirmer as Confirmer <EOL> import voodoo . configuration as ConfigurationManager <EOL> from weblab . data import ValidDatabaseSessionId <EOL> import weblab . data . server_type as ServerType <EOL> from weblab . core . coordinator . gateway import create as coordinator_create , SQLALCHEMY <EOL> import weblab . core . coordinator . config_parser as CoordinationConfigurationParser <EOL> import weblab . core . exc as coreExc <EOL> from weblab . data . command import Command <EOL> from weblab . data . experiments import ExperimentId , ExperimentUsage , CommandSent , FileSent , FinishedReservationResult <EOL> import weblab . data . dto . experiments as Category <EOL> import weblab . data . dto . experiments as Experiment <EOL> import weblab . data . dto . experiments as ExperimentAllowed <EOL> import voodoo . sessions . session_id as SessionId <EOL> from voodoo . gen import CoordAddress <EOL> laboratory_coordaddr = CoordAddress . translate ( <EOL> "<STR_LIT>" <EOL> ) <EOL> @ case_uses_module ( UserProcessingServer ) <EOL> @ case_uses_module ( Confirmer ) <EOL> class UserProcessingServerTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . coord_address = CoordAddress . translate ( "<STR_LIT>" ) <EOL> self . cfg_manager = ConfigurationManager . ConfigurationManager ( ) <EOL> self . cfg_manager . append_module ( configuration_module ) <EOL> self . cfg_manager . _set_value ( CoordinationConfigurationParser . COORDINATOR_LABORATORY_SERVERS , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . cfg_manager . _set_value ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> self . mocker = mocker . Mocker ( ) <EOL> self . lab_mock = self . mocker . mock ( ) <EOL> self . locator = FakeLocator ( self . lab_mock ) <EOL> coordinator = coordinator_create ( SQLALCHEMY , self . locator , self . cfg_manager ) <EOL> coordinator . _clean ( ) <EOL> coordinator . stop ( ) <EOL> self . ups = WrappedUPS ( <EOL> self . coord_address , <EOL> self . locator , <EOL> self . cfg_manager <EOL> ) <EOL> self . ups . _db_manager . _delete_all_uses ( ) <EOL> def tearDown ( self ) : <EOL> self . ups . stop ( ) <EOL> def test_reserve_session ( self ) : <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> session_manager = self . ups . _session_manager <EOL> sess = session_manager . get_session ( sess_id ) <EOL> self . assertEquals ( sess [ '<STR_LIT>' ] . username , db_sess_id . username ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> core_api . logout ( ) <EOL> def test_list_experiments ( self ) : <EOL> db_sess_id1 = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id1 , _ = self . ups . _reserve_session ( db_sess_id1 ) <EOL> with wlcontext ( self . ups , session_id = sess_id1 ) : <EOL> experiments = core_api . list_experiments ( ) <EOL> self . assertLessEqual ( <NUM_LIT:5> , len ( experiments ) ) <EOL> experiment_names = list ( ( experiment . experiment . name for experiment in experiments ) ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> core_api . logout ( ) <EOL> db_sess_id2 = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id2 , _ = self . ups . _reserve_session ( db_sess_id2 ) <EOL> with wlcontext ( self . ups , session_id = sess_id2 ) : <EOL> experiments = core_api . list_experiments ( ) <EOL> self . assertEquals ( <NUM_LIT:7> , len ( experiments ) ) <EOL> experiment_names = list ( ( experiment . experiment . name for experiment in experiments ) ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> self . assertTrue ( '<STR_LIT>' in experiment_names ) <EOL> core_api . logout ( ) <EOL> def test_get_user_information ( self ) : <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> user = core_api . get_user_information ( ) <EOL> self . assertEquals ( "<STR_LIT>" , user . login ) <EOL> self . assertEquals ( "<STR_LIT>" , user . full_name ) <EOL> self . assertEquals ( "<STR_LIT>" , user . email ) <EOL> core_api . logout ( ) <EOL> def test_get_reservation_info ( self ) : <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> exp_id = ExperimentId ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> lab_sess_id = SessionId . SessionId ( "<STR_LIT>" ) <EOL> self . lab_mock . reserve_experiment ( exp_id , "<STR_LIT:{}>" ) <EOL> self . mocker . result ( lab_sess_id ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . lab_mock . resolve_experiment_address ( lab_sess_id ) <EOL> self . mocker . result ( CoordAddress . translate ( '<STR_LIT>' ) ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . mocker . replay ( ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> reservation = core_api . reserve_experiment ( exp_id , "<STR_LIT:{}>" , "<STR_LIT:{}>" ) <EOL> with wlcontext ( self . ups , reservation_id = reservation . reservation_id ) : <EOL> reservation_info = core_api . get_reservation_info ( ) <EOL> self . assertEquals ( '<STR_LIT>' , reservation_info . exp_name ) <EOL> self . assertEquals ( '<STR_LIT>' , reservation_info . cat_name ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> core_api . logout ( ) <EOL> def test_reserve_experiment ( self ) : <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> exp_id = ExperimentId ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> self . assertRaises ( <EOL> coreExc . UnknownExperimentIdError , <EOL> core_api . reserve_experiment , exp_id , "<STR_LIT:{}>" , "<STR_LIT:{}>" ) <EOL> exp_id = ExperimentId ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> lab_sess_id = SessionId . SessionId ( "<STR_LIT>" ) <EOL> self . lab_mock . reserve_experiment ( exp_id , "<STR_LIT:{}>" ) <EOL> self . mocker . result ( lab_sess_id ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . lab_mock . resolve_experiment_address ( lab_sess_id ) <EOL> self . mocker . result ( CoordAddress . translate ( '<STR_LIT>' ) ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . mocker . replay ( ) <EOL> reservation = core_api . reserve_experiment ( exp_id , "<STR_LIT:{}>" , "<STR_LIT:{}>" ) <EOL> self . assertTrue ( isinstance ( reservation , Reservation . Reservation ) ) <EOL> core_api . logout ( ) <EOL> def test_get_experiment_use_by_id_found ( self ) : <EOL> reservations , usages = self . _store_two_reservations ( ) <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> finished_result = core_api . get_experiment_use_by_id ( reservations [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( finished_result . is_finished ( ) ) <EOL> finished_result . experiment_use . experiment_use_id = None <EOL> self . assertEquals ( FinishedReservationResult ( usages [ <NUM_LIT:0> ] . load_files ( '<STR_LIT:.>' ) ) , finished_result ) <EOL> def test_get_experiment_uses_by_id_found ( self ) : <EOL> reservations , usages = self . _store_two_reservations ( ) <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> experiment_results = core_api . get_experiment_uses_by_id ( reservations ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( experiment_results ) ) <EOL> self . assertTrue ( experiment_results [ <NUM_LIT:0> ] . is_finished ( ) ) <EOL> experiment_results [ <NUM_LIT:0> ] . experiment_use . experiment_use_id = None <EOL> self . assertEquals ( FinishedReservationResult ( usages [ <NUM_LIT:0> ] . load_files ( '<STR_LIT:.>' ) ) , experiment_results [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( experiment_results [ <NUM_LIT:1> ] . is_forbidden ( ) ) <EOL> def test_get_experiment_uses_by_id_notfound ( self ) : <EOL> reservations , usages = self . _store_two_reservations ( ) <EOL> reservation1 = self . _reserve_experiment ( ) <EOL> reservation2 = self . _reserve_experiment ( ) <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> experiment_results = core_api . get_experiment_uses_by_id ( ( reservations [ <NUM_LIT:0> ] , reservation1 , reservation2 ) ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( experiment_results ) ) <EOL> self . assertTrue ( experiment_results [ <NUM_LIT:0> ] . is_finished ( ) ) <EOL> experiment_results [ <NUM_LIT:0> ] . experiment_use . experiment_use_id = None <EOL> self . assertEquals ( FinishedReservationResult ( usages [ <NUM_LIT:0> ] . load_files ( '<STR_LIT:.>' ) ) , experiment_results [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( experiment_results [ <NUM_LIT:1> ] . is_alive ( ) ) <EOL> self . assertTrue ( experiment_results [ <NUM_LIT:2> ] . is_alive ( ) ) <EOL> def _reserve_experiment ( self ) : <EOL> db_sess_id = ValidDatabaseSessionId ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> sess_id , _ = self . ups . _reserve_session ( db_sess_id ) <EOL> with wlcontext ( self . ups , session_id = sess_id ) : <EOL> exp_id = ExperimentId ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> lab_sess_id = SessionId . SessionId ( "<STR_LIT>" ) <EOL> self . lab_mock . reserve_experiment ( exp_id , "<STR_LIT:{}>" ) <EOL> self . mocker . result ( lab_sess_id ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . lab_mock . resolve_experiment_address ( lab_sess_id ) <EOL> self . mocker . result ( CoordAddress . translate ( '<STR_LIT>' ) ) <EOL> self . mocker . count ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . mocker . replay ( ) <EOL> reservation = core_api . reserve_experiment ( exp_id , "<STR_LIT:{}>" , "<STR_LIT:{}>" ) <EOL> return reservation . reservation_id <EOL> def _store_two_reservations ( self ) : <EOL> db_gw = self . ups . _db_manager <EOL> session = db_gw . Session ( ) <EOL> try : <EOL> student1 = db_gw . _get_user ( session , '<STR_LIT>' ) <EOL> student2 = db_gw . _get_user ( session , '<STR_LIT>' ) <EOL> finally : <EOL> session . close ( ) <EOL> reservation_id1 = SessionId . SessionId ( u'<STR_LIT:5>' ) <EOL> initial_usage1 = ExperimentUsage ( ) <EOL> initial_usage1 . start_date = time . time ( ) <EOL> initial_usage1 . end_date = time . time ( ) <EOL> initial_usage1 . from_ip = u"<STR_LIT>" <EOL> initial_usage1 . experiment_id = ExperimentId ( u"<STR_LIT>" , u"<STR_LIT>" ) <EOL> initial_usage1 . coord_address = CoordAddress ( u"<STR_LIT>" , u"<STR_LIT>" , u"<STR_LIT>" ) <EOL> initial_usage1 . reservation_id = reservation_id1 . id <EOL> initial_usage1 . request_info = { '<STR_LIT>' : '<STR_LIT:user>' , '<STR_LIT>' : student1 . id } <EOL> valid_file_path = os . path . relpath ( os . sep . join ( ( '<STR_LIT:test>' , '<STR_LIT>' ) ) ) <EOL> file1 = FileSent ( valid_file_path , u'<STR_LIT>' , time . time ( ) ) <EOL> file2 = FileSent ( valid_file_path , u'<STR_LIT>' , <EOL> time . time ( ) , Command ( u'<STR_LIT>' ) , <EOL> time . time ( ) , file_info = u'<STR_LIT>' ) <EOL> command1 = CommandSent ( Command ( u"<STR_LIT>" ) , time . time ( ) ) <EOL> command2 = CommandSent ( Command ( u"<STR_LIT>" ) , time . time ( ) , <EOL> Command ( u"<STR_LIT>" ) , time . time ( ) ) <EOL> initial_usage1 . append_command ( command1 ) <EOL> initial_usage1 . append_command ( command2 ) <EOL> initial_usage1 . append_file ( file1 ) <EOL> initial_usage1 . append_file ( file2 ) <EOL> reservation_id2 = SessionId . SessionId ( u'<STR_LIT>' ) <EOL> initial_usage2 = ExperimentUsage ( ) <EOL> initial_usage2 . start_date = time . time ( ) <EOL> initial_usage2 . end_date = time . time ( ) <EOL> initial_usage2 . from_ip = u"<STR_LIT>" <EOL> initial_usage2 . experiment_id = ExperimentId ( u"<STR_LIT>" , u"<STR_LIT>" ) <EOL> initial_usage2 . coord_address = CoordAddress ( u"<STR_LIT>" , u"<STR_LIT>" , u"<STR_LIT>" ) <EOL> initial_usage2 . reservation_id = reservation_id2 . id <EOL> initial_usage2 . request_info = { '<STR_LIT>' : '<STR_LIT:user>' , '<STR_LIT>' : student2 . id } <EOL> file1 = FileSent ( valid_file_path , u'<STR_LIT>' , time . time ( ) ) <EOL> file2 = FileSent ( valid_file_path , u'<STR_LIT>' , <EOL> time . time ( ) , Command ( u'<STR_LIT>' ) , <EOL> time . time ( ) , file_info = u'<STR_LIT>' ) <EOL> command1 = CommandSent ( Command ( u"<STR_LIT>" ) , time . time ( ) ) <EOL> command2 = CommandSent ( Command ( u"<STR_LIT>" ) , time . time ( ) , <EOL> Command ( u"<STR_LIT>" ) , time . time ( ) ) <EOL> initial_usage2 . append_command ( command1 ) <EOL> initial_usage2 . append_command ( command2 ) <EOL> initial_usage2 . append_file ( file1 ) <EOL> initial_usage2 . append_file ( file2 ) <EOL> self . ups . _db_manager . store_experiment_usage ( '<STR_LIT>' , initial_usage1 ) <EOL> self . ups . _db_manager . store_experiment_usage ( '<STR_LIT>' , initial_usage2 ) <EOL> return ( reservation_id1 , reservation_id2 ) , ( initial_usage1 , initial_usage2 ) <EOL> class FakeLocator ( object ) : <EOL> def __init__ ( self , lab ) : <EOL> self . lab = lab <EOL> def get_server ( self , server_type ) : <EOL> if server_type == ServerType . Laboratory : <EOL> return self . lab <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def get_server_from_coordaddr ( self , coord_addr , server_type ) : <EOL> if server_type == ServerType . Laboratory and laboratory_coordaddr == coord_addr : <EOL> return self . lab <EOL> raise Exception ( "<STR_LIT>" ) <EOL> class FakeFacade ( object ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> pass <EOL> def start ( self ) : <EOL> pass <EOL> def stop ( self ) : <EOL> pass <EOL> class WrappedUPS ( UserProcessingServer . UserProcessingServer ) : <EOL> FACADE_SERVER = FakeFacade <EOL> def generate_experiment ( exp_name , exp_cat_name ) : <EOL> cat = Category . ExperimentCategory ( exp_cat_name ) <EOL> client = Experiment . ExperimentClient ( "<STR_LIT>" , { } ) <EOL> exp = Experiment . Experiment ( exp_name , cat , '<STR_LIT>' , '<STR_LIT>' , client ) <EOL> return exp <EOL> def generate_experiment_allowed ( time_allowed , exp_name , exp_cat_name ) : <EOL> exp = generate_experiment ( exp_name , exp_cat_name ) <EOL> return ExperimentAllowed . ExperimentAllowed ( exp , time_allowed , <NUM_LIT:5> , True , '<STR_LIT>' % exp_name , <NUM_LIT:1> , '<STR_LIT:user>' ) <EOL> def suite ( ) : <EOL> return unittest . makeSuite ( UserProcessingServerTestCase ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
