| Index: gslib/tests/test_signurl.py
|
| ===================================================================
|
| --- gslib/tests/test_signurl.py (revision 0)
|
| +++ gslib/tests/test_signurl.py (revision 0)
|
| @@ -0,0 +1,202 @@
|
| +# -*- coding: utf-8 -*-
|
| +# Copyright 2014 Google Inc. All Rights Reserved.
|
| +#
|
| +# Licensed under the Apache License, Version 2.0 (the "License");
|
| +# you may not use this file except in compliance with the License.
|
| +# You may obtain a copy of the License at
|
| +#
|
| +# http://www.apache.org/licenses/LICENSE-2.0
|
| +#
|
| +# Unless required by applicable law or agreed to in writing, software
|
| +# distributed under the License is distributed on an "AS IS" BASIS,
|
| +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
| +# See the License for the specific language governing permissions and
|
| +# limitations under the License.
|
| +"""Tests for signurl command."""
|
| +from datetime import timedelta
|
| +import pkgutil
|
| +
|
| +import gslib.commands.signurl
|
| +from gslib.commands.signurl import HAVE_OPENSSL
|
| +from gslib.exception import CommandException
|
| +import gslib.tests.testcase as testcase
|
| +from gslib.tests.testcase.integration_testcase import SkipForS3
|
| +from gslib.tests.util import ObjectToURI as suri
|
| +from gslib.tests.util import unittest
|
| +
|
| +
|
| +# pylint: disable=protected-access
|
| +@unittest.skipUnless(HAVE_OPENSSL, 'signurl requires pyopenssl.')
|
| +@SkipForS3('Signed URLs are only supported for gs:// URLs.')
|
| +class TestSignUrl(testcase.GsUtilIntegrationTestCase):
|
| + """Integration tests for signurl command."""
|
| +
|
| + def _GetKsFile(self):
|
| + if not hasattr(self, 'ks_file'):
|
| + # Dummy pkcs12 keystore generated with the command
|
| +
|
| + # openssl req -new -passout pass:notasecret -batch \
|
| + # -x509 -keyout signed_url_test.key -out signed_url_test.pem \
|
| + # -subj '/CN=test.apps.googleusercontent.com'
|
| +
|
| + # &&
|
| +
|
| + # openssl pkcs12 -export -passin pass:notasecret \
|
| + # -passout pass:notasecret -inkey signed_url_test.key \
|
| + # -in signed_url_test.pem -out test.p12
|
| +
|
| + # &&
|
| +
|
| + # rm signed_url_test.key signed_url_test.pem
|
| + contents = pkgutil.get_data('gslib', 'tests/test_data/test.p12')
|
| + self.ks_file = self.CreateTempFile(contents=contents)
|
| + return self.ks_file
|
| +
|
| + def testSignUrlOutput(self):
|
| + """Tests signurl output of a sample object."""
|
| +
|
| + object_url = self.CreateObject(contents='z')
|
| + stdout = self.RunGsUtil(['signurl', '-p', 'notasecret',
|
| + self._GetKsFile(), suri(object_url)],
|
| + return_stdout=True)
|
| +
|
| + self.assertIn(object_url.uri, stdout)
|
| + self.assertIn('test@developer.gserviceaccount.com', stdout)
|
| + self.assertIn('Expires=', stdout)
|
| + self.assertIn('\tGET\t', stdout)
|
| +
|
| + stdout = self.RunGsUtil(['signurl', '-m', 'PUT', '-p',
|
| + 'notasecret', self._GetKsFile(),
|
| + 'gs://test/test.txt'], return_stdout=True)
|
| +
|
| + self.assertIn('test@developer.gserviceaccount.com', stdout)
|
| + self.assertIn('Expires=', stdout)
|
| + self.assertIn('\tPUT\t', stdout)
|
| +
|
| + def testSignUrlWithWildcard(self):
|
| + objs = ['test1', 'test2', 'test3']
|
| + bucket = self.CreateBucket()
|
| + obj_urls = []
|
| +
|
| + for obj_name in objs:
|
| + obj_urls.append(self.CreateObject(bucket_uri=bucket,
|
| + object_name=obj_name, contents=''))
|
| +
|
| + stdout = self.RunGsUtil(['signurl', '-p',
|
| + 'notasecret', self._GetKsFile(),
|
| + suri(bucket) + '/*'], return_stdout=True)
|
| +
|
| + # Header, 3 signed urls, trailing newline
|
| + self.assertEquals(len(stdout.split('\n')), 5)
|
| +
|
| + for obj_url in obj_urls:
|
| + self.assertIn(suri(obj_url), stdout)
|
| +
|
| + def testSignUrlOfNonObjectUrl(self):
|
| + """Tests the signurl output of a non-existent file."""
|
| + self.RunGsUtil(['signurl', self._GetKsFile(), 'gs://'],
|
| + expected_status=1, stdin='notasecret')
|
| + self.RunGsUtil(['signurl', 'file://tmp/abc'], expected_status=1)
|
| +
|
| +
|
| +@unittest.skipUnless(HAVE_OPENSSL, 'signurl requires pyopenssl.')
|
| +class UnitTestSignUrl(testcase.GsUtilUnitTestCase):
|
| + """Unit tests for the signurl command."""
|
| +
|
| + def setUp(self):
|
| + super(UnitTestSignUrl, self).setUp()
|
| + self.ks_contents = pkgutil.get_data('gslib', 'tests/test_data/test.p12')
|
| +
|
| + def testDurationSpec(self):
|
| + tests = [('1h', timedelta(hours=1)),
|
| + ('2d', timedelta(days=2)),
|
| + ('5D', timedelta(days=5)),
|
| + ('35s', timedelta(seconds=35)),
|
| + ('1h', timedelta(hours=1)),
|
| + ('33', timedelta(hours=33)),
|
| + ('22m', timedelta(minutes=22)),
|
| + ('3.7', None),
|
| + ('27Z', None),
|
| + ]
|
| +
|
| + for inp, expected in tests:
|
| + try:
|
| + td = gslib.commands.signurl._DurationToTimeDelta(inp)
|
| + self.assertEquals(td, expected)
|
| + except CommandException:
|
| + if expected is not None:
|
| + self.fail('{0} failed to parse')
|
| +
|
| + def testSignPut(self):
|
| + """Tests the return value of the _GenSignedUrl function with \
|
| + a PUT method."""
|
| +
|
| + expected = ('https://storage.googleapis.com/test/test.txt?'
|
| + 'GoogleAccessId=test@developer.gserviceaccount.com'
|
| + '&Expires=1391816302&Signature=A6QbgTA8cXZCtjy2xCr401bdi0e'
|
| + '7zChTBQ6BX61L7AfytTGEQDMD%2BbvOQKjX7%2FsEh77cmzcSxOEKqTLUD'
|
| + 'bbkPgPqW3j8sGPSRX9VM58bgj1vt9yU8cRKoegFHXAqsATx2G5rc%2FvEl'
|
| + 'iFp9UWMfVj5TaukqlBAVuzZWlyx0aQa9tCKXRtC9YcxORxG41RfiowA2kd8'
|
| + 'XBTQt4M9XTzpVyr5rVMzfr2LvtGf9UAJvlt8p6T6nThl2vy9%2FwBoPcMFa'
|
| + 'OWQcGTagwjyKWDcI1vQPIFQLGftAcv3QnGZxZTtg8pZW%2FIxRJrBhfFfcA'
|
| + 'c62hDKyaU2YssSMy%2FjUJynWx3TIiJjhg%3D%3D')
|
| +
|
| + expiration = 1391816302
|
| + ks, client_id = (gslib.commands.signurl
|
| + ._ReadKeystore(self.ks_contents, 'notasecret'))
|
| + signed_url = (gslib.commands.signurl
|
| + ._GenSignedUrl(ks.get_privatekey(),
|
| + client_id, 'PUT', '',
|
| + '', expiration, 'test/test.txt'))
|
| + self.assertEquals(expected, signed_url)
|
| +
|
| + def testSignurlPutContentype(self):
|
| + """Tests the return value of the _GenSignedUrl function with \
|
| + a PUT method and specified content type."""
|
| +
|
| + expected = ('https://storage.googleapis.com/test/test.txt?'
|
| + 'GoogleAccessId=test@developer.gserviceaccount.com&'
|
| + 'Expires=1391816302&Signature=APn%2BCCVcQrfc1fKQXrs'
|
| + 'PEZFj9%2FmASO%2BolR8xwgBY6PbWMkcCtrUVFBauP6t4NxqZO'
|
| + 'UnbOFYTZYzul0RC57ZkEWJp3VcyDIHcn6usEE%2FTzUHhbDCDW'
|
| + 'awAkZS7p8kO8IIACuJlF5s9xZmZzaEBtzF0%2BBOsGgBPBlg2y'
|
| + 'zrhFB6cyyAwNiUgmhLQaVkdobnSwtI5QJkvXoIjJb6hhLiVbLC'
|
| + 'rWdgSZVusjAKGlWCJsM%2B4TkCR%2Bi8AnrkECngcMHuJ9mYbS'
|
| + 'XI1VfEmcnRVcfkKkJGZGctaDIWK%2FMTEmfYCW6USt3Zk2WowJ'
|
| + 'SGuJHqEcFz0kyfAlkpmG%2Fl5E1FQROYqLN2kZQ%3D%3D')
|
| +
|
| + expiration = 1391816302
|
| + ks, client_id = (gslib.commands.signurl
|
| + ._ReadKeystore(self.ks_contents,
|
| + 'notasecret'))
|
| + signed_url = (gslib.commands.signurl
|
| + ._GenSignedUrl(ks.get_privatekey(),
|
| + client_id, 'PUT', '',
|
| + 'text/plain', expiration,
|
| + 'test/test.txt'))
|
| + self.assertEquals(expected, signed_url)
|
| +
|
| + def testSignurlGet(self):
|
| + """Tests the return value of the _GenSignedUrl function with \
|
| + a GET method."""
|
| +
|
| + expected = ('https://storage.googleapis.com/test/test.txt?'
|
| + 'GoogleAccessId=test@developer.gserviceaccount.com&'
|
| + 'Expires=0&Signature=TCZwe32cU%2BMksmLiSY9shHXQjLs1'
|
| + 'F3y%2F%2F1M0UhiK4qsPRVNZVwI7YWvv2qa2Xa%2BVBBafboF0'
|
| + '1%2BWvx3ZG316pwpNIRR6y7jNnE0LvQmHE8afbm2VYCi%2B2JS'
|
| + 'ZK2YZFJAyEek8si53jhYQEmaRq1zPfGbX84B2FJ8v4iI%2FTC1'
|
| + 'I9OE5vHF0sWwIR9d73JDrFLjaync7QYFWRExdwvqlQX%2BPO3r'
|
| + 'OG9Ns%2BcQFIN7npnsVjH28yNY9gBzXya8LYmNvUx6bWHWZMiu'
|
| + 'fLwDZ0jejNeDZTOfQGRM%2B0vY7NslzaT06W1wo8P7McSkAZEl'
|
| + 'DCbhR0Vo1fturPMwmAhi88f0qzRzywbg%3D%3D')
|
| +
|
| + expiration = 0
|
| + ks, client_id = (gslib.commands.signurl
|
| + ._ReadKeystore(self.ks_contents,
|
| + 'notasecret'))
|
| + signed_url = (gslib.commands.signurl
|
| + ._GenSignedUrl(ks.get_privatekey(),
|
| + client_id, 'GET', '',
|
| + '', expiration, 'test/test.txt'))
|
| + self.assertEquals(expected, signed_url)
|
|
|
| Property changes on: gslib/tests/test_signurl.py
|
| ___________________________________________________________________
|
| Added: svn:eol-style
|
| + LF
|
|
|
|
|