<s> """<STR_LIT>""" <EOL> import logging <EOL> import os . path <EOL> import pkg_resources <EOL> import shutil <EOL> import sys <EOL> import time <EOL> import tempfile <EOL> import unittest <EOL> import nose <EOL> from multiprocessing . managers import RemoteError <EOL> from openmdao . main . api import Assembly , FileMetadata , SimulationRoot , set_as_top <EOL> from openmdao . main . eggchecker import check_save_load <EOL> from openmdao . main . exceptions import RunInterrupted <EOL> from openmdao . main . objserverfactory import ObjServerFactory <EOL> from openmdao . main . rbac import Credentials , get_credentials <EOL> from openmdao . lib . components . external_code import ExternalCode <EOL> from openmdao . main . datatypes . api import Int , File , FileRef , Str <EOL> from openmdao . test . cluster import init_cluster <EOL> from openmdao . util . testutil import assert_raises <EOL> from openmdao . util . fileutil import onerror <EOL> ORIG_DIR = os . getcwd ( ) <EOL> DIRECTORY = pkg_resources . resource_filename ( '<STR_LIT>' , '<STR_LIT:test>' ) <EOL> TMPDIR = os . getcwd ( ) <EOL> ENV_FILE = '<STR_LIT>' <EOL> INP_FILE = '<STR_LIT>' <EOL> INP_DATA = '<STR_LIT>' <EOL> class Sleeper ( ExternalCode ) : <EOL> """<STR_LIT>""" <EOL> delay = Int ( <NUM_LIT:1> , units = '<STR_LIT:s>' , iotype = '<STR_LIT>' ) <EOL> env_filename = Str ( iotype = '<STR_LIT>' ) <EOL> infile = File ( iotype = '<STR_LIT>' , local_path = '<STR_LIT:input>' ) <EOL> outfile = File ( FileRef ( '<STR_LIT>' ) , iotype = '<STR_LIT>' ) <EOL> def __init__ ( self ) : <EOL> super ( Sleeper , self ) . __init__ ( ) <EOL> self . directory = TMPDIR <EOL> self . external_files = [ <EOL> FileMetadata ( path = '<STR_LIT>' , <EOL> input = True , constant = True ) , <EOL> ] <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> self . command = [ '<STR_LIT>' , '<STR_LIT>' , str ( self . delay ) ] <EOL> if self . env_filename : <EOL> self . command . append ( self . env_filename ) <EOL> super ( Sleeper , self ) . execute ( ) <EOL> class Unique ( Sleeper ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( Unique , self ) . __init__ ( ) <EOL> self . create_instance_dir = True <EOL> class Model ( Assembly ) : <EOL> """<STR_LIT>""" <EOL> infile = File ( iotype = '<STR_LIT>' , local_path = '<STR_LIT:input>' ) <EOL> outfile = File ( FileRef ( '<STR_LIT>' ) , iotype = '<STR_LIT>' ) <EOL> def configure ( self ) : <EOL> self . add ( '<STR_LIT:a>' , Unique ( ) ) <EOL> self . add ( '<STR_LIT:b>' , Unique ( ) ) <EOL> self . driver . workflow . add ( [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class TestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> global TMPDIR <EOL> self . startdir = os . getcwd ( ) <EOL> self . tempdir = tempfile . mkdtemp ( prefix = '<STR_LIT>' ) <EOL> TMPDIR = self . tempdir <EOL> os . chdir ( self . tempdir ) <EOL> SimulationRoot . chroot ( self . tempdir ) <EOL> shutil . copy ( os . path . join ( DIRECTORY , '<STR_LIT>' ) , <EOL> os . path . join ( self . tempdir , '<STR_LIT>' ) ) <EOL> with open ( INP_FILE , '<STR_LIT:w>' ) as out : <EOL> out . write ( INP_DATA ) <EOL> dum = Assembly ( ) <EOL> def tearDown ( self ) : <EOL> SimulationRoot . chroot ( self . startdir ) <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_normal ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> sleeper = set_as_top ( Sleeper ( ) ) <EOL> sleeper . env_filename = ENV_FILE <EOL> sleeper . env_vars = { '<STR_LIT>' : '<STR_LIT>' } <EOL> sleeper . external_files . append ( <EOL> FileMetadata ( path = ENV_FILE , output = True ) ) <EOL> sleeper . infile = FileRef ( INP_FILE , sleeper , input = True ) <EOL> sleeper . stderr = None <EOL> sleeper . run ( ) <EOL> self . assertEqual ( sleeper . return_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( sleeper . timed_out , False ) <EOL> self . assertEqual ( os . path . exists ( ENV_FILE ) , True ) <EOL> with open ( ENV_FILE , '<STR_LIT>' ) as inp : <EOL> data = inp . readline ( ) . rstrip ( ) <EOL> self . assertEqual ( data , sleeper . env_vars [ '<STR_LIT>' ] ) <EOL> with sleeper . outfile . open ( ) as inp : <EOL> result = inp . read ( ) <EOL> self . assertEqual ( result , INP_DATA ) <EOL> sleeper . stderr = '<STR_LIT>' <EOL> sleeper . delay = - <NUM_LIT:1> <EOL> assert_raises ( self , '<STR_LIT>' , globals ( ) , locals ( ) , RuntimeError , <EOL> '<STR_LIT>' ) <EOL> sleeper . delay = <NUM_LIT:1> <EOL> sleeper . env_vars = { '<STR_LIT>' : '<STR_LIT:1>' } <EOL> sleeper . stdin = '<STR_LIT>' <EOL> sleeper . stdout = '<STR_LIT>' <EOL> sleeper . stderr = '<STR_LIT>' <EOL> with open ( '<STR_LIT>' , '<STR_LIT:w>' ) as out : <EOL> out . write ( '<STR_LIT>' ) <EOL> sleeper . run ( ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:r>' ) as inp : <EOL> self . assertEqual ( inp . read ( ) , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:r>' ) as inp : <EOL> self . assertEqual ( inp . read ( ) , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> os . remove ( '<STR_LIT:input>' ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . remove ( '<STR_LIT>' ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . remove ( '<STR_LIT>' ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . remove ( '<STR_LIT>' ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . remove ( '<STR_LIT>' ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> sleeper . external_files . append ( <EOL> FileMetadata ( path = '<STR_LIT>' , input = True ) ) <EOL> assert_raises ( self , '<STR_LIT>' , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> def test_remote ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> init_cluster ( allow_shell = True ) <EOL> sleeper = set_as_top ( Sleeper ( ) ) <EOL> sleeper . env_filename = ENV_FILE <EOL> sleeper . env_vars = { '<STR_LIT>' : '<STR_LIT>' } <EOL> sleeper . external_files . append ( <EOL> FileMetadata ( path = ENV_FILE , output = True ) ) <EOL> sleeper . infile = FileRef ( INP_FILE , sleeper , input = True ) <EOL> sleeper . timeout = <NUM_LIT:5> <EOL> sleeper . resources = { '<STR_LIT>' : <NUM_LIT:1> } <EOL> sleeper . run ( ) <EOL> self . assertEqual ( sleeper . return_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( sleeper . timed_out , False ) <EOL> self . assertEqual ( os . path . exists ( ENV_FILE ) , True ) <EOL> with open ( ENV_FILE , '<STR_LIT:r>' ) as inp : <EOL> data = inp . readline ( ) . rstrip ( ) <EOL> self . assertEqual ( data , sleeper . env_vars [ '<STR_LIT>' ] ) <EOL> with sleeper . outfile . open ( ) as inp : <EOL> result = inp . read ( ) <EOL> self . assertEqual ( result , INP_DATA ) <EOL> sleeper . stdin = '<STR_LIT>' <EOL> assert_raises ( self , '<STR_LIT>' , globals ( ) , locals ( ) , ValueError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:w>' ) as out : <EOL> out . write ( '<STR_LIT>' ) <EOL> sleeper . stdin = '<STR_LIT>' <EOL> sleeper . stdout = '<STR_LIT>' <EOL> sleeper . stderr = ExternalCode . STDOUT <EOL> sleeper . run ( ) <EOL> sleeper . stderr = None <EOL> sleeper . run ( ) <EOL> def test_bad_alloc ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> extcode = set_as_top ( ExternalCode ( ) ) <EOL> extcode . command = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> extcode . resources = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:localhost>' : False } <EOL> assert_raises ( self , '<STR_LIT>' , globals ( ) , locals ( ) , <EOL> RuntimeError , '<STR_LIT>' ) <EOL> def test_copy ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> extcode = set_as_top ( ExternalCode ( ) ) <EOL> assert_raises ( self , "<STR_LIT>" , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> try : <EOL> shutil . copy ( '<STR_LIT>' , os . path . join ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> extcode . copy_inputs ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( os . path . exists ( '<STR_LIT>' ) , True ) <EOL> finally : <EOL> shutil . rmtree ( '<STR_LIT>' , onerror = onerror ) <EOL> if os . path . exists ( '<STR_LIT>' ) : <EOL> os . remove ( '<STR_LIT>' ) <EOL> assert_raises ( self , "<STR_LIT>" , <EOL> globals ( ) , locals ( ) , RuntimeError , <EOL> "<STR_LIT>" ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> try : <EOL> shutil . copy ( '<STR_LIT>' , os . path . join ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> extcode . copy_results ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( os . path . exists ( '<STR_LIT>' ) , True ) <EOL> finally : <EOL> shutil . rmtree ( '<STR_LIT>' , onerror = onerror ) <EOL> if os . path . exists ( '<STR_LIT>' ) : <EOL> os . remove ( '<STR_LIT>' ) <EOL> def test_save_load ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> sleeper = set_as_top ( Sleeper ( ) ) <EOL> sleeper . name = '<STR_LIT>' <EOL> sleeper . infile = FileRef ( INP_FILE , sleeper , input = True ) <EOL> retcode = check_save_load ( sleeper ) <EOL> self . assertEqual ( retcode , <NUM_LIT:0> ) <EOL> def test_timeout ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> sleeper = set_as_top ( Sleeper ( ) ) <EOL> sleeper . delay = <NUM_LIT:5> <EOL> sleeper . timeout = <NUM_LIT:1> <EOL> try : <EOL> sleeper . run ( ) <EOL> except RunInterrupted as exc : <EOL> self . assertEqual ( str ( exc ) , '<STR_LIT>' ) <EOL> self . assertEqual ( sleeper . timed_out , True ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_badcmd ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> extcode = set_as_top ( ExternalCode ( ) ) <EOL> extcode . command = [ '<STR_LIT>' ] <EOL> try : <EOL> extcode . run ( ) <EOL> except ValueError as exc : <EOL> msg = "<STR_LIT>" <EOL> self . assertEqual ( str ( exc ) , msg ) <EOL> self . assertEqual ( extcode . return_code , - <NUM_LIT> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_nullcmd ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> extcode = set_as_top ( ExternalCode ( ) ) <EOL> extcode . stdout = '<STR_LIT>' <EOL> extcode . stderr = ExternalCode . STDOUT <EOL> try : <EOL> extcode . run ( ) <EOL> except ValueError as exc : <EOL> self . assertEqual ( str ( exc ) , '<STR_LIT>' ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> finally : <EOL> if os . path . exists ( extcode . stdout ) : <EOL> os . remove ( extcode . stdout ) <EOL> def test_unique ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> model = set_as_top ( Model ( ) ) <EOL> for comp in ( model . a , model . b ) : <EOL> self . assertEqual ( comp . create_instance_dir , False ) <EOL> self . assertEqual ( comp . return_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( comp . timed_out , False ) <EOL> self . assertEqual ( model . a . directory , '<STR_LIT:a>' ) <EOL> self . assertEqual ( model . b . directory , '<STR_LIT:b>' ) <EOL> model . infile = FileRef ( INP_FILE , model , input = True ) <EOL> model . run ( ) <EOL> for comp in ( model . a , model . b ) : <EOL> self . assertEqual ( comp . return_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( comp . timed_out , False ) <EOL> with model . outfile . open ( ) as inp : <EOL> result = inp . read ( ) <EOL> self . assertEqual ( result , INP_DATA ) <EOL> def test_rsh ( self ) : <EOL> logging . debug ( '<STR_LIT>' ) <EOL> logging . debug ( '<STR_LIT>' ) <EOL> testdir = '<STR_LIT>' <EOL> if os . path . exists ( testdir ) : <EOL> shutil . rmtree ( testdir , onerror = onerror ) <EOL> os . mkdir ( testdir ) <EOL> os . chdir ( testdir ) <EOL> factory = None <EOL> try : <EOL> typname = '<STR_LIT>' <EOL> factory = ObjServerFactory ( allowed_types = [ typname ] ) <EOL> exec_comp = factory . create ( typname ) <EOL> try : <EOL> exec_comp . command = [ '<STR_LIT>' ] <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> exec_comp . set ( '<STR_LIT>' , [ '<STR_LIT>' ] ) <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> finally : <EOL> if factory is not None : <EOL> factory . cleanup ( ) <EOL> os . chdir ( '<STR_LIT:..>' ) <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> keep_dirs = int ( os . environ . get ( '<STR_LIT>' , '<STR_LIT:0>' ) ) <EOL> if not keep_dirs : <EOL> shutil . rmtree ( testdir , onerror = onerror ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> sys . argv . append ( '<STR_LIT>' ) <EOL> sys . argv . append ( '<STR_LIT>' ) <EOL> nose . runmodule ( ) </s>
