<s> """<STR_LIT>""" <EOL> import unittest <EOL> import tempfile <EOL> import StringIO <EOL> import os <EOL> import logging <EOL> import shutil <EOL> from openmdao . main . api import Assembly , Case , set_as_top <EOL> from openmdao . test . execcomp import ExecComp <EOL> from openmdao . lib . casehandlers . api import DBCaseIterator , DBCaseRecorder , DumpCaseRecorder , case_db_to_dict <EOL> from openmdao . lib . drivers . api import SimpleCaseIterDriver , CaseIteratorDriver <EOL> from openmdao . main . uncertain_distributions import NormalDistribution <EOL> from openmdao . main . datatypes . api import List , Dict , Str <EOL> from openmdao . util . testutil import assert_raises <EOL> from openmdao . util . fileutil import onerror <EOL> from openmdao . main . caseiter import caseiter_to_dict <EOL> class TracedExecComp ( ExecComp ) : <EOL> label = Str ( iotype = '<STR_LIT>' ) <EOL> def execute ( self ) : <EOL> super ( TracedExecComp , self ) . execute ( ) <EOL> class DBCaseRecorderTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . top = top = set_as_top ( Assembly ( ) ) <EOL> driver = top . add ( '<STR_LIT>' , SimpleCaseIterDriver ( ) ) <EOL> top . add ( '<STR_LIT>' , ExecComp ( exprs = [ '<STR_LIT>' ] ) ) <EOL> top . add ( '<STR_LIT>' , ExecComp ( exprs = [ '<STR_LIT>' ] ) ) <EOL> top . comp1 . add ( '<STR_LIT>' , Dict ( { } , iotype = '<STR_LIT>' ) ) <EOL> top . comp1 . add ( '<STR_LIT>' , List ( [ ] , iotype = '<STR_LIT>' ) ) <EOL> top . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> driver . workflow . add ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> cases = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> i = float ( i ) <EOL> inputs = [ ( '<STR_LIT>' , i ) , ( '<STR_LIT>' , i * <NUM_LIT:2> ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:a>' : '<STR_LIT:b>' } ) , <EOL> ( '<STR_LIT>' , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) ] <EOL> cases . append ( Case ( inputs = inputs , outputs = outputs ) ) <EOL> Case . set_vartree_inputs ( driver , cases ) <EOL> driver . add_responses ( outputs ) <EOL> def test_inoutDB ( self ) : <EOL> self . top . recorders = [ DBCaseRecorder ( ) ] <EOL> self . top . run ( ) <EOL> cases = [ case for case in self . top . recorders [ <NUM_LIT:0> ] . get_iterator ( ) ] <EOL> Case . set_vartree_inputs ( self . top . driver , cases ) <EOL> sout = StringIO . StringIO ( ) <EOL> self . top . recorders = [ DumpCaseRecorder ( sout ) ] <EOL> self . top . run ( ) <EOL> expected = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> lines = sout . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> count = <NUM_LIT:0> <EOL> for index , line in enumerate ( lines ) : <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> count += <NUM_LIT:1> <EOL> if count != <NUM_LIT:9> : <EOL> continue <EOL> for i in range ( len ( expected ) ) : <EOL> if expected [ i ] . startswith ( '<STR_LIT>' ) : <EOL> self . assertTrue ( lines [ index + i ] . startswith ( '<STR_LIT>' ) ) <EOL> elif expected [ i ] . startswith ( '<STR_LIT>' ) : <EOL> self . assertTrue ( lines [ index + i ] . startswith ( '<STR_LIT>' ) ) <EOL> else : <EOL> self . assertEqual ( lines [ index + i ] , expected [ i ] ) <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_pickle_conversion ( self ) : <EOL> recorder = DBCaseRecorder ( ) <EOL> inputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> recorder . register ( self , inputs , outputs ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> inputs = [ i , i * <NUM_LIT> ] <EOL> outputs = [ i * <NUM_LIT> , NormalDistribution ( float ( i ) , <NUM_LIT:0.5> ) ] <EOL> recorder . record ( self , inputs , outputs , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> iterator = recorder . get_iterator ( ) <EOL> for i , case in enumerate ( iterator ) : <EOL> self . assertTrue ( isinstance ( case [ '<STR_LIT>' ] , NormalDistribution ) ) <EOL> self . assertEqual ( case [ '<STR_LIT>' ] . mu , float ( i ) ) <EOL> self . assertEqual ( case [ '<STR_LIT>' ] . sigma , <NUM_LIT:0.5> ) <EOL> self . assertTrue ( isinstance ( case [ '<STR_LIT>' ] , float ) ) <EOL> self . assertEqual ( case [ '<STR_LIT>' ] , i * <NUM_LIT> ) <EOL> self . assertEqual ( case [ '<STR_LIT>' ] , i * <NUM_LIT> ) <EOL> def test_query ( self ) : <EOL> recorder = DBCaseRecorder ( ) <EOL> inputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> recorder . register ( self , inputs , outputs ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> inputs = [ i , i * <NUM_LIT> ] <EOL> outputs = [ i * <NUM_LIT> , NormalDistribution ( float ( i ) , <NUM_LIT:0.5> ) ] <EOL> recorder . record ( self , inputs , outputs , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> iterator = recorder . get_iterator ( ) <EOL> iterator . selectors = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> count = <NUM_LIT:0> <EOL> for i , case in enumerate ( iterator ) : <EOL> count += <NUM_LIT:1> <EOL> for value in case . values ( ) : <EOL> self . assertTrue ( value >= <NUM_LIT:0> and value < <NUM_LIT:3> ) <EOL> self . assertEqual ( count , <NUM_LIT:3> ) <EOL> def test_tables_already_exist ( self ) : <EOL> dbdir = tempfile . mkdtemp ( ) <EOL> dbname = os . path . join ( dbdir , '<STR_LIT>' ) <EOL> recorder = DBCaseRecorder ( dbname ) <EOL> recorder . close ( ) <EOL> recorder = DBCaseRecorder ( dbname , append = True ) <EOL> recorder . close ( ) <EOL> try : <EOL> recorder = DBCaseRecorder ( dbname ) <EOL> recorder . close ( ) <EOL> except Exception as err : <EOL> self . assertEqual ( '<STR_LIT>' , str ( err ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> shutil . rmtree ( dbdir , onerror = onerror ) <EOL> except OSError : <EOL> logging . error ( "<STR_LIT>" , dbdir ) <EOL> def test_db_to_dict ( self ) : <EOL> tmpdir = tempfile . mkdtemp ( ) <EOL> dfile = os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> recorder = DBCaseRecorder ( dfile ) <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> inputs = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> recorder . register ( self , inputs , outputs ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> inputs = [ i , i * <NUM_LIT:2> , i * <NUM_LIT:3> ] <EOL> outputs = [ i * i , float ( '<STR_LIT>' ) ] <EOL> recorder . record ( self , inputs , outputs , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> varnames = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> varinfo = case_db_to_dict ( dfile , varnames ) <EOL> self . assertEqual ( len ( varinfo ) , <NUM_LIT:3> ) <EOL> for lst in varinfo . values ( ) : <EOL> self . assertEqual ( len ( lst ) , <NUM_LIT:10> ) <EOL> varinfo = caseiter_to_dict ( recorder . get_iterator ( ) , varnames ) <EOL> for lst in varinfo . values ( ) : <EOL> self . assertEqual ( len ( lst ) , <NUM_LIT:10> ) <EOL> try : <EOL> shutil . rmtree ( tmpdir , onerror = onerror ) <EOL> except OSError : <EOL> logging . error ( "<STR_LIT>" , tmpdir ) <EOL> def test_string ( self ) : <EOL> recorder = DBCaseRecorder ( ) <EOL> inputs = [ '<STR_LIT:str>' , '<STR_LIT>' , '<STR_LIT:list>' ] <EOL> recorder . register ( self , inputs , [ ] ) <EOL> inputs = [ '<STR_LIT>' , u'<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> recorder . record ( self , inputs , [ ] , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for case in recorder . get_iterator ( ) : <EOL> self . assertEqual ( case [ '<STR_LIT:str>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( case [ '<STR_LIT>' ] , u'<STR_LIT>' ) <EOL> self . assertEqual ( case [ '<STR_LIT:list>' ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_close ( self ) : <EOL> recorder = DBCaseRecorder ( ) <EOL> inps = [ '<STR_LIT:str>' , '<STR_LIT>' , '<STR_LIT:list>' ] <EOL> recorder . register ( self , inps , [ ] ) <EOL> inputs = [ '<STR_LIT>' , u'<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> recorder . record ( self , inputs , [ ] , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> recorder . close ( ) <EOL> recorder . record ( self , inputs , [ ] , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> tmpdir = tempfile . mkdtemp ( ) <EOL> try : <EOL> dfile = os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> recorder = DBCaseRecorder ( dfile ) <EOL> recorder . register ( self , inps , [ ] ) <EOL> recorder . record ( self , inputs , [ ] , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> recorder . close ( ) <EOL> code = "<STR_LIT>" <EOL> assert_raises ( self , code , globals ( ) , locals ( ) , RuntimeError , <EOL> '<STR_LIT>' ) <EOL> finally : <EOL> try : <EOL> shutil . rmtree ( tmpdir , onerror = onerror ) <EOL> except OSError : <EOL> logging . error ( "<STR_LIT>" , tmpdir ) <EOL> class NestedCaseTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . tdir = tempfile . mkdtemp ( ) <EOL> self . num_cases = <NUM_LIT:5> <EOL> def tearDown ( self ) : <EOL> try : <EOL> shutil . rmtree ( self . tdir , onerror = onerror ) <EOL> except OSError : <EOL> logging . error ( "<STR_LIT>" , self . tdir ) <EOL> def _create_assembly ( self , dbname , drivertype ) : <EOL> asm = Assembly ( ) <EOL> driver = asm . add ( '<STR_LIT>' , drivertype ( ) ) <EOL> asm . add ( '<STR_LIT>' , TracedExecComp ( exprs = [ '<STR_LIT>' ] ) ) <EOL> asm . add ( '<STR_LIT>' , TracedExecComp ( exprs = [ '<STR_LIT>' ] ) ) <EOL> asm . connect ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> driver . workflow . add ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> asm . recorders = [ DBCaseRecorder ( dbname , append = True ) ] <EOL> return asm <EOL> def _create_nested_assemblies ( self , dbname , drivertype ) : <EOL> top = set_as_top ( self . _create_assembly ( dbname , drivertype ) ) <EOL> top . add ( '<STR_LIT>' , self . _create_assembly ( dbname , drivertype ) ) <EOL> top . driver . workflow . add ( '<STR_LIT>' ) <EOL> top . asm . add ( '<STR_LIT>' , self . _create_assembly ( dbname , drivertype ) ) <EOL> top . asm . driver . workflow . add ( '<STR_LIT>' ) <EOL> top . recorders = [ DBCaseRecorder ( dbname , append = True ) ] <EOL> Case . set_vartree_inputs ( top . driver , self . _create_cases ( <NUM_LIT:1> ) ) <EOL> Case . set_vartree_inputs ( top . asm . driver , self . _create_cases ( <NUM_LIT:2> ) ) <EOL> Case . set_vartree_inputs ( top . asm . asm . driver , self . _create_cases ( <NUM_LIT:3> ) ) <EOL> return top <EOL> def _create_nested_workflows ( self , dbname , drivertype ) : <EOL> top = set_as_top ( self . _create_assembly ( dbname , drivertype ) ) <EOL> driver2 = top . add ( '<STR_LIT>' , drivertype ( ) ) <EOL> top . driver . workflow . add ( [ '<STR_LIT>' ] ) <EOL> driver3 = top . add ( '<STR_LIT>' , drivertype ( ) ) <EOL> top . driver2 . workflow . add ( [ '<STR_LIT>' ] ) <EOL> top . driver3 . workflow . add ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> Case . set_vartree_inputs ( top . driver , self . _create_cases ( <NUM_LIT:1> ) ) <EOL> Case . set_vartree_inputs ( top . driver2 , self . _create_cases ( <NUM_LIT:2> ) ) <EOL> Case . set_vartree_inputs ( top . driver3 , self . _create_cases ( <NUM_LIT:3> ) ) <EOL> return top <EOL> def _create_cases ( self , level ) : <EOL> outputs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> cases = [ ] <EOL> for i in range ( self . num_cases ) : <EOL> i = float ( i ) <EOL> inputs = [ ( '<STR_LIT>' , <NUM_LIT:100> * level + i ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:100> * level + i + <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' % ( level , i ) ) ] <EOL> cases . append ( Case ( inputs = inputs , outputs = outputs ) ) <EOL> return cases <EOL> def _get_level_cases ( self , caseiter ) : <EOL> levels = [ [ ] , [ ] , [ ] ] <EOL> for case in caseiter : <EOL> if '<STR_LIT>' in case and case [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) : <EOL> levels [ <NUM_LIT:0> ] . append ( case ) <EOL> elif '<STR_LIT>' in case and case [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) : <EOL> levels [ <NUM_LIT:1> ] . append ( case ) <EOL> elif '<STR_LIT>' in case and case [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) : <EOL> levels [ <NUM_LIT:2> ] . append ( case ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> return levels <EOL> def _check_cases ( self , caseiter ) : <EOL> levels = self . _get_level_cases ( caseiter ) <EOL> for i , level in enumerate ( levels ) : <EOL> if i > <NUM_LIT:0> : <EOL> parents = [ c . uuid for c in levels [ i - <NUM_LIT:1> ] ] <EOL> for j , case in enumerate ( level ) : <EOL> if j == <NUM_LIT:0> : <EOL> parent_uuid = case . parent_uuid <EOL> if i > <NUM_LIT:0> : <EOL> self . assertTrue ( parent_uuid ) <EOL> else : <EOL> if i > <NUM_LIT:0> : <EOL> self . assertTrue ( case . parent_uuid in parents ) <EOL> return levels <EOL> def test_nested_assemblies_simple ( self ) : <EOL> dbname = os . path . join ( self . tdir , '<STR_LIT>' ) <EOL> self . top = self . _create_nested_assemblies ( dbname , SimpleCaseIterDriver ) <EOL> self . top . run ( ) <EOL> levels = self . _check_cases ( DBCaseIterator ( dbname ) ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:0> ] ) , self . top . comp1 . exec_count ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:1> ] ) , self . top . asm . comp1 . exec_count ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:2> ] ) , self . top . asm . asm . comp1 . exec_count ) <EOL> def test_nested_assemblies_caseiter ( self ) : <EOL> dbname = os . path . join ( self . tdir , '<STR_LIT>' ) <EOL> self . top = self . _create_nested_assemblies ( dbname , CaseIteratorDriver ) <EOL> self . top . run ( ) <EOL> levels = self . _check_cases ( DBCaseIterator ( dbname ) ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:0> ] ) , self . top . comp1 . exec_count ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:1> ] ) , self . top . asm . comp1 . exec_count ) <EOL> self . assertEqual ( len ( levels [ <NUM_LIT:2> ] ) , self . top . asm . asm . comp1 . exec_count ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
