<s> import pytest <EOL> if pytest . PLATFORM == "<STR_LIT:win32>" : <EOL> pytest . skip ( "<STR_LIT>" ) <EOL> from circuits . io import Process , write <EOL> def test ( manager , watcher ) : <EOL> p = Process ( [ "<STR_LIT>" , "<STR_LIT>" ] ) . register ( manager ) <EOL> assert watcher . wait ( "<STR_LIT>" ) <EOL> p . start ( ) <EOL> assert watcher . wait ( "<STR_LIT>" , p . channel ) <EOL> assert watcher . wait ( "<STR_LIT>" , p . channel ) <EOL> s = p . stdout . getvalue ( ) <EOL> assert s == b"<STR_LIT>" <EOL> def test2 ( manager , watcher , tmpdir ) : <EOL> foo = tmpdir . ensure ( "<STR_LIT>" ) <EOL> p = Process ( <EOL> [ "<STR_LIT>" . format ( str ( foo ) ) ] , shell = True ) . register ( manager ) <EOL> assert watcher . wait ( "<STR_LIT>" ) <EOL> p . start ( ) <EOL> assert watcher . wait ( "<STR_LIT>" , p . channel ) <EOL> p . fire ( write ( "<STR_LIT>" ) , p . _stdin ) <EOL> assert watcher . wait ( "<STR_LIT>" , p . _stdin ) <EOL> p . stop ( ) <EOL> assert watcher . wait ( "<STR_LIT>" , p . _stdout . channel ) <EOL> with foo . open ( "<STR_LIT:r>" ) as f : <EOL> assert f . read ( ) == "<STR_LIT>" <EOL> def test_two_procs ( manager , watcher ) : <EOL> p1 = Process ( [ "<STR_LIT>" , "<STR_LIT:1>" ] ) . register ( manager ) <EOL> p2 = Process ( "<STR_LIT>" , shell = True ) . register ( manager ) <EOL> p1 . start ( ) <EOL> p2 . start ( ) <EOL> assert watcher . wait ( "<STR_LIT>" , p1 . channel ) <EOL> assert p1 . _terminated <EOL> assert not p2 . _terminated <EOL> assert not p2 . _stdout_closed <EOL> assert not p2 . _stderr_closed <EOL> watcher . clear ( ) <EOL> s1 = p1 . stdout . getvalue ( ) <EOL> assert s1 == b"<STR_LIT>" <EOL> assert watcher . wait ( "<STR_LIT>" , p2 . channel ) <EOL> assert p2 . _terminated <EOL> assert p2 . _stdout_closed <EOL> assert p2 . _stderr_closed <EOL> s2 = p2 . stdout . getvalue ( ) <EOL> assert s2 == b"<STR_LIT>" </s>
