diff options
Diffstat (limited to 'test.py')
| -rw-r--r-- | test.py | 67 |
1 files changed, 42 insertions, 25 deletions
@@ -1,12 +1,19 @@ import unittest import os import requests -from awsauth import S3Auth import hashlib -import base64 +import sys +from awsauth import S3Auth + +PY3 = sys.version > '3' + +if PY3: + from base64 import encodebytes as encodestring +else: + from base64 import encodestring -TEST_BUCKET = 'testpolpol' +TEST_BUCKET = 'testpolpol2' ACCESS_KEY = 'ACCESSKEYXXXXXXXXXXXX' SECRET_KEY = 'AWSSECRETKEYXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' if 'AWS_ACCESS_KEY' in os.environ: @@ -14,31 +21,40 @@ if 'AWS_ACCESS_KEY' in os.environ: if 'AWS_SECRET_KEY' in os.environ: SECRET_KEY = os.environ['AWS_SECRET_KEY'] + class TestAWS(unittest.TestCase): def setUp(self): - self.auth=S3Auth(ACCESS_KEY, SECRET_KEY) - + self.auth = S3Auth(ACCESS_KEY, SECRET_KEY) + + def get_content_md5(self, data): + hashdig = hashlib.md5(data.encode('utf-8').strip()).digest() + signature = encodestring(hashdig) + if PY3: + return signature.decode('utf-8').strip() + return signature.strip() + def test_put_get_delete(self): + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/myfile.txt' testdata = 'Sam is sweet' - r = requests.put('http://'+ TEST_BUCKET + '.s3.amazonaws.com/myfile.txt', data=testdata, auth=self.auth) + r = requests.put(url, data=testdata, auth=self.auth) self.assertEqual(r.status_code, 200) # Downloading a file - r = requests.get('http://'+ TEST_BUCKET + '.s3.amazonaws.com/myfile.txt', auth=self.auth) - self.assertEqual(r.status_code, 200) + r = requests.get(url, auth=self.auth) + self.assertEqual(r.status_code, 200) self.assertEqual(r.text, 'Sam is sweet') # Removing a file - r = requests.delete('http://'+ TEST_BUCKET + '.s3.amazonaws.com/myfile.txt', auth=self.auth) - self.assertEqual(r.status_code, 204) - + r = requests.delete(url, auth=self.auth) + self.assertEqual(r.status_code, 204) + def test_put_get_delete_filnamehasplus(self): testdata = 'Sam is sweet' filename = 'my+file.txt' - url = 'http://'+ TEST_BUCKET + '.s3.amazonaws.com/%s'%(filename) + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/%s' % (filename) r = requests.put(url, data=testdata, auth=self.auth) self.assertEqual(r.status_code, 200) # Downloading a file r = requests.get(url, auth=self.auth) - self.assertEqual(r.status_code, 200) + self.assertEqual(r.status_code, 200) self.assertEqual(r.text, testdata) # Removing a file r = requests.delete(url, auth=self.auth) @@ -47,19 +63,19 @@ class TestAWS(unittest.TestCase): def test_put_get_delete_filname_encoded(self): testdata = 'Sam is sweet' filename = 'my%20file.txt' - url = 'http://'+ TEST_BUCKET + '.s3.amazonaws.com/%s'%(filename) + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/%s' % (filename) r = requests.put(url, data=testdata, auth=self.auth) self.assertEqual(r.status_code, 200) # Downloading a file r = requests.get(url, auth=self.auth) - self.assertEqual(r.status_code, 200) + self.assertEqual(r.status_code, 200) self.assertEqual(r.text, testdata) # Removing a file r = requests.delete(url, auth=self.auth) self.assertEqual(r.status_code, 204) def test_put_get_delete_cors(self): - url = 'http://'+ TEST_BUCKET + '.s3.amazonaws.com/?cors' + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/?cors' testdata = '<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">\ <CORSRule>\ <AllowedOrigin>*</AllowedOrigin>\ @@ -68,8 +84,8 @@ class TestAWS(unittest.TestCase): <AllowedHeader>Authorization</AllowedHeader>\ </CORSRule>\ </CORSConfiguration>' - content_md5 = base64.encodestring(hashlib.md5(testdata.strip()).digest()).strip() - r = requests.put(url, data=testdata, auth=self.auth, headers={'content-md5': content_md5}) + headers = {'content-md5': self.get_content_md5(testdata)} + r = requests.put(url, data=testdata, auth=self.auth, headers=headers) self.assertEqual(r.status_code, 200) # Downloading current cors configuration r = requests.get(url, auth=self.auth) @@ -79,7 +95,7 @@ class TestAWS(unittest.TestCase): self.assertEqual(r.status_code, 204) def test_put_get_delete_tagging(self): - url = 'http://'+ TEST_BUCKET + '.s3.amazonaws.com/?tagging' + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/?tagging' testdata = '<Tagging>\ <TagSet>\ <Tag>\ @@ -88,8 +104,8 @@ class TestAWS(unittest.TestCase): </Tag>\ </TagSet>\ </Tagging>' - content_md5 = base64.encodestring(hashlib.md5(testdata.strip()).digest()).strip() - r = requests.put(url, data=testdata, auth=self.auth, headers={'content-md5': content_md5}) + headers = {'content-md5': self.get_content_md5(testdata)} + r = requests.put(url, data=testdata, auth=self.auth, headers=headers) self.assertEqual(r.status_code, 204) # Downloading current cors configuration r = requests.get(url, auth=self.auth) @@ -99,15 +115,16 @@ class TestAWS(unittest.TestCase): self.assertEqual(r.status_code, 204) def test_put_get_notification(self): - url = 'http://'+ TEST_BUCKET + '.s3.amazonaws.com/?notification' + url = 'http://' + TEST_BUCKET + '.s3.amazonaws.com/?notification' testdata = '<NotificationConfiguration></NotificationConfiguration>' - content_md5 = base64.encodestring(hashlib.md5(testdata.strip()).digest()).strip() - r = requests.put(url, data=testdata, auth=self.auth, headers={'content-md5': content_md5}) + headers = {'content-md5': self.get_content_md5(testdata)} + r = requests.put(url, data=testdata, auth=self.auth, headers=headers) self.assertEqual(r.status_code, 200) # Downloading current cors configuration r = requests.get(url, auth=self.auth) self.assertEqual(r.status_code, 200) - # No Delete ?notification API, empty <NotificationConfiguration> tag is default + # No Delete ?notification API, empty <NotificationConfiguration> + # tag is default if __name__ == '__main__': unittest.main() |