~tribaal/txaws/xss-hardening

« back to all changes in this revision

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

  • Committer: kapil.thangavelu at canonical
  • Date: 2011-09-22 20:17:22 UTC
  • mfrom: (91.2.10 fix-s3-port)
  • mto: This revision was merged to the branch mainline in revision 99.
  • Revision ID: kapil.thangavelu@canonical.com-20110922201722-l4plcnk5wrr7eten
merge fix-s3-port-and-bucket-op [a=hazmat,jimbaker,spampas][r=therve][f=829609,824403]

Openstack compatibiity fixes.

 - S3 bucket operations use a trailing slash.
 - S3 Host and port are now configurable.
 - Security group parsing is now more forgiving.

Show diffs side-by-side

added added

removed removed

Lines of Context:
34
34
 
35
35
    def test_get_path_with_bucket(self):
36
36
        url_context = client.URLContext(self.endpoint, bucket="mystuff")
37
 
        self.assertEquals(url_context.get_path(), "/mystuff")
 
37
        self.assertEquals(url_context.get_path(), "/mystuff/")
38
38
 
39
39
    def test_get_path_with_bucket_and_object(self):
40
40
        url_context = client.URLContext(
62
62
            url_context.get_url(),
63
63
            "http://localhost/mydocs/notes.txt")
64
64
 
 
65
    def test_custom_port_endpoint(self):
 
66
        test_uri='http://0.0.0.0:12345/'
 
67
        endpoint = AWSServiceEndpoint(uri=test_uri)
 
68
        self.assertEquals(endpoint.port, 12345)
 
69
        self.assertEquals(endpoint.scheme, 'http')
 
70
        context = client.URLContext(service_endpoint=endpoint,
 
71
                bucket="foo",
 
72
                object_name="bar")
 
73
        self.assertEquals(context.get_host(), '0.0.0.0')
 
74
        self.assertEquals(context.get_url(), test_uri + 'foo/bar')
 
75
 
 
76
    def test_custom_port_endpoint_https(self):
 
77
        test_uri='https://0.0.0.0:12345/'
 
78
        endpoint = AWSServiceEndpoint(uri=test_uri)
 
79
        self.assertEquals(endpoint.port, 12345)
 
80
        self.assertEquals(endpoint.scheme, 'https')
 
81
        context = client.URLContext(service_endpoint=endpoint,
 
82
                bucket="foo",
 
83
                object_name="bar")
 
84
        self.assertEquals(context.get_host(), '0.0.0.0')
 
85
        self.assertEquals(context.get_url(), test_uri + 'foo/bar')
 
86
 
 
87
 
65
88
URLContextTestCase.skip = s3clientSkip
66
89
 
67
90
 
611
634
    def test_get_canonicalized_resource(self):
612
635
        query = client.Query(action="PUT", bucket="images")
613
636
        result = query.get_canonicalized_resource()
614
 
        self.assertEquals(result, "/images")
 
637
        self.assertEquals(result, "/images/")
615
638
 
616
639
    def test_get_canonicalized_resource_with_object_name(self):
617
640
        query = client.Query(
619
642
        result = query.get_canonicalized_resource()
620
643
        self.assertEquals(result, "/images/advicedog.jpg")
621
644
 
 
645
    def test_get_canonicalized_resource_with_slashed_object_name(self):
 
646
        query = client.Query(
 
647
            action="PUT", bucket="images", object_name="/advicedog.jpg")
 
648
        result = query.get_canonicalized_resource()
 
649
        self.assertEquals(result, "/images/advicedog.jpg")
 
650
 
622
651
    def test_sign(self):
623
652
        query = client.Query(action="PUT", creds=self.creds)
624
653
        signed = query.sign({})