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
|
#!/usr/bin/env python
import json
from boto.cloudtrail.layer1 import CloudTrailConnection
from tests.unit import AWSMockServiceTestCase
class TestDescribeTrails(AWSMockServiceTestCase):
connection_class = CloudTrailConnection
def default_body(self):
return b'''
{"trailList":
[
{
"IncludeGlobalServiceEvents": false,
"Name": "test",
"SnsTopicName": "cloudtrail-1",
"S3BucketName": "cloudtrail-1"
}
]
}'''
def test_describe(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.describe_trails()
self.assertEqual(1, len(api_response['trailList']))
self.assertEqual('test', api_response['trailList'][0]['Name'])
self.assert_request_parameters({})
target = self.actual_request.headers['X-Amz-Target']
self.assertTrue('DescribeTrails' in target)
def test_describe_name_list(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.describe_trails(
trail_name_list=['test'])
self.assertEqual(1, len(api_response['trailList']))
self.assertEqual('test', api_response['trailList'][0]['Name'])
self.assertEqual(json.dumps({
'trailNameList': ['test']
}), self.actual_request.body.decode('utf-8'))
target = self.actual_request.headers['X-Amz-Target']
self.assertTrue('DescribeTrails' in target)
class TestCreateTrail(AWSMockServiceTestCase):
connection_class = CloudTrailConnection
def default_body(self):
return b'''
{"trail":
{
"IncludeGlobalServiceEvents": false,
"Name": "test",
"SnsTopicName": "cloudtrail-1",
"S3BucketName": "cloudtrail-1"
}
}'''
def test_create(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.create_trail(
'test', 'cloudtrail-1', sns_topic_name='cloudtrail-1',
include_global_service_events=False)
self.assertEqual('test', api_response['trail']['Name'])
self.assertEqual('cloudtrail-1', api_response['trail']['S3BucketName'])
self.assertEqual('cloudtrail-1', api_response['trail']['SnsTopicName'])
self.assertEqual(False,
api_response['trail']['IncludeGlobalServiceEvents'])
target = self.actual_request.headers['X-Amz-Target']
self.assertTrue('CreateTrail' in target)
|
