1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
# Copyright (c) 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish, dis-
# tribute, sublicense, and/or sell copies of the Software, and to permit
# persons to whom the Software is furnished to do so, subject to the fol-
# lowing conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
#
import unittest
import time
from boto.elastictranscoder.layer1 import ElasticTranscoderConnection
from boto.elastictranscoder.exceptions import ValidationException
import boto.s3
import boto.sns
import boto.iam
import boto.sns
class TestETSLayer1PipelineManagement(unittest.TestCase):
def setUp(self):
self.api = ElasticTranscoderConnection()
self.s3 = boto.connect_s3()
self.sns = boto.connect_sns()
self.iam = boto.connect_iam()
self.sns = boto.connect_sns()
self.timestamp = str(int(time.time()))
self.input_bucket = 'boto-pipeline-%s' % self.timestamp
self.output_bucket = 'boto-pipeline-out-%s' % self.timestamp
self.role_name = 'boto-ets-role-%s' % self.timestamp
self.pipeline_name = 'boto-pipeline-%s' % self.timestamp
self.s3.create_bucket(self.input_bucket)
self.s3.create_bucket(self.output_bucket)
self.addCleanup(self.s3.delete_bucket, self.input_bucket)
self.addCleanup(self.s3.delete_bucket, self.output_bucket)
self.role = self.iam.create_role(self.role_name)
self.role_arn = self.role['create_role_response']['create_role_result']\
['role']['arn']
self.addCleanup(self.iam.delete_role, self.role_name)
def create_pipeline(self):
pipeline = self.api.create_pipeline(
self.pipeline_name, self.input_bucket,
self.output_bucket, self.role_arn,
{'Progressing': '', 'Completed': '', 'Warning': '', 'Error': ''})
pipeline_id = pipeline['Pipeline']['Id']
self.addCleanup(self.api.delete_pipeline, pipeline_id)
return pipeline_id
def test_create_delete_pipeline(self):
pipeline = self.api.create_pipeline(
self.pipeline_name, self.input_bucket,
self.output_bucket, self.role_arn,
{'Progressing': '', 'Completed': '', 'Warning': '', 'Error': ''})
pipeline_id = pipeline['Pipeline']['Id']
self.api.delete_pipeline(pipeline_id)
def test_can_retrieve_pipeline_information(self):
pipeline_id = self.create_pipeline()
# The pipeline shows up in list_pipelines
pipelines = self.api.list_pipelines()['Pipelines']
pipeline_names = [p['Name'] for p in pipelines]
self.assertIn(self.pipeline_name, pipeline_names)
# The pipeline shows up in read_pipeline
response = self.api.read_pipeline(pipeline_id)
self.assertEqual(response['Pipeline']['Id'], pipeline_id)
def test_update_pipeline(self):
pipeline_id = self.create_pipeline()
self.api.update_pipeline_status(pipeline_id, 'Paused')
response = self.api.read_pipeline(pipeline_id)
self.assertEqual(response['Pipeline']['Status'], 'Paused')
def test_update_pipeline_notification(self):
pipeline_id = self.create_pipeline()
response = self.sns.create_topic('pipeline-errors')
topic_arn = response['CreateTopicResponse']['CreateTopicResult']\
['TopicArn']
self.addCleanup(self.sns.delete_topic, topic_arn)
self.api.update_pipeline_notifications(
pipeline_id,
{'Progressing': '', 'Completed': '',
'Warning': '', 'Error': topic_arn})
response = self.api.read_pipeline(pipeline_id)
self.assertEqual(response['Pipeline']['Notifications']['Error'],
topic_arn)
def test_list_jobs_by_pipeline(self):
pipeline_id = self.create_pipeline()
response = self.api.list_jobs_by_pipeline(pipeline_id)
self.assertEqual(response['Jobs'], [])
def test_proper_error_when_pipeline_does_not_exist(self):
with self.assertRaises(ValidationException):
self.api.read_pipeline('badpipelineid')
|
