Revision e4c7cbeb

b/snf-pithos-app/pithos/api/test/objects.py
1160 1160
                                   source_data[:upto - offset + 1] +
1161 1161
                                   initial_data[upto + 1:]))
1162 1162

  
1163
    def test_update_range_from_invalid_other_object(self):
1164
        src = self.object
1165
        dest = get_random_name()
1166

  
1167
        url = join_urls(self.pithos_path, self.user, self.container, src)
1168
        r = self.get(url)
1169

  
1170
        # update zero length object
1171
        url = join_urls(self.pithos_path, self.user, self.container, dest)
1172
        initial_data = get_random_data()
1173
        length = len(initial_data)
1174
        r = self.put(url, data=initial_data)
1175
        self.assertEqual(r.status_code, 201)
1176

  
1177
        offset = random.randint(1, length - 2)
1178
        upto = random.randint(offset, length - 1)
1179

  
1180
        # source object does not start with /
1181
        r = self.post(url,
1182
                      HTTP_CONTENT_RANGE='bytes %s-%s/*' % (offset, upto),
1183
                      HTTP_X_SOURCE_OBJECT='%s/%s' % (self.container, src))
1184
        self.assertEqual(r.status_code, 400)
1185

  
1186
        # source object does not exist
1187
        r = self.post(url,
1188
                      HTTP_CONTENT_RANGE='bytes %s-%s/*' % (offset, upto),
1189
                      HTTP_X_SOURCE_OBJECT='/%s/%s1' % (self.container, src))
1190
        self.assertEqual(r.status_code, 404)
1191

  
1192
    def test_update_from_other_version(self):
1193
        versions = []
1194
        info = self.get_object_info(self.container, self.object)
1195
        versions.append(info['X-Object-Version'])
1196
        pre_length = int(info['Content-Length'])
1197

  
1198
        # update object
1199
        d1, r = self.upload_object(self.container, self.object,
1200
                                   length=pre_length - 1)[1:]
1201
        self.assertTrue('X-Object-Version' in r)
1202
        versions.append(r['X-Object-Version'])
1203

  
1204
        # update object
1205
        d2, r = self.upload_object(self.container, self.object,
1206
                                   length=pre_length - 2)[1:]
1207
        self.assertTrue('X-Object-Version' in r)
1208
        versions.append(r['X-Object-Version'])
1209

  
1210
        # get previous version
1211
        url = join_urls(self.pithos_path, self.user, self.container,
1212
                        self.object)
1213
        r = self.get('%s?version=list&format=json' % url)
1214
        self.assertEqual(r.status_code, 200)
1215
        l = json.loads(r.content)['versions']
1216
        self.assertEqual(len(l), 3)
1217
        self.assertEqual([str(v[0]) for v in l], versions)
1218

  
1219
        # update with the previous version
1220
        r = self.post(url,
1221
                      HTTP_CONTENT_RANGE='bytes 0-/*',
1222
                      HTTP_X_SOURCE_OBJECT='/%s/%s' % (self.container,
1223
                                                       self.object),
1224
                      HTTP_X_SOURCE_VERSION=versions[0])
1225
        self.assertEqual(r.status_code, 204)
1226

  
1227
        # check content
1228
        r = self.get(url)
1229
        content = r.content
1230
        self.assertEqual(len(content), pre_length)
1231
        self.assertEqual(content, self.object_data)
1232

  
1233
        # update object
1234
        d3, r = self.upload_object(self.container, self.object,
1235
                                   length=len(d2) + 1)[1:]
1236
        self.assertTrue('X-Object-Version' in r)
1237
        versions.append(r['X-Object-Version'])
1238

  
1239
        # update with the previous version
1240
        r = self.post(url,
1241
                      HTTP_CONTENT_RANGE='bytes 0-/*',
1242
                      HTTP_X_SOURCE_OBJECT='/%s/%s' % (self.container,
1243
                                                       self.object),
1244
                      HTTP_X_SOURCE_VERSION=versions[-2])
1245
        self.assertEqual(r.status_code, 204)
1246

  
1247
        # check content
1248
        r = self.get(url)
1249
        content = r.content
1250
        self.assertEqual(content, d2 + d3[-1])
1251

  
1163 1252

  
1164 1253
class ObjectDelete(PithosAPITest):
1165 1254
    def setUp(self):

Also available in: Unified diff