<s> from __future__ import with_statement <EOL> from contextlib import contextmanager <EOL> from copy import deepcopy <EOL> from fudge . patcher import with_patched_object <EOL> from functools import partial <EOL> from types import StringTypes <EOL> import copy <EOL> import getpass <EOL> import os <EOL> import re <EOL> import shutil <EOL> import sys <EOL> import tempfile <EOL> from fudge import Fake , patched_context , clear_expectations , with_patched_object <EOL> from nose . tools import raises <EOL> from nose import SkipTest <EOL> from fabric . context_managers import settings <EOL> from fabric . state import env , output <EOL> from fabric . sftp import SFTP <EOL> import fabric . network <EOL> from fabric . network import normalize , to_dict <EOL> from server import PORT , PASSWORDS , USER , HOST <EOL> from mock_streams import mock_streams <EOL> class FabricTest ( object ) : <EOL> """<STR_LIT>""" <EOL> def setup ( self ) : <EOL> clear_expectations ( ) <EOL> self . previous_env = copy . deepcopy ( env ) <EOL> self . previous_output = output . items ( ) <EOL> self . env_setup ( ) <EOL> self . tmpdir = tempfile . mkdtemp ( ) <EOL> def set_network ( self ) : <EOL> env . update ( to_dict ( '<STR_LIT>' % ( USER , HOST , PORT ) ) ) <EOL> def env_setup ( self ) : <EOL> env . disable_known_hosts = True <EOL> self . set_network ( ) <EOL> env . password = PASSWORDS [ USER ] <EOL> env . use_shell = False <EOL> def teardown ( self ) : <EOL> env . clear ( ) <EOL> env . update ( self . previous_env ) <EOL> output . update ( self . previous_output ) <EOL> shutil . rmtree ( self . tmpdir ) <EOL> clear_expectations ( ) <EOL> def path ( self , * path_parts ) : <EOL> return os . path . join ( self . tmpdir , * path_parts ) <EOL> def mkfile ( self , path , contents ) : <EOL> dest = self . path ( path ) <EOL> with open ( dest , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( contents ) <EOL> return dest <EOL> def exists_remotely ( self , path ) : <EOL> return SFTP ( env . host_string ) . exists ( path ) <EOL> def exists_locally ( self , path ) : <EOL> return os . path . exists ( path ) <EOL> def password_response ( password , times_called = None , silent = True ) : <EOL> """<STR_LIT>""" <EOL> fake = Fake ( '<STR_LIT>' , callable = True ) <EOL> if isinstance ( password , StringTypes ) : <EOL> passwords = [ password ] <EOL> else : <EOL> passwords = list ( password ) <EOL> echo = lambda x , y : y . write ( x + ( "<STR_LIT:\n>" if x != "<STR_LIT:U+0020>" else "<STR_LIT>" ) ) <EOL> fake = fake . returns ( passwords . pop ( <NUM_LIT:0> ) ) <EOL> if not silent : <EOL> fake = fake . calls ( echo ) <EOL> for pw in passwords : <EOL> fake = fake . next_call ( ) . returns ( pw ) <EOL> if not silent : <EOL> fake = fake . calls ( echo ) <EOL> if times_called : <EOL> fake = fake . times_called ( times_called ) <EOL> return patched_context ( getpass , '<STR_LIT>' , fake ) <EOL> def _assert_contains ( needle , haystack , invert ) : <EOL> matched = re . search ( needle , haystack , re . M ) <EOL> if ( invert and matched ) or ( not invert and not matched ) : <EOL> raise AssertionError ( "<STR_LIT>" % ( <EOL> needle , <EOL> "<STR_LIT>" if invert else "<STR_LIT>" , <EOL> haystack <EOL> ) ) <EOL> assert_contains = partial ( _assert_contains , invert = False ) <EOL> assert_not_contains = partial ( _assert_contains , invert = True ) <EOL> def line_prefix ( prefix , string ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT:\n>" . join ( prefix + x for x in string . splitlines ( ) ) <EOL> def eq_ ( result , expected , msg = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : expected , '<STR_LIT:result>' : result } <EOL> aka = """<STR_LIT>""" % params <EOL> default_msg = """<STR_LIT>""" % params <EOL> if ( repr ( result ) != str ( result ) ) or ( repr ( expected ) != str ( expected ) ) : <EOL> default_msg += aka <EOL> assert result == expected , msg or default_msg <EOL> def eq_contents ( path , text ) : <EOL> with open ( path ) as fd : <EOL> eq_ ( text , fd . read ( ) ) <EOL> def support ( path ) : <EOL> return os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' , path ) <EOL> fabfile = support <EOL> @ contextmanager <EOL> def path_prefix ( module ) : <EOL> i = <NUM_LIT:0> <EOL> sys . path . insert ( i , os . path . dirname ( module ) ) <EOL> yield <EOL> sys . path . pop ( i ) <EOL> def aborts ( func ) : <EOL> return raises ( SystemExit ) ( mock_streams ( '<STR_LIT>' ) ( func ) ) <EOL> def _patched_input ( func , fake ) : <EOL> return func ( sys . modules [ '<STR_LIT>' ] , '<STR_LIT>' , fake ) <EOL> patched_input = partial ( _patched_input , patched_context ) <EOL> with_patched_input = partial ( _patched_input , with_patched_object ) </s>
