+ def test_update_object(self,
+ first_byte_pos=0,
+ last_byte_pos=499,
+ instance_length = True,
+ content_length = 500):
+ l = len(self.obj['data'])
+ if instance_length:
+ range = 'bytes %d-%d/%d' %(first_byte_pos,
+ last_byte_pos,
+ l)
+ else:
+ range = 'bytes %d-%d/*' %(first_byte_pos,
+ last_byte_pos)
+ partial = last_byte_pos - first_byte_pos + 1
+ data = get_random_data(partial)
+ more = {'HTTP_CONTENT_RANGE':range}
+ if content_length:
+ more.update({'CONTENT_LENGTH':'%s' % content_length})
+
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **more)
+
+ if partial < 0 or (instance_length and l <= last_byte_pos):
+ self.assertEqual(r.status_code, 202)
+ elif content_length and content_length != partial:
+ self.assertEqual(r.status_code, 400)
+ else:
+ self.assertEqual(r.status_code, 204)
+
+ #check modified object
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(r.content[0:partial], data)
+ self.assertEqual(r.content[partial:l], self.obj['data'][partial:l])
+
+ def test_update_object_no_content_length(self):
+ self.test_update_object(content_length = None)
+
+ def test_update_object_invalid_content_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(content_length = 1000)
+
+ def test_update_object_with_unknown_instance_length(self):
+ self.test_update_object(instance_length = False)
+
+ def test_update_object_invalid_range(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True)
+
+ def test_update_object_invalid_range_and_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True, -1)
+
+ def test_update_object_invalid_range_with_no_content_length(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ self.test_update_object(499, 0, True, content_length = None)
+
+ def test_update_object_out_of_limits(self):
+ with AssertContentInvariant(self.get_object, self.account, self.containers[0],
+ self.obj['name']):
+ l = len(self.obj['data'])
+ self.test_update_object(0, l+1, True)
+
+ def test_append(self):
+ data = get_random_data(500)
+ more = {'CONTENT_LENGTH':'500',
+ 'HTTP_CONTENT_RANGE':'bytes */*'}
+
+ r = self.update_object(self.account,
+ self.containers[0],
+ self.obj['name'],
+ data,
+ 'application/octet-stream',
+ **more)
+
+ self.assertEqual(r.status_code, 204)
+
+ r = self.get_object(self.account,
+ self.containers[0],
+ self.obj['name'])
+ self.assertEqual(len(r.content), len(self.obj['data']) + 500)
+
+ #def test_update_with_chunked_transfer(self):
+ # linenum = 5
+ # data = create_random_chunked_data(linenum)
+ # print data
+ # first_byte_pos=0,
+ # last_byte_pos=499,
+ # instance_length = True,
+ # content_length = 500
+ # l = len(self.obj['data'])
+ # meta = {'HTTP_TRANSFER_ENCODING':'chunked',
+ # 'HTTP_CONTENT_RANGE':'bytes 0-499/%d' %l}
+ # r = self.update_object(self.account,
+ # self.containers[0],
+ # self.obj['name'],
+ # data,
+ # 'application/octet-stream',
+ # **meta)
+ # print r.request, r.status_code, r
+