From e32641e135fd3ee9fb896e016d0872d3d47f7762 Mon Sep 17 00:00:00 2001 From: cloudsigma Date: Wed, 21 Oct 2026 16:59:40 +0000 Subject: [PATCH] Add validated Drive.delete_many collection deletion --- ci/run_offline.py | 2 +- src/cloudsigma/resource.py | 22 +++++ tests_offline/test_drive_delete_many.py | 126 ++++++++++++++++++++++++ 3 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 tests_offline/test_drive_delete_many.py diff --git a/ci/run_offline.py b/ci/run_offline.py index b770ae8..61a2e02 100644 --- a/ci/run_offline.py +++ b/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()) diff --git a/src/cloudsigma/resource.py b/src/cloudsigma/resource.py index 25db90e..6857414 100644 --- a/src/cloudsigma/resource.py +++ b/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. diff --git a/tests_offline/test_drive_delete_many.py b/tests_offline/test_drive_delete_many.py new file mode 100644 index 0000000..9954ab8 --- /dev/null +++ b/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)