<s> from __future__ import print_function <EOL> import os <EOL> import re <EOL> import socket <EOL> import warnings <EOL> import pytest <EOL> from process_tests import dump_on_error <EOL> from process_tests import wait_for_strings <EOL> import aspectlib <EOL> from aspectlib . test import mock <EOL> from aspectlib . test import record <EOL> from aspectlib . utils import PYPY <EOL> try : <EOL> import thread <EOL> except ImportError : <EOL> import _thread as thread <EOL> try : <EOL> from StringIO import StringIO <EOL> except ImportError : <EOL> from io import StringIO <EOL> LOG_TEST_SOCKET = r"""<STR_LIT>""" <EOL> def test_mock_builtin ( ) : <EOL> with aspectlib . weave ( open , mock ( '<STR_LIT>' ) ) : <EOL> assert open ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> assert open ( __file__ ) != '<STR_LIT>' <EOL> def test_mock_builtin_os ( ) : <EOL> print ( os . open . __name__ ) <EOL> with aspectlib . weave ( '<STR_LIT>' , mock ( '<STR_LIT>' ) ) : <EOL> assert os . open ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> assert os . open ( __file__ , <NUM_LIT:0> ) != '<STR_LIT>' <EOL> def test_record_warning ( ) : <EOL> with aspectlib . weave ( '<STR_LIT>' , record ) : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> assert warnings . warn . calls == [ ( None , ( '<STR_LIT>' , ) , { } ) ] <EOL> @ pytest . mark . skipif ( not hasattr ( os , '<STR_LIT>' ) , reason = "<STR_LIT>" ) <EOL> def test_fork ( ) : <EOL> with aspectlib . weave ( '<STR_LIT>' , mock ( '<STR_LIT>' ) ) : <EOL> pid = os . fork ( ) <EOL> if not pid : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> assert pid == '<STR_LIT>' <EOL> pid = os . fork ( ) <EOL> if not pid : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> assert pid != '<STR_LIT>' <EOL> def test_socket ( target = socket . socket ) : <EOL> buf = StringIO ( ) <EOL> with aspectlib . weave ( target , aspectlib . debug . log ( <EOL> print_to = buf , <EOL> stacktrace = <NUM_LIT:4> , <EOL> module = False <EOL> ) , lazy = True ) : <EOL> s = socket . socket ( ) <EOL> try : <EOL> s . connect ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:1> ) ) <EOL> except Exception : <EOL> pass <EOL> print ( buf . getvalue ( ) ) <EOL> assert re . match ( LOG_TEST_SOCKET , buf . getvalue ( ) ) <EOL> s = socket . socket ( ) <EOL> try : <EOL> s . connect ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:1> ) ) <EOL> except Exception : <EOL> pass <EOL> assert re . match ( LOG_TEST_SOCKET , buf . getvalue ( ) ) <EOL> def test_socket_as_string_target ( ) : <EOL> test_socket ( target = '<STR_LIT>' ) <EOL> def test_socket_meth ( meth = socket . socket . close ) : <EOL> calls = [ ] <EOL> with aspectlib . weave ( meth , record ( calls = calls ) ) : <EOL> s = socket . socket ( ) <EOL> assert s . close ( ) is None <EOL> assert calls == [ ( s , ( ) , { } ) ] <EOL> del calls [ : ] <EOL> s = socket . socket ( ) <EOL> assert s . close ( ) is None <EOL> assert calls == [ ] <EOL> def test_socket_meth_as_string_target ( ) : <EOL> test_socket_meth ( '<STR_LIT>' ) <EOL> def test_socket_all_methods ( ) : <EOL> buf = StringIO ( ) <EOL> with aspectlib . weave ( <EOL> socket . socket , <EOL> aspectlib . debug . log ( print_to = buf , stacktrace = False ) , <EOL> lazy = True , <EOL> methods = aspectlib . ALL_METHODS <EOL> ) : <EOL> s = socket . socket ( ) <EOL> assert "<STR_LIT>" in buf . getvalue ( ) <EOL> @ pytest . mark . skipif ( not hasattr ( os , '<STR_LIT>' ) or PYPY , reason = "<STR_LIT>" ) <EOL> def test_realsocket_makefile ( ) : <EOL> buf = StringIO ( ) <EOL> p = socket . socket ( ) <EOL> p . bind ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) ) <EOL> p . listen ( <NUM_LIT:1> ) <EOL> p . settimeout ( <NUM_LIT:1> ) <EOL> pid = os . fork ( ) <EOL> if pid : <EOL> with aspectlib . weave ( <EOL> [ '<STR_LIT>' if aspectlib . PY2 else '<STR_LIT>' ] + <EOL> ( [ '<STR_LIT>' , '<STR_LIT>' ] if aspectlib . PY2 else [ '<STR_LIT>' ] ) , <EOL> aspectlib . debug . log ( print_to = buf , stacktrace = False ) , <EOL> lazy = True , <EOL> methods = aspectlib . ALL_METHODS , <EOL> ) : <EOL> s = socket . socket ( ) <EOL> s . settimeout ( <NUM_LIT:1> ) <EOL> s . connect ( p . getsockname ( ) ) <EOL> if aspectlib . PY3 : <EOL> fh = s . makefile ( '<STR_LIT>' , buffering = <NUM_LIT:0> ) <EOL> else : <EOL> fh = s . makefile ( bufsize = <NUM_LIT:0> ) <EOL> fh . write ( b"<STR_LIT>" ) <EOL> fh . readline ( ) <EOL> with dump_on_error ( buf . getvalue ) : <EOL> wait_for_strings ( <EOL> buf . getvalue , <NUM_LIT:0> , <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> else : <EOL> try : <EOL> c , _ = p . accept ( ) <EOL> c . settimeout ( <NUM_LIT:1> ) <EOL> if aspectlib . PY3 : <EOL> f = c . makefile ( '<STR_LIT>' , buffering = <NUM_LIT:1> ) <EOL> else : <EOL> f = c . makefile ( bufsize = <NUM_LIT:1> ) <EOL> while f . readline ( ) : <EOL> f . write ( '<STR_LIT>' ) <EOL> finally : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def test_weave_os_module ( ) : <EOL> calls = [ ] <EOL> with aspectlib . weave ( '<STR_LIT>' , record ( calls = calls , extended = True ) , methods = "<STR_LIT>" ) : <EOL> os . getenv ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> os . walk ( '<STR_LIT:.>' ) <EOL> assert calls == [ <EOL> ( None , '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) , { } ) , <EOL> ( None , '<STR_LIT>' , ( '<STR_LIT:.>' , ) , { } ) <EOL> ] </s>
