Add validated Drive.delete_many collection deletion

ci/run_offline.py

@@ -18,5 +18,5 @@ def deny(event,args):
 sys.path.insert(0,str(ROOT))
 suite=unittest.defaultTestLoader.discover(str(ROOT/'tests_offline'))
 r=unittest.TextTestRunner(verbosity=2).run(suite)
-assert r.testsRun==255, 'unexpected test count'
+assert r.testsRun==263, 'unexpected test count'
 sys.exit(not r.wasSuccessful())

src/cloudsigma/resource.py

@@ -168,6 +168,28 @@ def clone(self, uuid, data=None):
 class Drive(ResourceBase):
     resource_name = 'drives'
 
+    def delete_many(self, uuids):
+        """Request ordinary drive collection deletion in one HTTP request.
+
+        Accept a nonempty list/tuple of nonempty ASCII letters, digits, hyphens
+        and underscores. This conservative SDK identifier policy is not public
+        UUID grammar. Validate the entire batch before sending; preserve order
+        and duplicates without changing the caller's sequence. HTTP 204 returns
+        None and acknowledges deletion started, not completion or atomicity.
+        No polling, retries, remote snapshots or snapshot deletion.
+        https://docs.cloudsigma.com/en/latest/drives.html#deleting
+        """
+        if not isinstance(uuids, (list, tuple)) or not uuids:
+            raise ValueError('drive identifiers must be a nonempty list or tuple')
+        objects = []
+        for uuid in uuids:
+            if not isinstance(uuid, str) or not uuid or any(
+                    char not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_'
+                    for char in uuid):
+                raise ValueError('invalid drive identifier')
+            objects.append({'uuid': uuid})
+        return self.c.delete(self._get_url(), data={'objects': objects})
+
     def set_scheduler(self, uuid, backup_policy_uuid):
         """Link an existing backup scheduler to a drive.
 

tests_offline/test_drive_delete_many.py

@@ -0,0 +1,126 @@
+"""Ordinary drive collection deletion acknowledgment, never completion."""
+import json
+import unittest
+import requests
+from cloudsigma.resource import Drive
+from cloudsigma import errors
+
+IDS = ['b137e217-42b6-4ecf-8575-d72efc2d3dbd',
+       'e035a488-8587-4a15-ab25-9b7343236bc9',
+       'feded33c-106f-49fa-a1c4-be5c718ad1b5']
+
+
+class Adapter(requests.adapters.BaseAdapter):
+    def __init__(self, status=204, failure=None):
+        self.seen = []
+        self.status = status
+        self.failure = failure
+
+    def send(self, request, **kwargs):
+        self.seen.append((request, kwargs))
+        if self.failure:
+            raise self.failure
+        response = requests.Response()
+        response.status_code = self.status
+        response._content = b'' if self.status == 204 else b'[{"error_type":"fixture"}]'
+        response.headers['X-REQUEST-ID'] = 'synthetic-request'
+        response.request = request
+        return response
+
+    def close(self):
+        pass
+
+
+class DriveDeleteManyTests(unittest.TestCase):
+    def setup_client(self, session, status=204, failure=None, cls=Drive):
+        adapter = Adapter(status, failure)
+        session.mount('https://', adapter)
+        resource = cls(api_endpoint='https://fixture.invalid/api/2.0/',
+                       username='fixture', password='synthetic', session=session)
+        return resource, adapter
+
+    def test_documented_collection_request(self):
+        for cls in (Drive,):
+            with self.subTest(cls=cls), requests.Session() as session:
+                resource, adapter = self.setup_client(session, cls=cls)
+                ids = list(IDS)
+                self.assertIsNone(resource.delete_many(ids))
+                self.assertEqual(ids, IDS)
+                self.assertEqual(len(adapter.seen), 1)
+                request, options = adapter.seen[0]
+                self.assertEqual(request.method, 'DELETE')
+                self.assertEqual(request.url, 'https://fixture.invalid/api/2.0/drives/')
+                self.assertEqual(json.loads(request.body), {'objects': [
+                    {'uuid': IDS[0]}, {'uuid': IDS[1]}, {'uuid': IDS[2]}]})
+                self.assertEqual(request.headers['Content-Type'], 'application/json')
+                self.assertIn('Authorization', request.headers)
+                self.assertEqual(options['timeout'], 15)
+
+    def test_order_and_duplicates(self):
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session)
+            resource.delete_many((IDS[2], IDS[0], IDS[2]))
+            self.assertEqual(json.loads(adapter.seen[0][0].body), {
+                'objects': [{'uuid': IDS[2]}, {'uuid': IDS[0]}, {'uuid': IDS[2]}]})
+
+    def test_invalid_input_no_dispatch(self):
+        invalid = [None, [], (), IDS[0], {}, iter(IDS), [IDS[0], None],
+                   [IDS[0], ''], [IDS[0], '../x'], [IDS[0], 'a/b'],
+                   [IDS[0], 'a?b'], [IDS[0], 'a#b'], [IDS[0], '%2f'],
+                   [IDS[0], 'x\n'], [IDS[0], b'bytes']]
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session)
+            for value in invalid:
+                with self.subTest(value=value), self.assertRaises(ValueError):
+                    resource.delete_many(value)
+            self.assertEqual(adapter.seen, [])
+
+    def test_http_errors(self):
+        for status, error in [(403, errors.PermissionError), (404, errors.ClientError),
+                              (500, errors.ServerError), (302, errors.ApiClientError)]:
+            with self.subTest(status=status), requests.Session() as session:
+                resource, adapter = self.setup_client(session, status)
+                with self.assertRaises(error) as caught:
+                    resource.delete_many(IDS)
+                self.assertEqual(caught.exception.status_code, status)
+                self.assertEqual(len(adapter.seen), 1)
+
+    def test_transport_error(self):
+        failure = requests.exceptions.Timeout('synthetic timeout')
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session, failure=failure)
+            with self.assertRaises(requests.exceptions.Timeout) as caught:
+                resource.delete_many(IDS)
+            self.assertIs(caught.exception, failure)
+            self.assertEqual(len(adapter.seen), 1)
+
+    def test_legacy_singleton_has_no_body(self):
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session)
+            self.assertIsNone(resource.delete(IDS[0]))
+            request = adapter.seen[0][0]
+            self.assertIsNone(request.body)
+            self.assertEqual(request.url, 'https://fixture.invalid/api/2.0/drives/' + IDS[0] + '/')
+
+    def test_policy_boundary_and_nonmutation(self):
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session)
+            ids = ['Az09-_']
+            resource.delete_many(ids)
+            self.assertEqual(ids, ['Az09-_'])
+            self.assertEqual(json.loads(adapter.seen[0][0].body),
+                             {'objects': [{'uuid': 'Az09-_'}]})
+            for invalid in ([True], [1], ['é'], ['a b'], ['\\']):
+                with self.subTest(value=invalid), self.assertRaises(ValueError):
+                    resource.delete_many(invalid)
+            self.assertEqual(len(adapter.seen), 1)
+
+    def test_response_hook_and_no_retry(self):
+        with requests.Session() as session:
+            resource, adapter = self.setup_client(session)
+            seen = []
+            resource.attach_response_hook(lambda response, **kwargs: seen.append(response))
+            self.assertIsNone(resource.delete_many(IDS))
+            self.assertEqual(len(seen), 1)
+            self.assertIs(seen[0], resource.c.resp)
+            self.assertEqual(len(adapter.seen), 1)