Release 5.49.0

linode_api4/groups/monitor.py

@@ -22,6 +22,7 @@
     AkamaiObjectStorageLogsDestinationDetails,
     CustomHTTPSLogsDestinationDetails,
     LogsStreamDetails,
+    LogsStreamQuota,
 )
 
 __all__ = [
@@ -625,3 +626,18 @@ def stream_create(
             )
 
         return LogsStream(self.client, result["id"], result)
+
+    def stream_quotas(self) -> PaginatedList:
+        """
+        Retrieve quota definitions available for the authenticated account's streams.
+
+        This reads from ``/monitor/streams/quotas`` on the API v4 endpoint.
+
+        API Documentation: TODO
+
+        :returns: A paginated list of stream quota definitions.
+        :rtype: PaginatedList[LogsStreamQuota]
+        """
+        return self.client._get_objects(
+            "/monitor/streams/quotas", LogsStreamQuota
+        )

linode_api4/objects/monitor.py

@@ -40,6 +40,7 @@
     "LogsStreamStatus",
     "LogsStreamDetails",
     "LogsStreamDestination",
+    "LogsStreamQuota",
 ]
 
 
@@ -276,8 +277,8 @@ class MonitorDashboard(Base):
         "id": Property(identifier=True),
         "created": Property(is_datetime=True),
         "label": Property(),
-        "service_type": Property(ServiceType),
-        "type": Property(DashboardType),
+        "service_type": Property(),
+        "type": Property(),
         "group_by": Property(),
         "widgets": Property(json_object=DashboardWidget),
         "updated": Property(is_datetime=True),
@@ -294,7 +295,7 @@ class MonitorService(Base):
     api_endpoint = "/monitor/services/{service_type}"
     id_attribute = "service_type"
     properties = {
-        "service_type": Property(ServiceType),
+        "service_type": Property(),
         "label": Property(),
         "alert": Property(json_object=ServiceAlert),
     }
@@ -495,7 +496,7 @@ class AlertDefinition(DerivedBase):
         "entity_ids": Property(mutable=True),
         "description": Property(mutable=True),
         "service_class": Property(alias_of="class"),
-        "scope": Property(AlertScope),
+        "scope": Property(),
         "regions": Property(mutable=True),
         "entities": Property(json_object=AlertEntities),
         "channel_ids": Property(mutable=True),
@@ -904,3 +905,16 @@ def history(self):
             "{}/history".format(LogsStream.api_endpoint.format(id=self.id)),
             LogsStreamHistory,
         )
+
+
+@dataclass
+class LogsStreamQuota(JSONObject):
+    """
+    Represents a quota definition for logs streams.
+    """
+
+    quota_id: str = ""
+    quota_name: str = ""
+    quota_limit: int = 0
+    quota_type: str = ""
+    description: str = ""

linode_api4/objects/region.py

@@ -68,6 +68,7 @@ class Capability(StrEnum):
     prefixlists = "Cloud Firewall Prefix Lists"
     current_prefixlists = "Cloud Firewall Prefix List Current References"
     gpudirect_rdma = "GPUDirect RDMA"
+    premium_nodebalancers = "Premium NodeBalancers"
 
 
 @dataclass

pyproject.toml

@@ -46,6 +46,7 @@ unit-test = [
 ]
 int-test = [
     "linode_api4[test-shared]",
+    "cryptography",
 ]
 lint = [
     "black>=23.1.0",

test/fixtures/monitor_streams_quotas.json

@@ -0,0 +1,21 @@
+{
+  "data": [
+    {
+      "quota_id": "aclp_audit_logs_streams",
+      "quota_name": "Number of Audit Logs Streams",
+      "description": "Current number of audit logs streams per account",
+      "quota_limit": 5,
+      "quota_type": "logs_streams_aclp_audit_logs_streams"
+    },
+    {
+      "quota_id": "aclp_lke_audit_logs_streams",
+      "quota_name": "Number of LKE Audit Logs Streams",
+      "description": "Current number of LKE audit logs streams per account",
+      "quota_limit": 10,
+      "quota_type": "logs_streams_aclp_lke_audit_logs_streams"
+    }
+  ],
+  "results": 2,
+  "page": 1,
+  "pages": 1
+}

test/integration/conftest.py

@@ -2,7 +2,6 @@
 import logging
 import os
 import random
-import subprocess
 import time
 from test.integration.helpers import (
     get_test_label,
@@ -14,6 +13,8 @@
 
 import pytest
 import requests
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ed25519
 from requests.exceptions import ConnectionError, RequestException
 
 from linode_api4 import (
@@ -32,7 +33,7 @@
 )
 from linode_api4.errors import ApiError
 from linode_api4.linode_client import LinodeClient, MonitorClient
-from linode_api4.objects import Region
+from linode_api4.objects import Capability, Region
 
 ENV_TOKEN_NAME = "LINODE_TOKEN"
 ENV_API_URL_NAME = "LINODE_API_URL"
@@ -42,10 +43,10 @@
 SKIP_E2E_FIREWALL = "SKIP_E2E_FIREWALL"
 
 ALL_ACCOUNT_AVAILABILITIES = {
-    "Linodes",
-    "NodeBalancers",
-    "Block Storage",
-    "Kubernetes",
+    Capability.linodes,
+    Capability.nodebalancers,
+    Capability.block_storage,
+    Capability.lke,
 }
 
 logger = logging.getLogger(__name__)
@@ -61,7 +62,7 @@ def get_api_url():
 
 def get_regions(
     client: LinodeClient,
-    capabilities: Optional[Set[str]] = None,
+    capabilities: Optional[Set[Capability]] = None,
     site_type: Optional[str] = None,
 ):
     region_override = os.environ.get(ENV_REGION_OVERRIDE)
@@ -115,7 +116,9 @@ def get_regions(
 
 
 def get_region(
-    client: LinodeClient, capabilities: Set[str] = None, site_type: str = "core"
+    client: LinodeClient,
+    capabilities: Optional[Set[Capability]] = None,
+    site_type: str = "core",
 ):
     return random.choice(get_regions(client, capabilities, site_type))
 
@@ -223,7 +226,11 @@ def create_inbound_rule(ipv4_address, ipv6_address):
 def create_linode(test_linode_client, e2e_test_firewall):
     client = test_linode_client
 
-    region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core")
+    region = get_region(
+        client,
+        {Capability.linodes, Capability.firewall},
+        site_type="core",
+    )
     label = get_test_label(length=8)
 
     linode_instance = client.linode.instance_create(
@@ -244,7 +251,11 @@ def create_linode(test_linode_client, e2e_test_firewall):
 def create_linode_for_pass_reset(test_linode_client, e2e_test_firewall):
     client = test_linode_client
 
-    region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core")
+    region = get_region(
+        client,
+        {Capability.linodes, Capability.firewall},
+        site_type="core",
+    )
     label = get_test_label(length=8)
     password = "aComplex@Password123"
 
@@ -263,17 +274,28 @@ def create_linode_for_pass_reset(test_linode_client, e2e_test_firewall):
 
 
 @pytest.fixture(scope="session")
-def ssh_key_gen(tmp_path_factory):
-    key_path = tmp_path_factory.mktemp("ssh-key-gen") / "sdk-sshkey"
-
-    subprocess.run(
-        ["ssh-keygen", "-q", "-t", "rsa", "-f", str(key_path), "-N", ""],
-        check=True,
+def ssh_key_gen():
+    key = ed25519.Ed25519PrivateKey.generate()
+
+    pub_key = (
+        key.public_key()
+        .public_bytes(
+            encoding=serialization.Encoding.OpenSSH,
+            format=serialization.PublicFormat.OpenSSH,
+        )
+        .decode()
+        .rstrip()
+    )
+    priv_key = (
+        key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.OpenSSH,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        .decode()
+        .rstrip()
     )
 
-    pub_key = key_path.with_suffix(".pub").read_text().rstrip()
-    priv_key = key_path.read_text().rstrip()
-
     yield pub_key, priv_key
 
 
@@ -337,7 +359,11 @@ def get_domain_status():
 @pytest.fixture(scope="session")
 def test_volume(test_linode_client):
     client = test_linode_client
-    region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core")
+    region = get_region(
+        client,
+        {Capability.linodes, Capability.firewall},
+        site_type="core",
+    )
     label = get_test_label(length=8)
 
     volume = client.volume_create(label=label, region=region)
@@ -356,7 +382,7 @@ def get_volume_status():
 @pytest.fixture(scope="session")
 def test_volume_with_encryption(test_linode_client):
     client = test_linode_client
-    region = get_region(client, {"Block Storage Encryption"})
+    region = get_region(client, {Capability.blockstorage_encryption})
     label = get_test_label(length=8)
 
     volume = client.volume_create(
@@ -388,7 +414,8 @@ def test_nodebalancer(test_linode_client):
     label = get_test_label(length=8)
 
     nodebalancer = client.nodebalancer_create(
-        region=get_region(client, capabilities={"NodeBalancers"}), label=label
+        region=get_region(client, capabilities={Capability.nodebalancers}),
+        label=label,
     )
 
     yield nodebalancer
@@ -475,11 +502,11 @@ def create_vpc(test_linode_client):
         region=get_region(
             test_linode_client,
             {
-                "VPCs",
-                "VPC IPv6 Stack",
-                "Linode Interfaces",
-                "Custom VPC IPv4 Ranges",
-                "Linodes",
+                Capability.vpcs,
+                Capability.vpc_ipv6_stack,
+                Capability.linode_interfaces,
+                Capability.vpc_custom_ipv4_ranges,
+                Capability.linodes,
             },
         ),
         description="test description",
@@ -497,7 +524,12 @@ def create_vpc_with_rdma_type(test_linode_client):
 
     # GPUDirect RDMA capability not available for now
     region = get_region(
-        test_linode_client, {"VPCs", "VPC IPv6 Stack", "Linode Interfaces"}
+        test_linode_client,
+        {
+            Capability.vpcs,
+            Capability.vpc_ipv6_stack,
+            Capability.linode_interfaces,
+        },
     )
 
     vpc = client.vpcs.create(
@@ -567,7 +599,10 @@ def create_vpc_with_ipv4(test_linode_client):
 
     vpc = client.vpcs.create(
         label=get_test_label(length=10),
-        region=get_region(client, {"VPCs", "Custom VPC IPv4 Ranges"}),
+        region=get_region(
+            client,
+            {Capability.vpcs, Capability.vpc_custom_ipv4_ranges},
+        ),
         description="integration test vpc with ipv4",
         ipv4=[{"range": "10.0.0.0/8"}],
     )
@@ -587,13 +622,13 @@ def create_multiple_vpcs(test_linode_client):
 
     vpc_1 = client.vpcs.create(
         label,
-        get_region(test_linode_client, {"VPCs"}),
+        get_region(test_linode_client, {Capability.vpcs}),
         description="test description",
     )
 
     vpc_2 = client.vpcs.create(
         label_2,
-        get_region(test_linode_client, {"VPCs"}),
+        get_region(test_linode_client, {Capability.vpcs}),
         description="test description",
     )
 
@@ -612,7 +647,7 @@ def create_placement_group(test_linode_client):
 
     pg = client.placement.group_create(
         label,
-        get_region(test_linode_client, {"Placement Group"}),
+        get_region(test_linode_client, {Capability.placement_group}),
         PlacementGroupType.anti_affinity_local,
         PlacementGroupPolicy.flexible,
     )
@@ -652,7 +687,11 @@ def pytest_configure(config):
 @pytest.fixture(scope="session")
 def linode_for_vlan_tests(test_linode_client, e2e_test_firewall):
     client = test_linode_client
-    region = get_region(client, {"Linodes", "Vlans"}, site_type="core")
+    region = get_region(
+        client,
+        {Capability.linodes, Capability.vlans},
+        site_type="core",
+    )
     label = get_test_label(length=8)
 
     linode_instance = client.linode.instance_create(
@@ -811,7 +850,11 @@ def test_monitor_client(get_monitor_token_for_db_entities):
 @pytest.fixture
 def create_reserved_ip(test_linode_client):
     client = test_linode_client
-    region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core")
+    region = get_region(
+        client,
+        {Capability.linodes, Capability.firewall},
+        site_type="core",
+    )
     reserved_ip = client.networking.reserved_ip_create(
         region=region, tags=["test"]
     )

test/integration/filters/fixtures.py

@@ -3,6 +3,8 @@
 
 import pytest
 
+from linode_api4.objects import Capability
+
 
 @pytest.fixture(scope="package")
 def domain_instance(test_linode_client):
@@ -23,7 +25,10 @@ def lke_cluster_instance(test_linode_client):
     node_type = test_linode_client.linode.types()[1]  # g6-standard-1
     version = test_linode_client.lke.versions()[0]
 
-    region = get_region(test_linode_client, {"Kubernetes", "Disk Encryption"})
+    region = get_region(
+        test_linode_client,
+        {Capability.lke, Capability.disk_encryption},
+    )
 
     node_pool = test_linode_client.lke.node_pool(node_type, 3)
     label = get_test_label() + "_cluster"