<s> """<STR_LIT>""" <EOL> import cPickle <EOL> import hashlib <EOL> import logging <EOL> from math import pi <EOL> from multiprocessing import AuthenticationError <EOL> from multiprocessing . managers import RemoteError <EOL> import os <EOL> import shutil <EOL> import socket <EOL> import sys <EOL> import traceback <EOL> import tempfile <EOL> import unittest <EOL> from Crypto . Random import get_random_bytes <EOL> from traits . api import CTrait <EOL> from openmdao . main . api import Assembly , Component , Container , Driver , set_as_top <EOL> from openmdao . main . container import get_closest_proxy <EOL> from openmdao . main . hasobjective import HasObjectives <EOL> from openmdao . main . hasparameters import HasParameters <EOL> from openmdao . main . interfaces import IComponent <EOL> from openmdao . main . mp_support import has_interface , is_instance <EOL> from openmdao . main . mp_util import read_server_config <EOL> from openmdao . main . objserverfactory import connect , start_server , RemoteFile <EOL> from openmdao . main . rbac import Credentials , get_credentials , set_credentials , AccessController , RoleError , rbac <EOL> from openmdao . main . datatypes . api import Float , Int , FileRef <EOL> from openmdao . lib . casehandlers . api import ListCaseRecorder <EOL> from openmdao . test . execcomp import ExecComp <EOL> from openmdao . util . decorators import add_delegate <EOL> from openmdao . util . publickey import get_key_pair <EOL> from openmdao . util . testutil import assert_raises , assert_rel_error <EOL> from openmdao . util . fileutil import onerror <EOL> _MODULE = '<STR_LIT>' <EOL> _SERVER_ID = <NUM_LIT:0> <EOL> class Box ( ExecComp ) : <EOL> """<STR_LIT>""" <EOL> pid = Int ( iotype = '<STR_LIT>' ) <EOL> def __init__ ( self ) : <EOL> super ( Box , self ) . __init__ ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> self . pid = os . getpid ( ) <EOL> sub = self . add ( '<STR_LIT>' , Container ( ) ) <EOL> sub . add ( '<STR_LIT>' , Int ( ) ) <EOL> def execute ( self ) : <EOL> print '<STR_LIT>' % ( self . width , self . height , self . depth , <EOL> socket . gethostname ( ) , self . pid ) <EOL> sys . stdout . flush ( ) <EOL> super ( Box , self ) . execute ( ) <EOL> def no_rbac ( self ) : <EOL> pass <EOL> @ rbac ( '<STR_LIT>' , proxy_types = [ RemoteFile ] ) <EOL> def open_in_parent ( self , path , mode ) : <EOL> try : <EOL> return self . parent . open ( path , mode ) <EOL> except Exception as exc : <EOL> self . _logger . debug ( '<STR_LIT>' , exc ) <EOL> self . _logger . debug ( traceback . format_exc ( ) ) <EOL> @ rbac ( '<STR_LIT>' ) <EOL> def cause_parent_error1 ( self ) : <EOL> return self . parent . no_such_variable <EOL> @ rbac ( '<STR_LIT>' ) <EOL> def cause_parent_error2 ( self ) : <EOL> return self . parent . get_trait ( '<STR_LIT>' ) <EOL> @ rbac ( '<STR_LIT>' ) <EOL> def cause_parent_error3 ( self ) : <EOL> return self . parent . xyzzy ( ) <EOL> class HollowSphere ( Component ) : <EOL> """<STR_LIT>""" <EOL> radius = Float ( <NUM_LIT:1.0> , low = <NUM_LIT:0.> , exclude_low = True , iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> thickness = Float ( <NUM_LIT> , iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> inner_volume = Float ( iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> volume = Float ( iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> solid_volume = Float ( iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> surface_area = Float ( iotype = '<STR_LIT>' , units = '<STR_LIT>' ) <EOL> pid = Int ( iotype = '<STR_LIT>' ) <EOL> def __init__ ( self ) : <EOL> super ( HollowSphere , self ) . __init__ ( ) <EOL> self . pid = os . getpid ( ) <EOL> def execute ( self ) : <EOL> self . surface_area = <NUM_LIT> * pi * self . radius * self . radius <EOL> self . inner_volume = <NUM_LIT> / <NUM_LIT> * pi * self . radius ** <NUM_LIT:3> <EOL> self . volume = <NUM_LIT> / <NUM_LIT> * pi * ( self . radius + self . thickness ) ** <NUM_LIT:3> <EOL> self . solid_volume = self . volume - self . inner_volume <EOL> @ add_delegate ( HasParameters ) <EOL> @ add_delegate ( HasObjectives ) <EOL> class BoxDriver ( Driver ) : <EOL> """<STR_LIT>""" <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> for width in range ( <NUM_LIT:1> , <NUM_LIT:2> ) : <EOL> for height in range ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> for depth in range ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> self . _logger . debug ( '<STR_LIT>' , width , height , depth ) <EOL> self . set_parameters ( ( float ( width ) , float ( height ) , float ( depth ) ) ) <EOL> self . run_iteration ( ) <EOL> volume , area = self . eval_objectives ( ) <EOL> self . _logger . debug ( '<STR_LIT>' , volume , area ) <EOL> class BoxSource ( ExecComp ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( BoxSource , self ) . __init__ ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> sub = self . add ( '<STR_LIT>' , Container ( ) ) <EOL> sub . add ( '<STR_LIT>' , Int ( ) ) <EOL> class BoxSink ( ExecComp ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( BoxSink , self ) . __init__ ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> class Model ( Assembly ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , box ) : <EOL> super ( Model , self ) . __init__ ( ) <EOL> self . recorders = [ ListCaseRecorder ( ) ] <EOL> self . add ( '<STR_LIT>' , BoxDriver ( ) ) <EOL> self . driver . workflow . add ( self . add ( '<STR_LIT:source>' , BoxSource ( ) ) . name ) <EOL> self . driver . workflow . add ( self . add ( '<STR_LIT>' , box ) . name ) <EOL> self . driver . workflow . add ( self . add ( '<STR_LIT>' , BoxSink ( ) ) . name ) <EOL> self . driver . add_parameter ( '<STR_LIT>' , low = <NUM_LIT> , high = <NUM_LIT> ) <EOL> self . driver . add_parameter ( '<STR_LIT>' , low = <NUM_LIT> , high = <NUM_LIT> ) <EOL> self . driver . add_parameter ( '<STR_LIT>' , low = <NUM_LIT> , high = <NUM_LIT> ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . driver . add_objective ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . driver . add_objective ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ rbac ( '<STR_LIT>' , proxy_types = [ RemoteFile ] ) <EOL> def open ( self , path , mode ) : <EOL> """<STR_LIT>""" <EOL> return RemoteFile ( open ( path , mode ) ) <EOL> @ rbac ( '<STR_LIT>' ) <EOL> def xyzzy ( self ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> class Protector ( AccessController ) : <EOL> """<STR_LIT>""" <EOL> def check_access ( self , role , methodname , obj , attr ) : <EOL> if not role : <EOL> raise RoleError ( '<STR_LIT>' ) <EOL> if role == '<STR_LIT>' : <EOL> return <EOL> if methodname != '<STR_LIT>' and self . user_attribute ( obj , attr ) : <EOL> return <EOL> raise RoleError ( "<STR_LIT>" <EOL> % ( methodname , attr , role ) ) <EOL> @ staticmethod <EOL> def user_attribute ( obj , attr ) : <EOL> if attr in obj . list_inputs ( ) or attr in obj . list_outputs ( ) or attr in ( '<STR_LIT>' , '<STR_LIT:name>' ) : <EOL> return True <EOL> return False <EOL> class ProtectedBox ( Box ) : <EOL> """<STR_LIT>""" <EOL> secret = Int ( ) <EOL> def __init__ ( self ) : <EOL> super ( ProtectedBox , self ) . __init__ ( ) <EOL> self . protector = Protector ( ) <EOL> @ rbac ( '<STR_LIT>' ) <EOL> def proprietary_method ( self ) : <EOL> pass <EOL> def get_access_controller ( self ) : <EOL> return self . protector <EOL> @ rbac ( ( '<STR_LIT>' , '<STR_LIT:user>' ) , proxy_types = [ FileRef ] ) <EOL> def get ( self , path ) : <EOL> if self . protector . user_attribute ( self , path ) : <EOL> return super ( ProtectedBox , self ) . get ( path ) <EOL> raise RoleError ( '<STR_LIT>' % path ) <EOL> @ rbac ( ( '<STR_LIT>' , '<STR_LIT:user>' ) , proxy_types = [ CTrait ] ) <EOL> def get_dyn_trait ( self , name , iotype = None , trait = None ) : <EOL> if self . protector . user_attribute ( self , name ) : <EOL> return super ( ProtectedBox , self ) . get_dyn_trait ( name , iotype , trait ) <EOL> raise RoleError ( '<STR_LIT>' % name ) <EOL> @ rbac ( ( '<STR_LIT>' , '<STR_LIT:user>' ) ) <EOL> def get_attr_w_copy ( self , name ) : <EOL> if self . protector . user_attribute ( self , name ) : <EOL> return super ( ProtectedBox , self ) . get_attr ( name ) <EOL> raise RoleError ( '<STR_LIT>' % name ) <EOL> @ rbac ( ( '<STR_LIT>' , '<STR_LIT:user>' ) ) <EOL> def set ( self , path , value ) : <EOL> if self . protector . user_attribute ( self , path ) : <EOL> return super ( ProtectedBox , self ) . set ( path , value ) <EOL> raise RoleError ( '<STR_LIT>' % path ) <EOL> class TestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def run ( self , result = None ) : <EOL> """<STR_LIT>""" <EOL> self . test_result = result or unittest . TestResult ( ) <EOL> return super ( TestCase , self ) . run ( self . test_result ) <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . startdir = os . getcwd ( ) <EOL> self . tempdir = tempfile . mkdtemp ( prefix = '<STR_LIT>' ) <EOL> os . chdir ( self . tempdir ) <EOL> self . factories = [ ] <EOL> self . servers = [ ] <EOL> self . server_dirs = [ ] <EOL> def start_factory ( self , port = None , allowed_users = None ) : <EOL> """<STR_LIT>""" <EOL> global _SERVER_ID <EOL> _SERVER_ID += <NUM_LIT:1> <EOL> server_dir = '<STR_LIT>' % _SERVER_ID <EOL> if os . path . exists ( server_dir ) : <EOL> shutil . rmtree ( server_dir , onerror = onerror ) <EOL> os . mkdir ( server_dir ) <EOL> os . chdir ( server_dir ) <EOL> self . server_dirs . append ( server_dir ) <EOL> try : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' , os . getpid ( ) ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> if port is None : <EOL> port = - <NUM_LIT:1> if _SERVER_ID & <NUM_LIT:1> else <NUM_LIT:0> <EOL> if allowed_users is None : <EOL> credentials = get_credentials ( ) <EOL> allowed_users = { credentials . user : credentials . public_key } <EOL> allowed_types = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> server , server_cfg = start_server ( port = port , <EOL> allowed_users = allowed_users , <EOL> allowed_types = allowed_types , <EOL> log_prefix = server_dir ) <EOL> self . servers . append ( server ) <EOL> cfg = read_server_config ( server_cfg ) <EOL> self . address = cfg [ '<STR_LIT:address>' ] <EOL> self . port = cfg [ '<STR_LIT:port>' ] <EOL> self . tunnel = cfg [ '<STR_LIT>' ] <EOL> self . key = cfg [ '<STR_LIT:key>' ] <EOL> logging . debug ( '<STR_LIT>' , server . pid ) <EOL> logging . debug ( '<STR_LIT>' , self . address ) <EOL> logging . debug ( '<STR_LIT>' , self . port ) <EOL> logging . debug ( '<STR_LIT>' , self . key ) <EOL> finally : <EOL> os . chdir ( '<STR_LIT:..>' ) <EOL> factory = connect ( self . address , self . port , self . tunnel , pubkey = self . key ) <EOL> self . factories . append ( factory ) <EOL> logging . debug ( '<STR_LIT>' , factory ) <EOL> return factory <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> for factory in self . factories : <EOL> factory . cleanup ( ) <EOL> for server in self . servers : <EOL> logging . debug ( '<STR_LIT>' , server . pid ) <EOL> server . terminate ( timeout = <NUM_LIT:10> ) <EOL> os . chdir ( self . startdir ) <EOL> if not os . environ . get ( '<STR_LIT>' , False ) : <EOL> try : <EOL> shutil . rmtree ( self . tempdir ) <EOL> except OSError : <EOL> pass <EOL> def test_1_client ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> factory = self . start_factory ( ) <EOL> types = factory . get_available_types ( ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> for typname , version in types : <EOL> logging . debug ( '<STR_LIT>' , typname , version ) <EOL> obj = factory . create ( _MODULE + '<STR_LIT>' ) <EOL> sphere_pid = obj . get ( '<STR_LIT>' ) <EOL> self . assertNotEqual ( sphere_pid , os . getpid ( ) ) <EOL> radius = obj . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( radius , <NUM_LIT:1.> ) <EOL> radius += <NUM_LIT:1> <EOL> obj . set ( '<STR_LIT>' , radius ) <EOL> new_radius = obj . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( new_radius , <NUM_LIT> ) <EOL> self . assertEqual ( obj . get ( '<STR_LIT>' ) , <NUM_LIT:0.> ) <EOL> self . assertEqual ( obj . get ( '<STR_LIT>' ) , <NUM_LIT:0.> ) <EOL> self . assertEqual ( obj . get ( '<STR_LIT>' ) , <NUM_LIT:0.> ) <EOL> self . assertEqual ( obj . get ( '<STR_LIT>' ) , <NUM_LIT:0.> ) <EOL> obj . run ( ) <EOL> assert_rel_error ( self , obj . get ( '<STR_LIT>' ) , <NUM_LIT> , <NUM_LIT> ) <EOL> assert_rel_error ( self , obj . get ( '<STR_LIT>' ) , <NUM_LIT> , <NUM_LIT> ) <EOL> assert_rel_error ( self , obj . get ( '<STR_LIT>' ) , <NUM_LIT> , <NUM_LIT> ) <EOL> assert_rel_error ( self , obj . get ( '<STR_LIT>' ) , <NUM_LIT> , <NUM_LIT> ) <EOL> try : <EOL> obj . set ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> except RemoteError as exc : <EOL> fragment = "<STR_LIT>" <EOL> if fragment not in str ( exc ) : <EOL> self . fail ( '<STR_LIT>' % ( fragment , exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> obj = factory . create ( _MODULE + '<STR_LIT>' ) <EOL> box_pid = obj . get ( '<STR_LIT>' ) <EOL> self . assertNotEqual ( box_pid , os . getpid ( ) ) <EOL> self . assertNotEqual ( box_pid , sphere_pid ) <EOL> obj . width += <NUM_LIT:2> <EOL> obj . height += <NUM_LIT:2> <EOL> obj . depth += <NUM_LIT:2> <EOL> self . assertEqual ( obj . width , <NUM_LIT> ) <EOL> self . assertEqual ( obj . height , <NUM_LIT> ) <EOL> self . assertEqual ( obj . depth , <NUM_LIT> ) <EOL> self . assertEqual ( obj . volume , <NUM_LIT:0.> ) <EOL> self . assertEqual ( obj . surface_area , <NUM_LIT:0.> ) <EOL> obj . run ( ) <EOL> self . assertEqual ( obj . volume , <NUM_LIT> ) <EOL> self . assertEqual ( obj . surface_area , <NUM_LIT> ) <EOL> try : <EOL> obj . no_rbac ( ) <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_2_model ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> factory = self . start_factory ( ) <EOL> box = factory . create ( _MODULE + '<STR_LIT>' ) <EOL> model = set_as_top ( Model ( box ) ) <EOL> model . run ( ) <EOL> for width in range ( <NUM_LIT:1> , <NUM_LIT:2> ) : <EOL> for height in range ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> for depth in range ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> case = model . recorders [ <NUM_LIT:0> ] . cases . pop ( <NUM_LIT:0> ) <EOL> self . assertEqual ( case . get_output ( '<STR_LIT>' ) , <EOL> width * height * depth ) <EOL> self . assertTrue ( is_instance ( model . box . parent , Assembly ) ) <EOL> self . assertTrue ( has_interface ( model . box . parent , IComponent ) ) <EOL> source = model . box . parent . source <EOL> self . assertEqual ( source . width_in , <NUM_LIT:1.> ) <EOL> obj , path = get_closest_proxy ( model , '<STR_LIT>' ) <EOL> self . assertEqual ( obj , model . box ) <EOL> self . assertEqual ( path , '<STR_LIT>' ) <EOL> obj , path = get_closest_proxy ( model , '<STR_LIT>' ) <EOL> self . assertEqual ( obj , model . source . subcontainer . subvar ) <EOL> self . assertEqual ( path , '<STR_LIT>' ) <EOL> obj , path = get_closest_proxy ( model . source . subcontainer , '<STR_LIT>' ) <EOL> self . assertEqual ( obj , model . source . subcontainer . subvar ) <EOL> self . assertEqual ( path , '<STR_LIT>' ) <EOL> tmp = model . box . open_in_parent ( '<STR_LIT>' , '<STR_LIT:w>' ) <EOL> tmp . close ( ) <EOL> os . remove ( '<STR_LIT>' ) <EOL> try : <EOL> box . cause_parent_error1 ( ) <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> box . cause_parent_error2 ( ) <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> box . cause_parent_error3 ( ) <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_3_access ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> user = '<STR_LIT>' + socket . gethostname ( ) <EOL> key_pair = get_key_pair ( user ) <EOL> data = '<STR_LIT:\n>' . join ( [ user , '<STR_LIT:0>' , key_pair . publickey ( ) . exportKey ( ) ] ) <EOL> hash = hashlib . sha256 ( data ) . digest ( ) <EOL> signature = key_pair . sign ( hash , get_random_bytes ) <EOL> spook = Credentials ( ( data , signature , None ) ) <EOL> credentials = get_credentials ( ) <EOL> allowed_users = { credentials . user : credentials . public_key , <EOL> spook . user : spook . public_key } <EOL> factory = self . start_factory ( allowed_users = allowed_users ) <EOL> saved = get_credentials ( ) <EOL> set_credentials ( spook ) <EOL> box = factory . create ( _MODULE + '<STR_LIT>' , <EOL> allowed_users = allowed_users ) <EOL> set_credentials ( saved ) <EOL> model = set_as_top ( Model ( box ) ) <EOL> model . run ( ) <EOL> for width in range ( <NUM_LIT:1> , <NUM_LIT:2> ) : <EOL> for height in range ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> for depth in range ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> case = model . recorders [ <NUM_LIT:0> ] . cases . pop ( <NUM_LIT:0> ) <EOL> self . assertEqual ( case . get_output ( '<STR_LIT>' ) , <EOL> width * height * depth ) <EOL> try : <EOL> i = model . box . secret <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> model . box . proprietary_method ( ) <EOL> except RemoteError as exc : <EOL> msg = "<STR_LIT>" <EOL> logging . debug ( '<STR_LIT>' , msg ) <EOL> logging . debug ( '<STR_LIT>' , exc ) <EOL> self . assertTrue ( msg in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> saved = get_credentials ( ) <EOL> set_credentials ( spook ) <EOL> try : <EOL> i = model . box . secret <EOL> model . box . proprietary_method ( ) <EOL> finally : <EOL> set_credentials ( saved ) <EOL> def test_4_authkey ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> factory = self . start_factory ( ) <EOL> authkey = '<STR_LIT:password>' <EOL> server_dir = '<STR_LIT>' <EOL> if os . path . exists ( server_dir ) : <EOL> shutil . rmtree ( server_dir , onerror = onerror ) <EOL> os . mkdir ( server_dir ) <EOL> os . chdir ( server_dir ) <EOL> self . server_dirs . append ( server_dir ) <EOL> try : <EOL> logging . debug ( '<STR_LIT>' , authkey ) <EOL> allowed_types = [ '<STR_LIT>' ] <EOL> server , server_cfg = start_server ( authkey = authkey , <EOL> allowed_types = allowed_types , <EOL> timeout = <NUM_LIT:30> ) <EOL> cfg = read_server_config ( server_cfg ) <EOL> address = cfg [ '<STR_LIT:address>' ] <EOL> port = cfg [ '<STR_LIT:port>' ] <EOL> key = cfg [ '<STR_LIT:key>' ] <EOL> logging . debug ( '<STR_LIT>' , address ) <EOL> logging . debug ( '<STR_LIT>' , port ) <EOL> logging . debug ( '<STR_LIT>' , cfg [ '<STR_LIT>' ] ) <EOL> logging . debug ( '<STR_LIT>' , key ) <EOL> finally : <EOL> os . chdir ( '<STR_LIT:..>' ) <EOL> factory = None <EOL> try : <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , AuthenticationError , <EOL> '<STR_LIT>' ) <EOL> factory = connect ( address , port , authkey = authkey ) <EOL> logging . debug ( '<STR_LIT>' , factory ) <EOL> box = factory . create ( _MODULE + '<STR_LIT>' ) <EOL> model = set_as_top ( Model ( box ) ) <EOL> model . run ( ) <EOL> for width in range ( <NUM_LIT:1> , <NUM_LIT:2> ) : <EOL> for height in range ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> for depth in range ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> case = model . recorders [ <NUM_LIT:0> ] . cases . pop ( <NUM_LIT:0> ) <EOL> self . assertEqual ( case . get_output ( '<STR_LIT>' ) , <EOL> width * height * depth ) <EOL> finally : <EOL> if factory is not None : <EOL> factory . cleanup ( ) <EOL> logging . debug ( '<STR_LIT>' , <EOL> authkey , server . pid ) <EOL> server . terminate ( timeout = <NUM_LIT:10> ) <EOL> server = None <EOL> def test_5_misc ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> factory = self . start_factory ( ) <EOL> server = factory . create ( '<STR_LIT>' ) <EOL> factory . release ( server ) <EOL> msg1 = "<STR_LIT>" <EOL> msg2 = "<STR_LIT>" <EOL> try : <EOL> reply = server . echo ( '<STR_LIT:hello>' ) <EOL> except RuntimeError as exc : <EOL> if str ( exc ) [ : len ( msg1 ) ] != msg1 and str ( exc ) [ : len ( msg2 ) ] != msg2 : <EOL> self . fail ( '<STR_LIT>' % exc ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> server = factory . create ( '<STR_LIT>' ) <EOL> reply = server . echo ( '<STR_LIT:hello>' ) <EOL> factory . release ( server ) <EOL> msg1 = "<STR_LIT>" <EOL> msg2 = "<STR_LIT>" <EOL> try : <EOL> reply = server . echo ( '<STR_LIT:hello>' ) <EOL> except RuntimeError as exc : <EOL> if str ( exc ) [ : len ( msg1 ) ] != msg1 and str ( exc ) [ : len ( msg2 ) ] != msg2 : <EOL> self . fail ( '<STR_LIT>' % exc ) <EOL> except EOFError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> server = factory . create ( '<STR_LIT>' ) <EOL> factory . release ( server ) <EOL> msg1 = "<STR_LIT>" <EOL> msg2 = "<STR_LIT>" "<STR_LIT>" <EOL> try : <EOL> factory . release ( server ) <EOL> except ValueError as exc : <EOL> self . assertEqual ( str ( exc ) [ : len ( msg1 ) ] , msg1 ) <EOL> except RemoteError as exc : <EOL> self . assertTrue ( msg2 in str ( exc ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . assertFalse ( has_interface ( factory , HasObjectives ) ) <EOL> address = socket . gethostname ( ) <EOL> junk_port = <NUM_LIT> <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , "<STR_LIT>" ) <EOL> code = compile ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_raises ( self , '<STR_LIT>' , globals ( ) , locals ( ) , <EOL> cPickle . PicklingError , "<STR_LIT>" ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . getLogger ( ) . setLevel ( logging . DEBUG ) <EOL> import nose <EOL> nose . runmodule ( ) </s>
