<s> import time <EOL> import sys <EOL> import os <EOL> import tempfile <EOL> import tornado <EOL> from datetime import datetime <EOL> from circus . py3compat import StringIO <EOL> from circus . client import make_message <EOL> from circus . tests . support import TestCircus , async_poll_for , truncate_file <EOL> from circus . tests . support import TestCase , EasyTestSuite , skipIf , IS_WINDOWS <EOL> from circus . stream import FileStream , WatchedFileStream <EOL> from circus . stream import TimedRotatingFileStream <EOL> from circus . stream import FancyStdoutStream <EOL> def run_process ( testfile , * args , ** kw ) : <EOL> try : <EOL> sys . stdout . write ( '<STR_LIT>' ) <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . write ( '<STR_LIT>' ) <EOL> sys . stderr . flush ( ) <EOL> with open ( testfile , '<STR_LIT>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:1.> ) <EOL> except : <EOL> return <NUM_LIT:1> <EOL> class TestWatcher ( TestCircus ) : <EOL> dummy_process = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> super ( TestWatcher , self ) . setUp ( ) <EOL> fd , self . stdout = tempfile . mkstemp ( ) <EOL> os . close ( fd ) <EOL> fd , self . stderr = tempfile . mkstemp ( ) <EOL> os . close ( fd ) <EOL> self . stdout_stream = FileStream ( self . stdout ) <EOL> self . stderr_stream = FileStream ( self . stderr ) <EOL> self . stdout_arg = { '<STR_LIT>' : self . stdout_stream } <EOL> self . stderr_arg = { '<STR_LIT>' : self . stderr_stream } <EOL> def tearDown ( self ) : <EOL> self . stdout_stream . close ( ) <EOL> self . stderr_stream . close ( ) <EOL> if os . path . exists ( self . stdout ) : <EOL> os . remove ( self . stdout ) <EOL> if os . path . exists ( self . stderr ) : <EOL> os . remove ( self . stderr ) <EOL> @ tornado . gen . coroutine <EOL> def _start_arbiter ( self ) : <EOL> yield self . start_arbiter ( cmd = self . dummy_process , <EOL> stdout_stream = self . stdout_arg , <EOL> stderr_stream = self . stderr_arg ) <EOL> @ tornado . gen . coroutine <EOL> def restart_arbiter ( self ) : <EOL> yield self . arbiter . restart ( ) <EOL> @ tornado . gen . coroutine <EOL> def call ( self , _cmd , ** props ) : <EOL> msg = make_message ( _cmd , ** props ) <EOL> resp = yield self . cli . call ( msg ) <EOL> raise tornado . gen . Return ( resp ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> @ tornado . testing . gen_test <EOL> def test_file_stream ( self ) : <EOL> yield self . _start_arbiter ( ) <EOL> stream = FileStream ( self . stdout , max_bytes = '<STR_LIT>' , backup_count = '<STR_LIT:3>' ) <EOL> self . assertTrue ( isinstance ( stream . _max_bytes , int ) ) <EOL> self . assertTrue ( isinstance ( stream . _backup_count , int ) ) <EOL> yield self . stop_arbiter ( ) <EOL> stream . close ( ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> @ tornado . testing . gen_test <EOL> def test_watched_file_stream ( self ) : <EOL> yield self . _start_arbiter ( ) <EOL> stream = WatchedFileStream ( self . stdout , <EOL> time_format = '<STR_LIT>' ) <EOL> self . assertTrue ( isinstance ( stream . _time_format , str ) ) <EOL> yield self . stop_arbiter ( ) <EOL> stream . close ( ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> @ tornado . testing . gen_test <EOL> def test_timed_rotating_file_stream ( self ) : <EOL> yield self . _start_arbiter ( ) <EOL> stream = TimedRotatingFileStream ( self . stdout , <EOL> rotate_when = '<STR_LIT:H>' , <EOL> rotate_interval = '<STR_LIT:5>' , <EOL> backup_count = '<STR_LIT:3>' , <EOL> utc = '<STR_LIT:True>' ) <EOL> self . assertTrue ( isinstance ( stream . _interval , int ) ) <EOL> self . assertTrue ( isinstance ( stream . _backup_count , int ) ) <EOL> self . assertTrue ( isinstance ( stream . _utc , bool ) ) <EOL> self . assertTrue ( stream . _suffix is not None ) <EOL> self . assertTrue ( stream . _ext_match is not None ) <EOL> self . assertTrue ( stream . _rollover_at > <NUM_LIT:0> ) <EOL> yield self . stop_arbiter ( ) <EOL> stream . close ( ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> @ tornado . testing . gen_test <EOL> def test_stream ( self ) : <EOL> yield self . _start_arbiter ( ) <EOL> res1 = yield async_poll_for ( self . stdout , '<STR_LIT>' ) <EOL> res2 = yield async_poll_for ( self . stderr , '<STR_LIT>' ) <EOL> self . assertTrue ( res1 ) <EOL> self . assertTrue ( res2 ) <EOL> truncate_file ( self . stdout ) <EOL> truncate_file ( self . stderr ) <EOL> yield self . restart_arbiter ( ) <EOL> res1 = yield async_poll_for ( self . stdout , '<STR_LIT>' ) <EOL> res2 = yield async_poll_for ( self . stderr , '<STR_LIT>' ) <EOL> self . assertTrue ( res1 ) <EOL> self . assertTrue ( res2 ) <EOL> yield self . stop_arbiter ( ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> @ tornado . testing . gen_test <EOL> def test_stop_and_restart ( self ) : <EOL> yield self . _start_arbiter ( ) <EOL> res1 = yield async_poll_for ( self . stdout , '<STR_LIT>' ) <EOL> res2 = yield async_poll_for ( self . stderr , '<STR_LIT>' ) <EOL> self . assertTrue ( res1 ) <EOL> self . assertTrue ( res2 ) <EOL> self . assertFalse ( self . stdout_stream . _file . closed ) <EOL> self . assertFalse ( self . stderr_stream . _file . closed ) <EOL> truncate_file ( self . stdout ) <EOL> truncate_file ( self . stderr ) <EOL> yield self . arbiter . watchers [ <NUM_LIT:0> ] . stop ( ) <EOL> self . assertTrue ( self . stdout_stream . _file . closed ) <EOL> self . assertTrue ( self . stderr_stream . _file . closed ) <EOL> yield self . arbiter . watchers [ <NUM_LIT:0> ] . start ( ) <EOL> res1 = yield async_poll_for ( self . stdout , '<STR_LIT>' ) <EOL> res2 = yield async_poll_for ( self . stderr , '<STR_LIT>' ) <EOL> self . assertTrue ( res1 ) <EOL> self . assertTrue ( res2 ) <EOL> self . assertFalse ( self . stdout_stream . _file . closed ) <EOL> self . assertFalse ( self . stderr_stream . _file . closed ) <EOL> yield self . stop_arbiter ( ) <EOL> class TestFancyStdoutStream ( TestCase ) : <EOL> def color_start ( self , code ) : <EOL> return '<STR_LIT>' % code <EOL> def color_end ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_stream ( self , * args , ** kw ) : <EOL> now = datetime . now ( ) <EOL> stream = FancyStdoutStream ( * args , ** kw ) <EOL> stream . out = StringIO ( ) <EOL> stream . now = lambda : now <EOL> return stream <EOL> def get_output ( self , stream ) : <EOL> data = { '<STR_LIT:data>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . out . getvalue ( ) <EOL> stream . out . close ( ) <EOL> expected = self . color_start ( stream . color_code ) <EOL> expected += stream . now ( ) . strftime ( stream . time_format ) + "<STR_LIT:U+0020>" <EOL> expected += "<STR_LIT>" + data [ '<STR_LIT:data>' ] + self . color_end ( ) <EOL> return output , expected <EOL> def test_random_colored_output ( self ) : <EOL> stream = self . get_stream ( ) <EOL> output , expected = self . get_output ( stream ) <EOL> self . assertEqual ( output , expected ) <EOL> def test_red_colored_output ( self ) : <EOL> stream = self . get_stream ( color = '<STR_LIT>' ) <EOL> output , expected = self . get_output ( stream ) <EOL> self . assertEqual ( output , expected ) <EOL> def test_time_formatting ( self ) : <EOL> stream = self . get_stream ( time_format = '<STR_LIT>' ) <EOL> output , expected = self . get_output ( stream ) <EOL> self . assertEqual ( output , expected ) <EOL> def test_data_split_into_lines ( self ) : <EOL> stream = self . get_stream ( color = '<STR_LIT>' ) <EOL> data = { '<STR_LIT:data>' : '<STR_LIT:\n>' . join ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . out . getvalue ( ) <EOL> stream . out . close ( ) <EOL> self . assertEqual ( len ( output . split ( '<STR_LIT:\n>' ) ) , <NUM_LIT:4> ) <EOL> def test_data_with_extra_lines ( self ) : <EOL> stream = self . get_stream ( color = '<STR_LIT>' ) <EOL> data = { '<STR_LIT:data>' : '<STR_LIT:\n>' . join ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . out . getvalue ( ) <EOL> stream . out . close ( ) <EOL> self . assertEqual ( len ( output . split ( '<STR_LIT:\n>' ) ) , <NUM_LIT:4> ) <EOL> def test_color_selections ( self ) : <EOL> for i , color in enumerate ( FancyStdoutStream . colors ) : <EOL> stream = self . get_stream ( color ) <EOL> self . assertEqual ( i + <NUM_LIT:1> , stream . color_code ) <EOL> stream . out . close ( ) <EOL> class TestFileStream ( TestCase ) : <EOL> stream_class = FileStream <EOL> def get_stream ( self , * args , ** kw ) : <EOL> now = datetime . now ( ) <EOL> stream = self . stream_class ( * args , ** kw ) <EOL> stream . _file . close ( ) <EOL> stream . _file = StringIO ( ) <EOL> stream . _open = lambda : stream . _file <EOL> stream . now = lambda : now <EOL> return stream <EOL> def get_output ( self , stream ) : <EOL> data = { '<STR_LIT:data>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . _file . getvalue ( ) <EOL> stream . _file . close ( ) <EOL> expected = stream . now ( ) . strftime ( stream . _time_format ) + "<STR_LIT:U+0020>" <EOL> expected += "<STR_LIT>" + data [ '<STR_LIT:data>' ] + '<STR_LIT:\n>' <EOL> return output , expected <EOL> @ skipIf ( IS_WINDOWS and sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> , <EOL> "<STR_LIT>" ) <EOL> def test_time_formatting ( self ) : <EOL> stream = self . get_stream ( time_format = '<STR_LIT>' ) <EOL> output , expected = self . get_output ( stream ) <EOL> self . assertEqual ( output , expected ) <EOL> @ skipIf ( IS_WINDOWS and sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> , <EOL> "<STR_LIT>" ) <EOL> def test_data_split_into_lines ( self ) : <EOL> stream = self . get_stream ( time_format = '<STR_LIT>' ) <EOL> data = { '<STR_LIT:data>' : '<STR_LIT:\n>' . join ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . _file . getvalue ( ) <EOL> stream . _file . close ( ) <EOL> self . assertEqual ( len ( output . split ( '<STR_LIT:\n>' ) ) , <NUM_LIT:4> ) <EOL> @ skipIf ( IS_WINDOWS and sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> , <EOL> "<STR_LIT>" ) <EOL> def test_data_with_extra_lines ( self ) : <EOL> stream = self . get_stream ( time_format = '<STR_LIT>' ) <EOL> data = { '<STR_LIT:data>' : '<STR_LIT:\n>' . join ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> output = stream . _file . getvalue ( ) <EOL> stream . _file . close ( ) <EOL> self . assertEqual ( len ( output . split ( '<STR_LIT:\n>' ) ) , <NUM_LIT:4> ) <EOL> @ skipIf ( IS_WINDOWS and sys . version_info [ <NUM_LIT:0> ] < <NUM_LIT:3> , <EOL> "<STR_LIT>" ) <EOL> def test_data_with_no_EOL ( self ) : <EOL> stream = self . get_stream ( ) <EOL> data = { '<STR_LIT:data>' : '<STR_LIT:*>' * <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } <EOL> stream ( data ) <EOL> stream ( data ) <EOL> output = stream . _file . getvalue ( ) <EOL> stream . _file . close ( ) <EOL> self . assertEqual ( output , '<STR_LIT:*>' * <NUM_LIT> ) <EOL> class TestWatchedFileStream ( TestFileStream ) : <EOL> stream_class = WatchedFileStream <EOL> def get_real_stream ( self , * args , ** kw ) : <EOL> now = datetime . now ( ) <EOL> stream = self . stream_class ( * args , ** kw ) <EOL> stream . now = lambda : now <EOL> return stream <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> def test_move_file ( self ) : <EOL> _test_fd , test_filename = tempfile . mkstemp ( ) <EOL> stream = self . get_real_stream ( filename = test_filename ) <EOL> line1_contents = '<STR_LIT>' <EOL> line2_contents = '<STR_LIT>' <EOL> file1 = test_filename + '<STR_LIT>' <EOL> stream ( { '<STR_LIT:data>' : line1_contents } ) <EOL> os . rename ( test_filename , file1 ) <EOL> stream ( { '<STR_LIT:data>' : line2_contents } ) <EOL> stream . close ( ) <EOL> with open ( test_filename ) as line2 : <EOL> self . assertEqual ( line2 . read ( ) . strip ( ) , line2_contents ) <EOL> with open ( file1 ) as line1 : <EOL> self . assertEqual ( line1 . read ( ) . strip ( ) , line1_contents ) <EOL> os . unlink ( test_filename ) <EOL> os . unlink ( file1 ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
