← Back to branch summary

~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/ec2/tests/test_client.py

  • Committer: Duncan McGreggor
  • Date: 2009-09-05 00:26:12 UTC
  • mto: This revision was merged to the branch mainline in revision 17.
  • Revision ID: duncan@canonical.com-20090905002612-cuk3ndv1tkpadgag
Replaced all occurances of ' in all the modules with " (excepting proper usage
of apostrophes).

Show diffs side-by-side

added added

removed removed

Lines of Context:
txaws/ec2/tests/test_client.py
50
50
class EC2ClientTestCase(TXAWSTestCase):
51
51
 
52
52
    def test_init_no_creds(self):
53
 
        os.environ['AWS_SECRET_ACCESS_KEY'] = 'foo'
54
 
        os.environ['AWS_ACCESS_KEY_ID'] = 'bar'
 
53
        os.environ["AWS_SECRET_ACCESS_KEY"] = "foo"
 
54
        os.environ["AWS_ACCESS_KEY_ID"] = "bar"
55
55
        ec2 = client.EC2Client()
56
56
        self.assertNotEqual(None, ec2.creds)
57
57
 
129
129
    def test_describe_instances(self):
130
130
        class StubQuery(object):
131
131
            def __init__(stub, action, creds, endpoint):
132
 
                self.assertEqual(action, 'DescribeInstances')
 
132
                self.assertEqual(action, "DescribeInstances")
133
133
                self.assertEqual(creds.access_key, "foo")
134
134
                self.assertEqual(creds.secret_key, "bar")
135
135
            def submit(self):
143
143
    def test_describe_instances_required(self):
144
144
        class StubQuery(object):
145
145
            def __init__(stub, action, creds, endpoint):
146
 
                self.assertEqual(action, 'DescribeInstances')
 
146
                self.assertEqual(action, "DescribeInstances")
147
147
                self.assertEqual(creds.access_key, "foo")
148
148
                self.assertEqual(creds.secret_key, "bar")
149
149
            def submit(self):
158
158
    def test_terminate_instances(self):
159
159
        class StubQuery(object):
160
160
            def __init__(stub, action, creds, endpoint, other_params):
161
 
                self.assertEqual(action, 'TerminateInstances')
 
161
                self.assertEqual(action, "TerminateInstances")
162
162
                self.assertEqual(creds.access_key, "foo")
163
163
                self.assertEqual(creds.secret_key, "bar")
164
164
                self.assertEqual(
165
 
                    {'InstanceId.1': 'i-1234', 'InstanceId.2': 'i-5678'},
 
165
                    {"InstanceId.1": "i-1234", "InstanceId.2": "i-5678"},
166
166
                    other_params)
167
167
            def submit(self):
168
168
                return succeed(payload.sample_terminate_instances_result)
170
170
        endpoint = AWSServiceEndpoint(uri=EC2_ENDPOINT_US)
171
171
        ec2 = client.EC2Client(creds=creds, endpoint=endpoint,
172
172
                               query_factory=StubQuery)
173
 
        d = ec2.terminate_instances('i-1234', 'i-5678')
 
173
        d = ec2.terminate_instances("i-1234", "i-5678")
174
174
        def check_transition(changes):
175
 
            self.assertEqual([('i-1234', 'running', 'shutting-down'),
176
 
                ('i-5678', 'shutting-down', 'shutting-down')], sorted(changes))
 
175
            self.assertEqual([("i-1234", "running", "shutting-down"),
 
176
                ("i-5678", "shutting-down", "shutting-down")], sorted(changes))
177
177
        d.addCallback(check_transition)
178
178
        return d
179
179
 
288
288
 
289
289
    def setUp(self):
290
290
        TXAWSTestCase.setUp(self)
291
 
        self.creds = AWSCredentials('foo', 'bar')
 
291
        self.creds = AWSCredentials("foo", "bar")
292
292
        self.endpoint = AWSServiceEndpoint(uri=EC2_ENDPOINT_US)
293
293
 
294
294
    def test_init_minimum(self):
295
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint)
296
 
        self.assertTrue('Timestamp' in query.params)
297
 
        del query.params['Timestamp']
 
295
        query = client.Query("DescribeInstances", self.creds, self.endpoint)
 
296
        self.assertTrue("Timestamp" in query.params)
 
297
        del query.params["Timestamp"]
298
298
        self.assertEqual(
299
 
            {'AWSAccessKeyId': 'foo',
300
 
             'Action': 'DescribeInstances',
301
 
             'SignatureMethod': 'HmacSHA1',
302
 
             'SignatureVersion': '2',
303
 
             'Version': '2008-12-01'},
 
299
            {"AWSAccessKeyId": "foo",
 
300
             "Action": "DescribeInstances",
 
301
             "SignatureMethod": "HmacSHA1",
 
302
             "SignatureVersion": "2",
 
303
             "Version": "2008-12-01"},
304
304
            query.params)
305
305
 
306
306
    def test_init_requires_action(self):
310
310
        self.assertRaises(TypeError, client.Query, None)
311
311
 
312
312
    def test_init_other_args_are_params(self):
313
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint,
314
 
            {'InstanceId.0': '12345'},
 
313
        query = client.Query("DescribeInstances", self.creds, self.endpoint,
 
314
            {"InstanceId.0": "12345"},
315
315
            time_tuple=(2007,11,12,13,14,15,0,0,0))
316
316
        self.assertEqual(
317
 
            {'AWSAccessKeyId': 'foo',
318
 
             'Action': 'DescribeInstances',
319
 
             'InstanceId.0': '12345',
320
 
             'SignatureMethod': 'HmacSHA1',
321
 
             'SignatureVersion': '2',
322
 
             'Timestamp': '2007-11-12T13:14:15Z',
323
 
             'Version': '2008-12-01'},
 
317
            {"AWSAccessKeyId": "foo",
 
318
             "Action": "DescribeInstances",
 
319
             "InstanceId.0": "12345",
 
320
             "SignatureMethod": "HmacSHA1",
 
321
             "SignatureVersion": "2",
 
322
             "Timestamp": "2007-11-12T13:14:15Z",
 
323
             "Version": "2008-12-01"},
324
324
            query.params)
325
325
 
326
326
    def test_sorted_params(self):
327
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint,
328
 
            {'fun': 'games'},
 
327
        query = client.Query("DescribeInstances", self.creds, self.endpoint,
 
328
            {"fun": "games"},
329
329
            time_tuple=(2007,11,12,13,14,15,0,0,0))
330
330
        self.assertEqual([
331
 
            ('AWSAccessKeyId', 'foo'),
332
 
            ('Action', 'DescribeInstances'),
333
 
            ('SignatureMethod', 'HmacSHA1'),
334
 
            ('SignatureVersion', '2'),
335
 
            ('Timestamp', '2007-11-12T13:14:15Z'),
336
 
            ('Version', '2008-12-01'),
337
 
            ('fun', 'games'),
 
331
            ("AWSAccessKeyId", "foo"),
 
332
            ("Action", "DescribeInstances"),
 
333
            ("SignatureMethod", "HmacSHA1"),
 
334
            ("SignatureVersion", "2"),
 
335
            ("Timestamp", "2007-11-12T13:14:15Z"),
 
336
            ("Version", "2008-12-01"),
 
337
            ("fun", "games"),
338
338
            ], query.sorted_params())
339
339
 
340
340
    def test_encode_unreserved(self):
341
 
        all_unreserved = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'
342
 
            'abcdefghijklmnopqrstuvwxyz0123456789-_.~')
343
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint)
 
341
        all_unreserved = ("ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
342
            "abcdefghijklmnopqrstuvwxyz0123456789-_.~")
 
343
        query = client.Query("DescribeInstances", self.creds, self.endpoint)
344
344
        self.assertEqual(all_unreserved, query.encode(all_unreserved))
345
345
 
346
346
    def test_encode_space(self):
347
 
        """This may be just 'url encode', but the AWS manual isn't clear."""
348
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint)
349
 
        self.assertEqual('a%20space', query.encode('a space'))
 
347
        """This may be just "url encode", but the AWS manual isn't clear."""
 
348
        query = client.Query("DescribeInstances", self.creds, self.endpoint)
 
349
        self.assertEqual("a%20space", query.encode("a space"))
350
350
 
351
351
    def test_canonical_query(self):
352
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint,
353
 
            {'fu n': 'g/ames', 'argwithnovalue':'',
354
 
             'InstanceId.1': 'i-1234'},
 
352
        query = client.Query("DescribeInstances", self.creds, self.endpoint,
 
353
            {"fu n": "g/ames", "argwithnovalue":"",
 
354
             "InstanceId.1": "i-1234"},
355
355
            time_tuple=(2007,11,12,13,14,15,0,0,0))
356
 
        expected_query = ('AWSAccessKeyId=foo&Action=DescribeInstances'
357
 
            '&InstanceId.1=i-1234'
358
 
            '&SignatureMethod=HmacSHA1&SignatureVersion=2&'
359
 
            'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01&'
360
 
            'argwithnovalue=&fu%20n=g%2Fames')
 
356
        expected_query = ("AWSAccessKeyId=foo&Action=DescribeInstances"
 
357
            "&InstanceId.1=i-1234"
 
358
            "&SignatureMethod=HmacSHA1&SignatureVersion=2&"
 
359
            "Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01&"
 
360
            "argwithnovalue=&fu%20n=g%2Fames")
361
361
        self.assertEqual(expected_query, query.canonical_query_params())
362
362
 
363
363
    def test_signing_text(self):
364
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint,
 
364
        query = client.Query("DescribeInstances", self.creds, self.endpoint,
365
365
            time_tuple=(2007,11,12,13,14,15,0,0,0))
366
 
        signing_text = ('GET\n%s\n/\n' % self.endpoint.host +
367
 
            'AWSAccessKeyId=foo&Action=DescribeInstances&'
368
 
            'SignatureMethod=HmacSHA1&SignatureVersion=2&'
369
 
            'Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01')
 
366
        signing_text = ("GET\n%s\n/\n" % self.endpoint.host +
 
367
            "AWSAccessKeyId=foo&Action=DescribeInstances&"
 
368
            "SignatureMethod=HmacSHA1&SignatureVersion=2&"
 
369
            "Timestamp=2007-11-12T13%3A14%3A15Z&Version=2008-12-01")
370
370
        self.assertEqual(signing_text, query.signing_text())
371
371
 
372
372
    def test_sign(self):
373
 
        query = client.Query('DescribeInstances', self.creds, self.endpoint,
 
373
        query = client.Query("DescribeInstances", self.creds, self.endpoint,
374
374
            time_tuple=(2007,11,12,13,14,15,0,0,0))
375
375
        query.sign()
376
 
        self.assertEqual('JuCpwFA2H4OVF3Ql/lAQs+V6iMc=',
377
 
            query.params['Signature'])
 
376
        self.assertEqual("JuCpwFA2H4OVF3Ql/lAQs+V6iMc=",
 
377
            query.params["Signature"])
378
378
 
379
379
 
380
380
class TestEBS(TXAWSTestCase):