<s> import time <EOL> from boto . elastictranscoder . layer1 import ElasticTranscoderConnection <EOL> from boto . elastictranscoder . exceptions import ValidationException <EOL> from tests . compat import unittest <EOL> import boto . s3 <EOL> import boto . sns <EOL> import boto . iam <EOL> import boto . sns <EOL> class TestETSLayer1PipelineManagement ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . api = ElasticTranscoderConnection ( ) <EOL> self . s3 = boto . connect_s3 ( ) <EOL> self . sns = boto . connect_sns ( ) <EOL> self . iam = boto . connect_iam ( ) <EOL> self . sns = boto . connect_sns ( ) <EOL> self . timestamp = str ( int ( time . time ( ) ) ) <EOL> self . input_bucket = '<STR_LIT>' % self . timestamp <EOL> self . output_bucket = '<STR_LIT>' % self . timestamp <EOL> self . role_name = '<STR_LIT>' % self . timestamp <EOL> self . pipeline_name = '<STR_LIT>' % self . timestamp <EOL> self . s3 . create_bucket ( self . input_bucket ) <EOL> self . s3 . create_bucket ( self . output_bucket ) <EOL> self . addCleanup ( self . s3 . delete_bucket , self . input_bucket ) <EOL> self . addCleanup ( self . s3 . delete_bucket , self . output_bucket ) <EOL> self . role = self . iam . create_role ( self . role_name ) <EOL> self . role_arn = self . role [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . addCleanup ( self . iam . delete_role , self . role_name ) <EOL> def create_pipeline ( self ) : <EOL> pipeline = self . api . create_pipeline ( <EOL> self . pipeline_name , self . input_bucket , <EOL> self . output_bucket , self . role_arn , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> pipeline_id = pipeline [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . addCleanup ( self . api . delete_pipeline , pipeline_id ) <EOL> return pipeline_id <EOL> def test_create_delete_pipeline ( self ) : <EOL> pipeline = self . api . create_pipeline ( <EOL> self . pipeline_name , self . input_bucket , <EOL> self . output_bucket , self . role_arn , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> pipeline_id = pipeline [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . api . delete_pipeline ( pipeline_id ) <EOL> def test_can_retrieve_pipeline_information ( self ) : <EOL> pipeline_id = self . create_pipeline ( ) <EOL> pipelines = self . api . list_pipelines ( ) [ '<STR_LIT>' ] <EOL> pipeline_names = [ p [ '<STR_LIT:Name>' ] for p in pipelines ] <EOL> self . assertIn ( self . pipeline_name , pipeline_names ) <EOL> response = self . api . read_pipeline ( pipeline_id ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] [ '<STR_LIT>' ] , pipeline_id ) <EOL> def test_update_pipeline ( self ) : <EOL> pipeline_id = self . create_pipeline ( ) <EOL> self . api . update_pipeline_status ( pipeline_id , '<STR_LIT>' ) <EOL> response = self . api . read_pipeline ( pipeline_id ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_update_pipeline_notification ( self ) : <EOL> pipeline_id = self . create_pipeline ( ) <EOL> response = self . sns . create_topic ( '<STR_LIT>' ) <EOL> topic_arn = response [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . addCleanup ( self . sns . delete_topic , topic_arn ) <EOL> self . api . update_pipeline_notifications ( <EOL> pipeline_id , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : topic_arn } ) <EOL> response = self . api . read_pipeline ( pipeline_id ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> topic_arn ) <EOL> def test_list_jobs_by_pipeline ( self ) : <EOL> pipeline_id = self . create_pipeline ( ) <EOL> response = self . api . list_jobs_by_pipeline ( pipeline_id ) <EOL> self . assertEqual ( response [ '<STR_LIT>' ] , [ ] ) <EOL> def test_proper_error_when_pipeline_does_not_exist ( self ) : <EOL> with self . assertRaises ( ValidationException ) : <EOL> self . api . read_pipeline ( '<STR_LIT>' ) </s>
