diff --git a/ci/run_offline.py b/ci/run_offline.py index 1c29b00..b33053d 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==222, 'unexpected test count' +assert r.testsRun==228, 'unexpected test count' sys.exit(not r.wasSuccessful()) diff --git a/src/cloudsigma/resource.py b/src/cloudsigma/resource.py index 8d42147..25db90e 100644 --- a/src/cloudsigma/resource.py +++ b/src/cloudsigma/resource.py @@ -828,6 +828,21 @@ class HostAllocationPools(ResourceBase): class DriveUsers(ResourceBase): resource_name = 'driveusers' + def create(self, data, query_params=None): + """Send plain definitions as the documented singleton RDS request. + + Preserve explicit objects wrappers and list/tuple inputs through the + legacy base implementation for compatibility; those alternative wire + shapes are not established public contracts. Do not coerce fields or + infer count bounds. Query forwarding and response unwrapping remain + unchanged. Other resources retain ResourceBase.create behavior. + https://docs.cloudsigma.com/en/latest/drives.html#creating-defining-rds-drive + """ + if isinstance(data, dict) and 'objects' not in data: + return self.c.post(self._get_url(), data, return_list=False, + query_params=query_params or {}) + return super(DriveUsers, self).create(data, query_params=query_params) + def update_amount(self, drive_uuid, amount): """Update an RDS drive's user count with an amount-only definition. diff --git a/tests_offline/test_drive_users_create.py b/tests_offline/test_drive_users_create.py new file mode 100644 index 0000000..fd1bec9 --- /dev/null +++ b/tests_offline/test_drive_users_create.py @@ -0,0 +1,98 @@ +"""DriveUsers singleton correction and explicit legacy shape compatibility.""" +import copy +import json +import unittest +import requests +from cloudsigma.resource import DriveUsers, Drive, Server +from cloudsigma import errors +from test_drive_users_update_amount import Adapter, UUID, RESULT + + +class DriveUsersCreateTests(unittest.TestCase): + def client(self, session, status=201, failure=None, cls=DriveUsers): + adapter = Adapter(status, failure) + session.mount('https://', adapter) + return cls(api_endpoint='https://fixture.invalid/api/2.0/', + username='fixture', password='synthetic', session=session), adapter + + def test_documented_singleton(self): + with requests.Session() as session: + resource, adapter = self.client(session) + data = {'drive': UUID, 'amount': 5} + self.assertEqual(resource.create(data), RESULT) + self.assertEqual(data, {'drive': UUID, 'amount': 5}) + self.assertEqual(len(adapter.seen), 1) + req, options = adapter.seen[0] + self.assertEqual(req.method, 'POST') + self.assertEqual(req.url, 'https://fixture.invalid/api/2.0/driveusers/') + self.assertEqual(json.loads(req.body), {'drive': UUID, 'amount': 5}) + self.assertEqual(req.headers['Content-Type'], 'application/json') + self.assertIn('Authorization', req.headers) + self.assertEqual(options['timeout'], 15) + + def test_presence_and_query(self): + for amount in (0, False, -1, 9007199254740993, None): + with self.subTest(amount=amount), requests.Session() as session: + resource, adapter = self.client(session) + data = {'drive': UUID, 'amount': amount, 'extension': {'enabled': False, 'empty': []}} + before = copy.deepcopy(data) + query = {'synthetic': 'a b'} + resource.create(data, query) + req = adapter.seen[0][0] + self.assertEqual(json.loads(req.body), before) + self.assertIs(type(json.loads(req.body)['amount']), type(amount)) + self.assertEqual(req.url, 'https://fixture.invalid/api/2.0/driveusers/?synthetic=a+b') + self.assertEqual(data, before) + self.assertEqual(query, {'synthetic': 'a b'}) + + def test_legacy_shapes(self): + item = {'drive': UUID, 'amount': 5} + for data in ([item], (item,), [], (), {'objects': [item], 'extension': False}, {'objects': []}): + with self.subTest(data=data), requests.Session() as session: + resource, adapter = self.client(session) + before = copy.deepcopy(data) + self.assertEqual(resource.create(data, {'legacy': 0}), RESULT) + expected = data if isinstance(data, dict) else {'objects': list(data)} + self.assertEqual(json.loads(adapter.seen[0][0].body), expected) + self.assertEqual(data, before) + self.assertTrue(adapter.seen[0][0].url.endswith('?legacy=0')) + with requests.Session() as session: + resource, adapter = self.client(session) + for data in (None, False, 1, 'bad'): + with self.assertRaises(TypeError): + resource.create(data) + self.assertEqual(adapter.seen, []) + + def test_unrelated_creates_and_updates(self): + for cls in (Drive, Server): + with self.subTest(cls=cls), requests.Session() as session: + resource, adapter = self.client(session, cls=cls) + resource.create({'name': 'fixture'}) + self.assertEqual(json.loads(adapter.seen[0][0].body), {'objects': [{'name': 'fixture'}]}) + with requests.Session() as session: + resource, adapter = self.client(session, status=200) + resource.update(UUID, {'amount': 0, 'extension': False}) + resource.update_amount(UUID, 0) + self.assertEqual([json.loads(r.body) for r, _ in adapter.seen], + [{'amount': 0, 'extension': False}, {'amount': 0}]) + self.assertEqual([r.method for r, _ in adapter.seen], ['PUT', 'PUT']) + + def test_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.client(session, status=status) + with self.assertRaises(error) as caught: + resource.create({'drive': UUID, 'amount': 5}) + self.assertEqual(caught.exception.status_code, status) + self.assertEqual(caught.exception.request_id, 'synthetic-request') + 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.client(session, failure=failure) + with self.assertRaises(requests.exceptions.Timeout) as caught: + resource.create({'drive': UUID, 'amount': 5}) + self.assertIs(caught.exception, failure) + self.assertEqual(len(adapter.seen), 1)