<s> import os <EOL> import pytest <EOL> import s3_integration_help <EOL> import sys <EOL> from wal_e import cmd <EOL> _PREFIX_VARS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> _AWS_CRED_ENV_VARS = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> class AwsTestConfig ( object ) : <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , request ) : <EOL> self . env_vars = { } <EOL> self . monkeypatch = request . getfuncargvalue ( '<STR_LIT>' ) <EOL> for name in _AWS_CRED_ENV_VARS : <EOL> maybe_value = os . getenv ( name ) <EOL> self . env_vars [ name ] = maybe_value <EOL> def patch ( self , test_name , default_test_bucket ) : <EOL> for name in _PREFIX_VARS : <EOL> self . monkeypatch . delenv ( name , raising = False ) <EOL> for name , value in self . env_vars . iteritems ( ) : <EOL> if value is None : <EOL> self . monkeypatch . delenv ( name , raising = False ) <EOL> else : <EOL> self . monkeypatch . setenv ( name , value ) <EOL> self . monkeypatch . setenv ( '<STR_LIT>' , '<STR_LIT>' <EOL> . format ( default_test_bucket , test_name ) ) <EOL> self . monkeypatch . setenv ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def main ( self , * args ) : <EOL> self . monkeypatch . setattr ( sys , '<STR_LIT>' , [ '<STR_LIT>' ] + list ( args ) ) <EOL> return cmd . main ( ) <EOL> class AwsTestConfigSetImpl ( AwsTestConfig ) : <EOL> name = '<STR_LIT>' <EOL> def patch ( self , test_name , default_test_bucket ) : <EOL> self . monkeypatch . setenv ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> return AwsTestConfig . patch ( self , test_name , default_test_bucket ) <EOL> class AwsInstanceProfileTestConfig ( object ) : <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , request ) : <EOL> self . request = request <EOL> self . monkeypatch = request . getfuncargvalue ( '<STR_LIT>' ) <EOL> def patch ( self , test_name , default_test_bucket ) : <EOL> c = s3_integration_help . sts_conn ( ) <EOL> policy = s3_integration_help . make_policy ( default_test_bucket , <EOL> test_name ) <EOL> fed = c . get_federation_token ( default_test_bucket , policy = policy ) <EOL> for name in _AWS_CRED_ENV_VARS : <EOL> self . monkeypatch . delenv ( name , raising = False ) <EOL> self . monkeypatch . setenv ( '<STR_LIT>' , '<STR_LIT>' <EOL> . format ( default_test_bucket , test_name ) ) <EOL> self . monkeypatch . setenv ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> m = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : fed . credentials . access_key , <EOL> "<STR_LIT>" : fed . credentials . secret_key , <EOL> "<STR_LIT>" : fed . credentials . session_token , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> from boto import provider <EOL> self . monkeypatch . setattr ( provider . Provider , <EOL> '<STR_LIT>' , <EOL> lambda self : False ) <EOL> import test_aws_instance_profiles <EOL> if test_aws_instance_profiles . boto_flat_metadata ( ) : <EOL> m = { '<STR_LIT>' : m } <EOL> else : <EOL> m = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : m } } } <EOL> from boto import utils <EOL> self . monkeypatch . setattr ( utils , '<STR_LIT>' , <EOL> lambda * args , ** kwargs : m ) <EOL> def main ( self , * args ) : <EOL> self . monkeypatch . setattr ( <EOL> sys , '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] + list ( args ) ) <EOL> return cmd . main ( ) <EOL> def _make_fixture_param_and_ids ( ) : <EOL> ret = { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> def _add_config ( c ) : <EOL> ret [ '<STR_LIT>' ] . append ( c ) <EOL> ret [ '<STR_LIT>' ] . append ( c . name ) <EOL> if not s3_integration_help . no_real_s3_credentials ( ) : <EOL> _add_config ( AwsTestConfig ) <EOL> _add_config ( AwsTestConfigSetImpl ) <EOL> _add_config ( AwsInstanceProfileTestConfig ) <EOL> return ret <EOL> @ pytest . fixture ( ** _make_fixture_param_and_ids ( ) ) <EOL> def config ( request , monkeypatch , default_test_bucket ) : <EOL> config = request . param ( request ) <EOL> config . patch ( request . node . name , default_test_bucket ) <EOL> return config <EOL> class NoopPgBackupStatements ( object ) : <EOL> @ classmethod <EOL> def run_start_backup ( cls ) : <EOL> name = '<STR_LIT:0>' * <NUM_LIT:8> * <NUM_LIT:3> <EOL> offset = '<STR_LIT:0>' * <NUM_LIT:8> <EOL> return { '<STR_LIT>' : name , <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> @ classmethod <EOL> def run_stop_backup ( cls ) : <EOL> name = '<STR_LIT:1>' * <NUM_LIT:8> * <NUM_LIT:3> <EOL> offset = '<STR_LIT:1>' * <NUM_LIT:8> <EOL> return { '<STR_LIT>' : name , <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> @ classmethod <EOL> def pg_version ( cls ) : <EOL> return { '<STR_LIT:version>' : '<STR_LIT>' } <EOL> @ pytest . fixture <EOL> def noop_pg_backup_statements ( monkeypatch ) : <EOL> import wal_e . operator . backup <EOL> monkeypatch . setattr ( wal_e . operator . backup , '<STR_LIT>' , <EOL> NoopPgBackupStatements ) <EOL> monkeypatch . setattr ( wal_e . cmd , '<STR_LIT>' , <EOL> lambda * args , ** kwargs : None ) <EOL> @ pytest . fixture <EOL> def small_push_dir ( tmpdir ) : <EOL> """<STR_LIT>""" <EOL> contents = '<STR_LIT>' * <NUM_LIT> <EOL> push_dir = tmpdir . join ( '<STR_LIT>' ) . ensure ( dir = True ) <EOL> push_dir . join ( '<STR_LIT>' ) . write ( contents ) <EOL> push_dir . join ( '<STR_LIT>' ) . mksymlinkto ( '<STR_LIT>' ) <EOL> push_dir . join ( '<STR_LIT>' ) . ensure ( ) <EOL> return push_dir </s>
