Add global quotas and quota usage support for OBJ services

linode_api4/groups/object_storage.py

@@ -19,6 +19,7 @@
     ObjectStorageACL,
     ObjectStorageBucket,
     ObjectStorageCluster,
+    ObjectStorageGlobalQuota,
     ObjectStorageKeyPermission,
     ObjectStorageKeys,
     ObjectStorageQuota,
@@ -533,3 +534,18 @@ def quotas(self, *filters):
         :rtype: PaginatedList of ObjectStorageQuota
         """
         return self.client._get_and_filter(ObjectStorageQuota, *filters)
+
+    def global_quotas(self, *filters):
+        """
+        Lists the active account-level Object Storage quotas applied to your account.
+
+        API Documentation: TBD
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+
+        :returns: A list of account-level Object Storage Quotas that matched the query.
+        :rtype: PaginatedList of ObjectStorageGlobalQuota
+        """
+        return self.client._get_and_filter(ObjectStorageGlobalQuota, *filters)

linode_api4/objects/object_storage.py

@@ -596,6 +596,8 @@ class ObjectStorageQuota(Base):
         "description": Property(),
         "quota_limit": Property(),
         "resource_metric": Property(),
+        "quota_type": Property(),
+        "has_usage": Property(),
     }
 
     def usage(self):
@@ -614,3 +616,41 @@ def usage(self):
         )
 
         return ObjectStorageQuotaUsage.from_json(result)
+
+
+class ObjectStorageGlobalQuota(Base):
+    """
+    An account-level Object Storage quota.
+
+    API documentation: TBD
+    """
+
+    api_endpoint = "/object-storage/global-quotas/{quota_id}"
+    id_attribute = "quota_id"
+
+    properties = {
+        "quota_id": Property(identifier=True),
+        "quota_type": Property(),
+        "quota_name": Property(),
+        "description": Property(),
+        "resource_metric": Property(),
+        "quota_limit": Property(),
+        "has_usage": Property(),
+    }
+
+    def usage(self):
+        """
+        Gets usage data for a specific account-level Object Storage quota.
+
+        API documentation: https://techdocs.akamai.com/linode-api/reference/get-object-storage-global-quota-usage
+
+        :returns: The Object Storage Global Quota usage.
+        :rtype: ObjectStorageQuotaUsage
+        """
+
+        result = self._client.get(
+            f"{type(self).api_endpoint}/usage",
+            model=self,
+        )
+
+        return ObjectStorageQuotaUsage.from_json(result)

test/fixtures/object-storage_global-quotas.json

@@ -0,0 +1,25 @@
+{
+  "data": [
+    {
+      "quota_id": "obj-access-keys-per-account",
+      "quota_type": "obj-access-keys",
+      "quota_name": "Object Storage Access Keys per Account",
+      "description": "Maximum number of access keys this customer is allowed to have on their account.",
+      "resource_metric": "access_key",
+      "quota_limit": 100,
+      "has_usage": true
+    },
+    {
+      "quota_id": "obj-total-capacity-per-account",
+      "quota_type": "obj-total-capacity",
+      "quota_name": "Object Storage Total Capacity per Account",
+      "description": "Maximum total storage capacity in bytes this customer is allowed on their account.",
+      "resource_metric": "byte",
+      "quota_limit": 1099511627776,
+      "has_usage": true
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 2
+}

test/fixtures/object-storage_global-quotas_obj-access-keys-per-account.json

@@ -0,0 +1,9 @@
+{
+  "quota_id": "obj-access-keys-per-account",
+  "quota_type": "obj-access-keys",
+  "quota_name": "Object Storage Access Keys per Account",
+  "description": "Maximum number of access keys this customer is allowed to have on their account.",
+  "resource_metric": "access_key",
+  "quota_limit": 100,
+  "has_usage": true
+}

test/fixtures/object-storage_global-quotas_obj-access-keys-per-account_usage.json

@@ -0,0 +1,4 @@
+{
+  "quota_limit": 100,
+  "usage": 25
+}

test/fixtures/object-storage_quotas.json

@@ -7,7 +7,9 @@
       "endpoint_type": "E1",
       "s3_endpoint": "us-iad-1.linodeobjects.com",
       "quota_limit": 50,
-      "resource_metric": "object"
+      "resource_metric": "object",
+      "quota_type": "obj-objects",
+      "has_usage": true
     },
     {
       "quota_id": "obj-bucket-us-ord-1",
@@ -16,7 +18,9 @@
       "endpoint_type": "E1",
       "s3_endpoint": "us-iad-1.linodeobjects.com",
       "quota_limit": 50,
-      "resource_metric": "bucket"
+      "resource_metric": "bucket",
+      "quota_type": "obj-bucket",
+      "has_usage": true
     }
   ],
   "page": 1,

test/fixtures/object-storage_quotas_obj-objects-us-ord-1.json

@@ -5,5 +5,7 @@
   "endpoint_type": "E1",
   "s3_endpoint": "us-iad-1.linodeobjects.com",
   "quota_limit": 50,
-  "resource_metric": "object"
+  "resource_metric": "object",
+  "quota_type": "obj-objects",
+  "has_usage": true
 }

test/integration/models/object_storage/test_obj_quotas.py

@@ -1,6 +1,8 @@
 import pytest
 
+from linode_api4.errors import ApiError
 from linode_api4.objects.object_storage import (
+    ObjectStorageGlobalQuota,
     ObjectStorageQuota,
     ObjectStorageQuotaUsage,
 )
@@ -25,6 +27,8 @@ def test_list_and_get_obj_storage_quotas(test_linode_client):
     assert found_quota.description == get_quota.description
     assert found_quota.quota_limit == get_quota.quota_limit
     assert found_quota.resource_metric == get_quota.resource_metric
+    assert found_quota.quota_type == get_quota.quota_type
+    assert found_quota.has_usage == get_quota.has_usage
 
 
 def test_get_obj_storage_quota_usage(test_linode_client):
@@ -33,7 +37,21 @@ def test_get_obj_storage_quota_usage(test_linode_client):
     if len(quotas) < 1:
         pytest.skip("No available quota for testing. Skipping now...")
 
-    quota_id = quotas[0].quota_id
+    quota_with_usage = next(
+        (quota for quota in quotas if quota.has_usage), None
+    )
+
+    if quota_with_usage is None:
+        quota_id = quotas[0].quota_id
+        quota = test_linode_client.load(ObjectStorageQuota, quota_id)
+
+        # quota without usage should return an API error on usage retrieval
+        with pytest.raises(ApiError):
+            quota.usage()
+
+        return
+
+    quota_id = quota_with_usage.quota_id
     quota = test_linode_client.load(ObjectStorageQuota, quota_id)
 
     quota_usage = quota.usage()
@@ -43,3 +61,56 @@ def test_get_obj_storage_quota_usage(test_linode_client):
 
     if quota_usage.usage is not None:
         assert quota_usage.usage >= 0
+
+
+def test_list_and_get_obj_storage_global_quotas(test_linode_client):
+    quotas = test_linode_client.object_storage.global_quotas()
+
+    if len(quotas) < 1:
+        pytest.skip("No available global quota for testing. Skipping now...")
+
+    found_quota = quotas[0]
+
+    get_quota = test_linode_client.load(
+        ObjectStorageGlobalQuota, found_quota.quota_id
+    )
+
+    assert found_quota.quota_id == get_quota.quota_id
+    assert found_quota.quota_type == get_quota.quota_type
+    assert found_quota.quota_name == get_quota.quota_name
+    assert found_quota.description == get_quota.description
+    assert found_quota.resource_metric == get_quota.resource_metric
+    assert found_quota.quota_limit == get_quota.quota_limit
+    assert found_quota.has_usage == get_quota.has_usage
+
+
+def test_get_obj_storage_global_quota_usage(test_linode_client):
+    quotas = test_linode_client.object_storage.global_quotas()
+
+    if len(quotas) < 1:
+        pytest.skip("No available global quota for testing. Skipping now...")
+
+    quota_with_usage = next(
+        (quota for quota in quotas if quota.has_usage), None
+    )
+
+    if quota_with_usage is None:
+        quota_id = quotas[0].quota_id
+        quota = test_linode_client.load(ObjectStorageGlobalQuota, quota_id)
+
+        # quota without usage should return an API error on usage retrieval
+        with pytest.raises(ApiError):
+            quota.usage()
+
+        return
+
+    quota_id = quota_with_usage.quota_id
+    quota = test_linode_client.load(ObjectStorageGlobalQuota, quota_id)
+
+    quota_usage = quota.usage()
+
+    assert isinstance(quota_usage, ObjectStorageQuotaUsage)
+    assert quota_usage.quota_limit >= 0
+
+    if quota_usage.usage is not None:
+        assert quota_usage.usage >= 0

test/unit/objects/object_storage_test.py

@@ -6,6 +6,7 @@
     ObjectStorageACL,
     ObjectStorageBucket,
     ObjectStorageCluster,
+    ObjectStorageGlobalQuota,
     ObjectStorageQuota,
 )
 
@@ -306,6 +307,8 @@ def test_quota_get_and_list(self):
         self.assertEqual(quota.s3_endpoint, "us-iad-1.linodeobjects.com")
         self.assertEqual(quota.quota_limit, 50)
         self.assertEqual(quota.resource_metric, "object")
+        self.assertEqual(quota.quota_type, "obj-objects")
+        self.assertTrue(quota.has_usage)
 
         quota_usage_url = "/object-storage/quotas/obj-objects-us-ord-1/usage"
         with self.mock_get(quota_usage_url) as m:
@@ -335,3 +338,59 @@ def test_quota_get_and_list(self):
             )
             self.assertEqual(quotas[0].quota_limit, 50)
             self.assertEqual(quotas[0].resource_metric, "object")
+            self.assertEqual(quotas[0].quota_type, "obj-objects")
+            self.assertTrue(quotas[0].has_usage)
+
+    def test_global_quota_get_and_list(self):
+        """
+        Test that you can get and list account-level Object Storage global quotas and usage.
+        """
+        quota = ObjectStorageGlobalQuota(
+            self.client,
+            "obj-access-keys-per-account",
+        )
+
+        self.assertIsNotNone(quota)
+        self.assertEqual(quota.quota_id, "obj-access-keys-per-account")
+        self.assertEqual(quota.quota_type, "obj-access-keys")
+        self.assertEqual(
+            quota.quota_name,
+            "Object Storage Access Keys per Account",
+        )
+        self.assertEqual(
+            quota.description,
+            "Maximum number of access keys this customer is allowed to have on their account.",
+        )
+        self.assertEqual(quota.resource_metric, "access_key")
+        self.assertEqual(quota.quota_limit, 100)
+        self.assertTrue(quota.has_usage)
+
+        usage_url = (
+            "/object-storage/global-quotas/obj-access-keys-per-account/usage"
+        )
+        with self.mock_get(usage_url) as m:
+            usage = quota.usage()
+            self.assertIsNotNone(usage)
+            self.assertEqual(m.call_url, usage_url)
+            self.assertEqual(usage.quota_limit, 100)
+            self.assertEqual(usage.usage, 25)
+
+        list_url = "/object-storage/global-quotas"
+        with self.mock_get(list_url) as m:
+            quotas = self.client.object_storage.global_quotas()
+            self.assertIsNotNone(quotas)
+            self.assertEqual(m.call_url, list_url)
+            self.assertEqual(len(quotas), 2)
+            self.assertEqual(quotas[0].quota_id, "obj-access-keys-per-account")
+            self.assertEqual(quotas[0].quota_type, "obj-access-keys")
+            self.assertEqual(
+                quotas[0].quota_name,
+                "Object Storage Access Keys per Account",
+            )
+            self.assertEqual(
+                quotas[0].description,
+                "Maximum number of access keys this customer is allowed to have on their account.",
+            )
+            self.assertEqual(quotas[0].resource_metric, "access_key")
+            self.assertEqual(quotas[0].quota_limit, 100)
+            self.assertTrue(quotas[0].has_usage)