|
|
@@ -193,6 +193,28 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
|
|
|
" access generated url")
|
|
|
self.assertEqual(requests.get(url).content, content)
|
|
|
|
|
|
+ @helpers.skipIfNoService(['storage.buckets'])
|
|
|
+ def test_generate_url_write_permissions(self):
|
|
|
+ name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
|
|
|
+ test_bucket = self.provider.storage.buckets.create(name)
|
|
|
+
|
|
|
+ with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
|
|
|
+ obj_name = "hello_upload_download.txt"
|
|
|
+ obj = test_bucket.objects.create(obj_name)
|
|
|
+
|
|
|
+ with cb_helpers.cleanup_action(lambda: obj.delete()):
|
|
|
+ content = b"Hello World. Generate a url."
|
|
|
+ obj.upload(content)
|
|
|
+ target_stream = BytesIO()
|
|
|
+ obj.save_content(target_stream)
|
|
|
+
|
|
|
+ url = obj.generate_url(100, write=True)
|
|
|
+ if isinstance(self.provider, TestMockHelperMixin):
|
|
|
+ raise self.skipTest(
|
|
|
+ "Skipping rest of test - mock providers can't"
|
|
|
+ " access generated url")
|
|
|
+ self.assertEqual(requests.post(url).content, content)
|
|
|
+
|
|
|
@helpers.skipIfNoService(['storage.buckets'])
|
|
|
def test_upload_download_bucket_content_from_file(self):
|
|
|
name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
|