diff --git a/linode_api4/groups/linode.py b/linode_api4/groups/linode.py index 3ec99ddf0..07254fb3d 100644 --- a/linode_api4/groups/linode.py +++ b/linode_api4/groups/linode.py @@ -18,6 +18,7 @@ from linode_api4.objects.filtering import Filter from linode_api4.objects.linode import ( Backup, + InstanceACLPAlertsOptions, InstancePlacementGroupAssignment, InterfaceGeneration, NetworkInterface, @@ -167,6 +168,9 @@ def instance_create( boot_size: Optional[int] = None, authorized_users: Optional[List[str]] = None, ipv4: Optional[List[str]] = None, + alerts: Optional[ + Union[Dict[str, Any], InstanceACLPAlertsOptions] + ] = None, **kwargs, ): """ @@ -359,6 +363,9 @@ def instance_create( :param ipv4: A list of reserved IPv4 addresses to assign to this Instance. NOTE: Reserved IP feature may not currently be available to all users. :type ipv4: list[str] + :param alerts: ACLP alerts available to define during instance creation. + This is v4beta only and may not be available to all users. + :type alerts: dict[str, Any] or InstanceACLPAlertsOptions :returns: A new Instance object :rtype: Instance @@ -385,6 +392,7 @@ def instance_create( "image": image, "root_pass": root_pass, "authorized_keys": load_and_validate_keys(authorized_keys), + "alerts": alerts, "authorized_users": authorized_users, # These will automatically be flattened below "firewall_id": firewall, diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 46f37561f..f3c29e36a 100644 --- a/linode_api4/groups/monitor.py +++ b/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 + ) diff --git a/linode_api4/objects/linode.py b/linode_api4/objects/linode.py index f09e8ce34..78bc0710a 100644 --- a/linode_api4/objects/linode.py +++ b/linode_api4/objects/linode.py @@ -790,6 +790,31 @@ class UpgradeInterfacesResult(JSONObject): ) +@dataclass +class InstanceAlerts(JSONObject): + """ + Represents both legacy and ACLP alerts for a Linode Instance. + """ + + cpu: int = 0 + io: int = 0 + network_in: int = 0 + network_out: int = 0 + transfer_quota: int = 0 + system_alerts: Optional[List[int]] = None + user_alerts: Optional[List[int]] = None + + +@dataclass +class InstanceACLPAlertsOptions(JSONObject): + """ + Represents the ACLP alerts available to define during instance creation and cloning. + """ + + system_alerts: Optional[List[int]] = None + user_alerts: Optional[List[int]] = None + + @dataclass class InstanceBackupSchedule(JSONObject): """ @@ -842,7 +867,7 @@ class Instance(Base): "created": Property(is_datetime=True), "updated": Property(volatile=True, is_datetime=True), "region": Property(slug_relationship=Region), - "alerts": Property(mutable=True), + "alerts": Property(mutable=True, json_object=InstanceAlerts), "image": Property(slug_relationship=Image), "disks": Property(derived_class=Disk), "configs": Property(derived_class=Config), @@ -1949,6 +1974,9 @@ def clone( Dict[str, Any], int, ] = None, + alerts: Optional[ + Union[Dict[str, Any], InstanceACLPAlertsOptions] + ] = None, ): """ Clones this linode into a new linode or into a new linode in the given region @@ -1987,6 +2015,10 @@ def clone( :param placement_group: Information about the placement group to create this instance under. :type placement_group: Union[InstancePlacementGroupAssignment, PlacementGroup, Dict[str, Any], int] + :param alerts: ACLP monitor alert definitions associated with the cloned Instance. + This is under v4beta and may not be available to all users. + :type alerts: dict[str, Any] or InstanceACLPAlertsOptions + :returns: The cloned Instance. :rtype: Instance """ @@ -2017,6 +2049,7 @@ def clone( "label": label, "group": group, "with_backups": with_backups, + "alerts": alerts, "placement_group": _expand_placement_group_assignment( placement_group ), diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index c23e4cead..2a42b3eca 100644 --- a/linode_api4/objects/monitor.py +++ b/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 = "" diff --git a/linode_api4/objects/region.py b/linode_api4/objects/region.py index ebf64985a..1334f41e1 100644 --- a/linode_api4/objects/region.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 2e0e1ed41..e4e641aa7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,7 @@ unit-test = [ ] int-test = [ "linode_api4[test-shared]", + "cryptography", ] lint = [ "black>=23.1.0", diff --git a/test/fixtures/linode_instances.json b/test/fixtures/linode_instances.json index 9f3b5bbc7..d7d7010af 100644 --- a/test/fixtures/linode_instances.json +++ b/test/fixtures/linode_instances.json @@ -14,7 +14,9 @@ "network_out": 5, "cpu": 90, "transfer_quota": 80, - "io": 5000 + "io": 5000, + "system_alerts": [123,456], + "user_alerts": [555] }, "label": "linode123", "backups": { @@ -66,7 +68,9 @@ "network_out": 5, "cpu": 90, "transfer_quota": 80, - "io": 5000 + "io": 5000, + "system_alerts": [123,456], + "user_alerts": [555] }, "label": "linode456", "backups": { @@ -106,7 +110,9 @@ "network_out": 5, "cpu": 90, "transfer_quota": 80, - "io": 5000 + "io": 5000, + "system_alerts": [123,456], + "user_alerts": [555] }, "group": "test", "hypervisor": "kvm", diff --git a/test/fixtures/linode_instances_123_clone.json b/test/fixtures/linode_instances_123_clone.json new file mode 100644 index 000000000..bcfe6b695 --- /dev/null +++ b/test/fixtures/linode_instances_123_clone.json @@ -0,0 +1,45 @@ +{ + "id": 124, + "status": "running", + "type": "g6-standard-1", + "alerts": { + "network_in": 5, + "network_out": 5, + "cpu": 90, + "transfer_quota": 80, + "io": 5000, + "system_alerts": [123,456], + "user_alerts": [555] + }, + "group": "test", + "hypervisor": "kvm", + "label": "linode124", + "backups": { + "enabled": true, + "schedule": { + "window": "W02", + "day": "Scheduling" + } + }, + "specs": { + "memory": 2048, + "disk": 30720, + "vcpus": 1, + "transfer": 2000 + }, + "ipv6": "1235:abcd::1234:abcd:89ef:67cd/64", + "created": "2017-01-01T00:00:00", + "region": "us-east-1", + "ipv4": [ + "124.45.67.89" + ], + "updated": "2017-01-01T00:00:00", + "image": "linode/ubuntu24.04", + "tags": ["something"], + "host_uuid": "3b3ddd59d9a78bb8de041391075df44de62bfec8", + "watchdog_enabled": true, + "disk_encryption": "disabled", + "lke_cluster_id": null, + "placement_group": null, + "interface_generation": "linode" +} \ No newline at end of file diff --git a/test/fixtures/monitor_streams_quotas.json b/test/fixtures/monitor_streams_quotas.json new file mode 100644 index 000000000..6d2b83138 --- /dev/null +++ b/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 +} diff --git a/test/integration/conftest.py b/test/integration/conftest.py index 6470d9df1..394eb0284 100644 --- a/test/integration/conftest.py +++ b/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( @@ -808,10 +847,80 @@ def test_monitor_client(get_monitor_token_for_db_entities): return client, entity_ids +def get_alert_channels_list(test_linode_client): + channels = list(test_linode_client.monitor.alert_channels()) + if len(channels) == 0: + pytest.fail( + "No alert channels available for testing. Please create an alert channel and try again." + ) + + return channels + + +@pytest.fixture(scope="session") +def create_alert_service_definition(test_linode_client): + rule_criteria = { + "rules": [ + { + "aggregate_function": "min", + "dimension_filters": [ + { + "dimension_label": "node_type", + "label": "Node Type", + "operator": "eq", + "value": "primary", + } + ], + "label": "Memory Usage", + "metric": "memory_usage", + "operator": "eq", + "threshold": 95, + "unit": "percent", + } + ] + } + trigger_conditions = { + "criteria_condition": "ALL", + "evaluation_period_seconds": 300, + "polling_interval_seconds": 900, + "trigger_occurrences": 3, + } + alert = test_linode_client.monitor.create_alert_definition( + service_type="dbaas", + label=get_test_label() + "-service-definition", + severity=1, + description="description", + channel_ids=[get_alert_channels_list(test_linode_client)[0].id], + rule_criteria=rule_criteria, + trigger_conditions=trigger_conditions, + ) + + yield alert + + alert.delete() + + +def get_system_alerts(client: LinodeClient): + alerts = client.monitor.alert_definitions() + system_alerts = [] + for alert in alerts.lists[0]: + if alert.type == "system": + system_alerts.append(alert) + if len(system_alerts) == 0: + raise Exception( + "No system alert definitions found. Cannot run tests dependent on system alert definitions." + ) + return system_alerts + + @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"] ) diff --git a/test/integration/filters/fixtures.py b/test/integration/filters/fixtures.py index e753236dd..6a805a853 100644 --- a/test/integration/filters/fixtures.py +++ b/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" diff --git a/test/integration/linode_client/test_linode_client.py b/test/integration/linode_client/test_linode_client.py index 0cacdf437..b4fe95787 100644 --- a/test/integration/linode_client/test_linode_client.py +++ b/test/integration/linode_client/test_linode_client.py @@ -6,7 +6,12 @@ import pytest from linode_api4 import ApiError -from linode_api4.objects import ConfigInterface, ObjectStorageKeys, Region +from linode_api4.objects import ( + Capability, + ConfigInterface, + ObjectStorageKeys, + Region, +) def is_tag_created(client, tag_label): @@ -19,7 +24,11 @@ def is_tag_created(client, tag_label): @pytest.fixture(scope="session") def setup_client_and_linode(test_linode_client, e2e_test_firewall): client = test_linode_client - region = get_region(client, {"Kubernetes", "NodeBalancers"}, "core").id + region = get_region( + client, + {Capability.lke, Capability.nodebalancers}, + "core", + ).id label = get_test_label() @@ -228,7 +237,7 @@ def test_get_account_settings(test_linode_client): # LinodeGroupTests def test_create_linode_instance_without_image(test_linode_client): client = test_linode_client - region = get_region(client, {"Linodes"}, "core").id + region = get_region(client, {Capability.linodes}, "core").id label = get_test_label() linode_instance = client.linode.instance_create( @@ -252,7 +261,11 @@ def test_create_linode_instance_with_image(setup_client_and_linode): def test_create_linode_with_interfaces(test_linode_client): client = test_linode_client - region = get_region(client, {"Vlans", "Linodes"}, site_type="core").id + region = get_region( + client, + {Capability.vlans, Capability.linodes}, + site_type="core", + ).id label = get_test_label() linode_instance = client.linode.instance_create( @@ -325,7 +338,7 @@ def test_cluster_create_with_api_objects(test_linode_client): client = test_linode_client node_type = client.linode.types()[1] # g6-standard-1 version = client.lke.versions()[0] - region = get_region(client, {"Kubernetes"}) + region = get_region(client, {Capability.lke}) node_pool = client.lke.node_pool(node_type, 3) label = get_test_label() @@ -342,7 +355,7 @@ def test_cluster_create_with_api_objects(test_linode_client): def test_fails_to_create_cluster_with_invalid_version(test_linode_client): invalid_version = "a.12" client = test_linode_client - region = get_region(client, {"Kubernetes"}).id + region = get_region(client, {Capability.lke}).id with pytest.raises(ApiError, match="not valid") as exc_info: client.lke.cluster_create( diff --git a/test/integration/models/account/test_account.py b/test/integration/models/account/test_account.py index 2922b01fa..cb29f0de9 100644 --- a/test/integration/models/account/test_account.py +++ b/test/integration/models/account/test_account.py @@ -12,6 +12,7 @@ from linode_api4.objects import ( Account, AccountSettings, + Capability, ChildAccount, Event, Login, @@ -100,7 +101,11 @@ def test_update_maintenance_policy(test_linode_client): def test_latest_get_event(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() linode = client.linode.instance_create( diff --git a/test/integration/models/firewall/test_firewall.py b/test/integration/models/firewall/test_firewall.py index 9ccd90b05..7dd77622a 100644 --- a/test/integration/models/firewall/test_firewall.py +++ b/test/integration/models/firewall/test_firewall.py @@ -4,13 +4,17 @@ import pytest -from linode_api4.objects import Firewall, FirewallDevice +from linode_api4.objects import Capability, Firewall, FirewallDevice @pytest.fixture(scope="session") def linode_fw(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() linode_instance = client.linode.instance_create( diff --git a/test/integration/models/image/test_image.py b/test/integration/models/image/test_image.py index 56b1ecf0d..ff89fc692 100644 --- a/test/integration/models/image/test_image.py +++ b/test/integration/models/image/test_image.py @@ -6,7 +6,7 @@ import pytest from linode_api4 import LinodeClient -from linode_api4.objects import Image +from linode_api4.objects import Capability, Image DISALLOWED_IMAGE_REGIONS = { "gb-lon", @@ -30,7 +30,7 @@ def get_image_upload_regions(client: LinodeClient): region for region in get_regions( client, - capabilities={"Linodes", "Object Storage"}, + capabilities={Capability.linodes, Capability.object_storage}, site_type="core", ) if region.id not in DISALLOWED_IMAGE_REGIONS diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index 0d607223b..f619a2c5c 100644 --- a/test/integration/models/linode/test_linode.py +++ b/test/integration/models/linode/test_linode.py @@ -1,6 +1,6 @@ import ipaddress import time -from test.integration.conftest import get_region +from test.integration.conftest import get_region, get_system_alerts from test.integration.helpers import ( get_test_label, retry_sending_request, @@ -10,8 +10,9 @@ import pytest -from linode_api4.errors import ApiError +from linode_api4.errors import ApiError, UnexpectedResponseError from linode_api4.objects import ( + Capability, Config, ConfigInterface, ConfigInterfaceIPv4, @@ -29,7 +30,11 @@ @pytest.fixture(scope="session") def linode_with_volume_firewall(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() rules = { @@ -75,7 +80,11 @@ def linode_with_volume_firewall(test_linode_client): @pytest.fixture(scope="function") def linode_for_legacy_interface_tests(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( @@ -158,7 +167,11 @@ def linode_for_vpu_tests(test_linode_client, e2e_test_firewall): @pytest.fixture def linode_for_disk_tests(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() linode_instance = client.linode.instance_create( @@ -192,7 +205,10 @@ def linode_for_disk_tests(test_linode_client, e2e_test_firewall): @pytest.fixture def linode_with_block_storage_encryption(test_linode_client, e2e_test_firewall): client = test_linode_client - region = get_region(client, {"Linodes", "Block Storage Encryption"}) + region = get_region( + client, + {Capability.linodes, Capability.blockstorage_encryption}, + ) label = get_test_label() linode_instance = client.linode.instance_create( @@ -212,7 +228,11 @@ def linode_with_block_storage_encryption(test_linode_client, e2e_test_firewall): @pytest.fixture def create_linode_for_long_running_tests(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() linode_instance = client.linode.instance_create( @@ -233,7 +253,7 @@ def create_linode_for_long_running_tests(test_linode_client, e2e_test_firewall): def linode_with_disk_encryption(test_linode_client, request): client = test_linode_client - target_region = get_region(client, {"Disk Encryption"}) + target_region = get_region(client, {Capability.disk_encryption}) label = get_test_label(length=8) disk_encryption = request.param @@ -257,7 +277,11 @@ def linode_with_disk_encryption(test_linode_client, request): def create_linode_with_authorized_key(test_linode_client, ssh_key_gen): 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( @@ -280,10 +304,36 @@ def get_status(linode: Instance, status: str): return linode.status == status +def wait_for_clone_complete_and_delete_linode( + interval: int, timeout: int, linode: Instance +) -> object: + end_time = time.time() + timeout + while time.time() < end_time: + try: + return linode.delete() + except ApiError as err: + if "[400] Linode is the target of an ongoing clone" not in str(err): + raise UnexpectedResponseError(f"Unexpected delete linode error") + time.sleep(interval) + raise TimeoutError( + f"Timeout Error: not possible to delete just cloned linode in {timeout} seconds" + ) + + def instance_type_condition(linode: Instance, type: str): return type in str(linode.type) +def test_get_linodes_verify_alerts(test_linode_client, create_linode): + linodes_list = test_linode_client.linode.instances().lists[0] + assert len(linodes_list) > 0 + assert linodes_list[0].alerts.cpu + assert linodes_list[0].alerts.io + assert linodes_list[0].alerts.network_in + assert linodes_list[0].alerts.network_out + assert linodes_list[0].alerts.transfer_quota + + def test_get_linode(test_linode_client, linode_with_volume_firewall): linode = test_linode_client.load(Instance, linode_with_volume_firewall.id) @@ -311,7 +361,7 @@ def test_linode_transfer(test_linode_client, linode_with_volume_firewall): def test_linode_rebuild(test_linode_client): client = test_linode_client - region = get_region(client, {"Disk Encryption"}) + region = get_region(client, {Capability.disk_encryption}) label = get_test_label() + "_rebuild" @@ -337,6 +387,8 @@ def test_linode_rebuild(test_linode_client): assert linode.status == "rebuilding" assert linode.image.id == "linode/debian12" + assert linode.alerts.cpu + assert linode.alerts.io assert linode.disk_encryption == InstanceDiskEncryptionType.disabled @@ -373,7 +425,11 @@ def test_update_linode(create_linode): def test_delete_linode(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() linode_instance = client.linode.instance_create( @@ -401,6 +457,79 @@ def test_linode_reboot(create_linode): assert linode.status == "running" +def test_linode_alerts_workflow(test_linode_client, create_linode): + linode = create_linode + parent_linode_id = create_linode.id + assert linode.alerts.cpu == 90 + assert linode.alerts.io == 10000 + assert linode.alerts.network_in == 10 + assert linode.alerts.network_out == 10 + assert linode.alerts.transfer_quota == 80 + assert isinstance(linode.alerts.system_alerts, list) + assert isinstance(linode.alerts.user_alerts, list) + + linode = test_linode_client.load(Instance, parent_linode_id) + assert linode.alerts.cpu == 90 + assert linode.alerts.io == 10000 + assert linode.alerts.network_in == 10 + assert linode.alerts.network_out == 10 + assert linode.alerts.transfer_quota == 80 + assert isinstance(linode.alerts.system_alerts, list) + assert isinstance(linode.alerts.user_alerts, list) + + linode.alerts = { + "cpu": 50, + "io": 6000, + "network_in": 20, + "network_out": 20, + "transfer_quota": 40, + } + linode_save_status = linode.save() + assert linode_save_status == True + linode = test_linode_client.load(Instance, parent_linode_id) + assert linode.alerts["cpu"] == 50 + assert linode.alerts["io"] == 6000 + assert linode.alerts["network_in"] == 20 + assert linode.alerts["network_out"] == 20 + assert linode.alerts["transfer_quota"] == 40 + + wait_for_condition(10, 100, get_status, linode, "running") + new_linode = retry_sending_request( + 5, + linode.clone, + region=linode.region.id, + instance_type=linode.type.id, + label=get_test_label(), + ) + + try: + assert new_linode.alerts.cpu == 90 + assert new_linode.alerts.io == 10000 + assert new_linode.alerts.network_in == 10 + assert new_linode.alerts.network_out == 10 + assert new_linode.alerts.transfer_quota == 80 + assert isinstance(new_linode.alerts.system_alerts, list) + assert isinstance(new_linode.alerts.user_alerts, list) + finally: + wait_for_clone_complete_and_delete_linode(10, 100, new_linode) + + +def test_update_linode_aclp_alerts( + test_linode_client, create_linode, create_alert_service_definition +): + linode = create_linode + sample_system_alert = get_system_alerts(test_linode_client)[0].id + + linode.alerts = { + "user_alerts": [create_alert_service_definition.id], + "system_alerts": [sample_system_alert], + } + linode.save() + linode = test_linode_client.load(Instance, linode.id) + assert linode.alerts["user_alerts"] == [create_alert_service_definition.id] + assert linode.alerts["system_alerts"] == [sample_system_alert] + + def test_linode_shutdown(create_linode): linode = create_linode @@ -568,7 +697,7 @@ def test_linode_with_block_storage_encryption( linode_with_block_storage_encryption, ): linode = linode_with_block_storage_encryption - assert "Block Storage Encryption" in linode.capabilities + assert Capability.blockstorage_encryption in linode.capabilities def wait_for_disk_status(disk: Disk, timeout): @@ -648,9 +777,15 @@ def test_linode_ips(create_linode): def test_linode_initate_migration(test_linode_client, e2e_test_firewall): client = test_linode_client label = get_test_label() + "_migration" - region = get_region(client, {"Linodes", "Cloud Firewall"}, site_type="core") + region = get_region( + client, + {Capability.linodes, Capability.firewall}, + site_type="core", + ) region_migrate = get_region( - client, {"Linodes", "Cloud Firewall"}, site_type="core" + client, + {Capability.linodes, Capability.firewall}, + site_type="core", ) # Cannot migrate linode to the same region @@ -659,7 +794,9 @@ def test_linode_initate_migration(test_linode_client, e2e_test_firewall): break region_migrate = get_region( - client, {"Linodes", "Cloud Firewall"}, site_type="core" + client, + {Capability.linodes, Capability.firewall}, + site_type="core", ) else: pytest.skip("No alternative region to be used for linode migration") @@ -1186,7 +1323,11 @@ def test_delete_interface_containing_vpc( def test_create_linode_with_maintenance_policy(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() policies = client.maintenance.maintenance_policies() @@ -1233,7 +1374,11 @@ def test_expected_error_if_fields_authorized_users_authorized_keys_root_pass_are 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) with pytest.raises(ValueError) as create_instance_error: diff --git a/test/integration/models/lke/test_lke.py b/test/integration/models/lke/test_lke.py index d5b21089e..69a3d0655 100644 --- a/test/integration/models/lke/test_lke.py +++ b/test/integration/models/lke/test_lke.py @@ -18,6 +18,7 @@ from linode_api4.common import RegionPrice from linode_api4.errors import ApiError from linode_api4.objects import ( + Capability, LKECluster, LKENodePool, LKENodePoolTaint, @@ -31,7 +32,10 @@ def lke_cluster(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_pools = test_linode_client.lke.node_pool(node_type, 3) label = get_test_label() + "_cluster" @@ -49,7 +53,7 @@ def lke_cluster(test_linode_client): def lke_cluster_with_acl(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"}) + region = get_region(test_linode_client, {Capability.lke}) node_pools = test_linode_client.lke.node_pool(node_type, 1) label = get_test_label() + "_cluster" @@ -79,7 +83,7 @@ def lke_cluster_with_labels_and_taints(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"}) + region = get_region(test_linode_client, {Capability.lke}) node_pools = test_linode_client.lke.node_pool( node_type, @@ -114,7 +118,10 @@ def lke_cluster_with_labels_and_taints(test_linode_client): def lke_cluster_with_apl(test_linode_client): 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}, + ) # NOTE: g6-dedicated-4 is the minimum APL-compatible Linode type node_pools = test_linode_client.lke.node_pool("g6-dedicated-4", 3) @@ -144,7 +151,8 @@ def lke_cluster_enterprise(e2e_test_firewall, test_linode_client): )[0] region = get_region( - test_linode_client, {"Kubernetes Enterprise", "Disk Encryption"} + test_linode_client, + {Capability.lke_e, Capability.disk_encryption}, ) node_pools = test_linode_client.lke.node_pool( diff --git a/test/integration/models/lock/test_lock.py b/test/integration/models/lock/test_lock.py index 31f89b992..a964eeced 100644 --- a/test/integration/models/lock/test_lock.py +++ b/test/integration/models/lock/test_lock.py @@ -6,7 +6,7 @@ import pytest -from linode_api4.objects import Lock, LockType +from linode_api4.objects import Capability, Lock, LockType @pytest.fixture(scope="function") @@ -15,7 +15,11 @@ def linode_for_lock(test_linode_client, e2e_test_firewall): Create a Linode instance for testing locks. """ 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( diff --git a/test/integration/models/monitor/test_monitor_logs.py b/test/integration/models/monitor/test_monitor_logs.py index 9d4148c17..b5958d988 100644 --- a/test/integration/models/monitor/test_monitor_logs.py +++ b/test/integration/models/monitor/test_monitor_logs.py @@ -11,6 +11,7 @@ from linode_api4 import LinodeClient, LogsStreamType, PaginatedList, Region from linode_api4.objects import ( + Capability, ObjectStorageACL, ObjectStorageBucket, ObjectStorageKeys, @@ -19,6 +20,7 @@ AkamaiObjectStorageLogsDestinationDetails, LogsDestination, LogsStream, + LogsStreamQuota, LogsStreamStatus, ) @@ -34,7 +36,7 @@ @pytest.fixture(scope="session") def region(test_linode_client: LinodeClient): - region = get_region(test_linode_client, {"Object Storage"}) + region = get_region(test_linode_client, {Capability.object_storage}) yield region @@ -248,7 +250,7 @@ def test_fails_to_create_destination_invalid_type( ) assert excinfo.value.status == 400 assert excinfo.value.errors == [ - "Must be one of akamai_object_storage, custom_https" + "Must be one of akamai_object_storage, custom_https, traffic_peak" ] @@ -519,3 +521,21 @@ def test_update_stream_destinations( assert ( snapshot_updated.destinations[0].id == create_secondary_destination.id ) + + +@_SKIP_STREAM_TESTS +def test_list_stream_quotas(test_linode_client: LinodeClient): + """Test that stream quota definitions can be retrieved from the API.""" + quotas = test_linode_client.monitor.stream_quotas() + + assert isinstance(quotas, PaginatedList) + assert len(quotas) > 0 + assert all(isinstance(quota, LogsStreamQuota) for quota in quotas) + assert all( + quota.quota_id + and quota.quota_name + and quota.quota_type + and quota.description + for quota in quotas + ) + assert all(quota.quota_limit >= 0 for quota in quotas) diff --git a/test/integration/models/networking/test_networking.py b/test/integration/models/networking/test_networking.py index 18b6dabb1..de5b53f4c 100644 --- a/test/integration/models/networking/test_networking.py +++ b/test/integration/models/networking/test_networking.py @@ -17,6 +17,7 @@ from linode_api4 import ApiError, Instance, LinodeClient from linode_api4.objects import ( + Capability, Config, ConfigInterfaceIPv4, Firewall, @@ -35,7 +36,7 @@ base_url=get_api_url(), ca_path=get_api_ca_file(), ), - {"Linodes", "Cloud Firewall"}, + {Capability.linodes, Capability.firewall}, site_type="core", ) @@ -94,7 +95,7 @@ def create_linode_without_firewall(test_linode_client): """ client = test_linode_client - region = get_region(client, {"Cloud Firewall"}, "core").id + region = get_region(client, {Capability.firewall}, "core").id label = get_test_label() diff --git a/test/integration/models/nodebalancer/test_nodebalancer.py b/test/integration/models/nodebalancer/test_nodebalancer.py index 97e76e88f..3367f1477 100644 --- a/test/integration/models/nodebalancer/test_nodebalancer.py +++ b/test/integration/models/nodebalancer/test_nodebalancer.py @@ -11,6 +11,7 @@ from linode_api4 import ApiError, LinodeClient, NodeBalancer from linode_api4.objects import ( + Capability, NodeBalancerConfig, NodeBalancerNode, NodeBalancerType, @@ -24,7 +25,12 @@ base_url=get_api_url(), ca_path=get_api_ca_file(), ), - {"Linodes", "Cloud Firewall", "NodeBalancers"}, + { + Capability.linodes, + Capability.firewall, + Capability.nodebalancers, + Capability.premium_nodebalancers, + }, site_type="core", ) diff --git a/test/integration/models/placement/test_placement.py b/test/integration/models/placement/test_placement.py index 1a7e38e27..0badaa31e 100644 --- a/test/integration/models/placement/test_placement.py +++ b/test/integration/models/placement/test_placement.py @@ -13,6 +13,7 @@ PlacementGroupPolicy, PlacementGroupType, ) +from linode_api4.objects import Capability @pytest.mark.smoke @@ -76,7 +77,7 @@ def test_pg_migration( pg_outbound = client.placement.group_create( label_pg, - get_region(test_linode_client, {"Placement Group"}), + get_region(test_linode_client, {Capability.placement_group}), PlacementGroupType.anti_affinity_local, PlacementGroupPolicy.flexible, ) diff --git a/test/integration/models/sharegroups/test_sharegroups.py b/test/integration/models/sharegroups/test_sharegroups.py index e3ac79652..dcb23e574 100644 --- a/test/integration/models/sharegroups/test_sharegroups.py +++ b/test/integration/models/sharegroups/test_sharegroups.py @@ -5,6 +5,7 @@ import pytest from linode_api4.objects import ( + Capability, Image, ImageShareGroup, ImageShareGroupImagesToAdd, @@ -36,7 +37,11 @@ def wait_for_image_status( @pytest.fixture(scope="module") def sample_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( diff --git a/test/integration/models/volume/test_blockstorage.py b/test/integration/models/volume/test_blockstorage.py index 69a60868b..08baa8614 100644 --- a/test/integration/models/volume/test_blockstorage.py +++ b/test/integration/models/volume/test_blockstorage.py @@ -2,12 +2,17 @@ from test.integration.helpers import get_test_label, retry_sending_request from linode_api4 import Instance +from linode_api4.objects import Capability def test_config_create_with_extended_volume_limit(test_linode_client): client = test_linode_client - region = get_region(client, {"Linodes", "Block Storage"}, site_type="core") + region = get_region( + client, + {Capability.linodes, Capability.block_storage}, + site_type="core", + ) label = get_test_label() linode = client.linode.instance_create( @@ -46,7 +51,11 @@ def test_config_create_with_extended_volume_limit(test_linode_client): def test_config_create_with_device_map(test_linode_client): client = test_linode_client - region = get_region(client, {"Linodes", "Block Storage"}, site_type="core") + region = get_region( + client, + {Capability.linodes, Capability.block_storage}, + site_type="core", + ) label = get_test_label() linode = client.linode.instance_create( diff --git a/test/integration/models/volume/test_volume.py b/test/integration/models/volume/test_volume.py index 7f9045e2e..94765f701 100644 --- a/test/integration/models/volume/test_volume.py +++ b/test/integration/models/volume/test_volume.py @@ -15,7 +15,7 @@ import pytest from linode_api4 import LinodeClient -from linode_api4.objects import RegionPrice, Volume, VolumeType +from linode_api4.objects import Capability, RegionPrice, Volume, VolumeType TEST_REGION = get_region( LinodeClient( @@ -23,7 +23,7 @@ base_url=get_api_url(), ca_path=get_api_ca_file(), ), - {"Linodes", "Cloud Firewall"}, + {Capability.linodes, Capability.firewall}, site_type="core", ) diff --git a/test/integration/models/vpc/test_vpc.py b/test/integration/models/vpc/test_vpc.py index de4a36e5a..5d9473b53 100644 --- a/test/integration/models/vpc/test_vpc.py +++ b/test/integration/models/vpc/test_vpc.py @@ -3,6 +3,7 @@ import pytest from linode_api4 import VPC, ApiError, VPCIPv4DefaultRange, VPCSubnet +from linode_api4.objects import Capability @pytest.mark.smoke @@ -60,7 +61,7 @@ def test_fails_create_vpc_invalid_data(test_linode_client): with pytest.raises(ApiError) as excinfo: test_linode_client.vpcs.create( label="invalid_label!!", - region=get_region(test_linode_client, {"VPCs"}), + region=get_region(test_linode_client, {Capability.vpcs}), description="test description", ) assert excinfo.value.status == 400 diff --git a/test/unit/linode_client_test.py b/test/unit/linode_client_test.py index 78a542210..c17be6006 100644 --- a/test/unit/linode_client_test.py +++ b/test/unit/linode_client_test.py @@ -10,6 +10,7 @@ ObjectStorageACL, ObjectStorageCluster, ) +from linode_api4.objects.region import Capability class LinodeClientGeneralTest(ClientBaseCase): @@ -46,11 +47,11 @@ def test_get_account(self): self.assertEqual( a.capabilities, [ - "Linodes", - "NodeBalancers", - "Block Storage", - "Object Storage", - "Linode Interfaces", + Capability.linodes, + Capability.nodebalancers, + Capability.block_storage, + Capability.object_storage, + Capability.linode_interfaces, ], ) @@ -66,21 +67,21 @@ def test_get_regions(self): self.assertEqual( region.capabilities, [ - "Linodes", - "NodeBalancers", - "Block Storage", - "Object Storage", - "Linode Interfaces", + Capability.linodes, + Capability.nodebalancers, + Capability.block_storage, + Capability.object_storage, + Capability.linode_interfaces, ], ) else: self.assertEqual( region.capabilities, [ - "Linodes", - "NodeBalancers", - "Block Storage", - "Linode Interfaces", + Capability.linodes, + Capability.nodebalancers, + Capability.block_storage, + Capability.linode_interfaces, ], ) self.assertEqual(region.status, "ok") diff --git a/test/unit/objects/linode_test.py b/test/unit/objects/linode_test.py index eafbac328..01c8de058 100644 --- a/test/unit/objects/linode_test.py +++ b/test/unit/objects/linode_test.py @@ -26,6 +26,7 @@ Disk, Image, Instance, + InstanceACLPAlertsOptions, InstanceBackups, InstanceBackupSchedule, StackScript, @@ -61,6 +62,14 @@ def test_get_linode(self): self.assertEqual(linode.lke_cluster_id, None) self.assertEqual(linode.maintenance_policy, "linode/migrate") + self.assertEqual(linode.alerts.cpu, 90) + self.assertEqual(linode.alerts.io, 5000) + self.assertEqual(linode.alerts.network_in, 5) + self.assertEqual(linode.alerts.network_out, 5) + self.assertEqual(linode.alerts.transfer_quota, 80) + self.assertEqual(linode.alerts.system_alerts, [123, 456]) + self.assertEqual(linode.alerts.user_alerts, [555]) + self.assertIsInstance(linode.backups, InstanceBackups) self.assertTrue(linode.backups.enabled) self.assertTrue(linode.backups.available) @@ -226,6 +235,8 @@ def test_update_linode(self): "network_in": 5, "network_out": 5, "transfer_quota": 80, + "system_alerts": [123, 456], + "user_alerts": [555], }, "backups": { "enabled": True, @@ -783,6 +794,24 @@ def test_create_interface_vlan(self): LinodeInterfaceTest.assert_linode_124_interface_789(result) + def test_instance_clone_with_alerts(self): + src = Instance(self.client, 123) + + with self.mock_post("linode/instances/123/clone") as m: + src.clone( + region="us-east", + instance_type="g6-standard-1", + alerts=InstanceACLPAlertsOptions( + system_alerts=[123, 456], + user_alerts=[555], + ), + ) + + assert m.call_data["alerts"] == { + "system_alerts": [123, 456], + "user_alerts": [555], + } + def test_create_interface_rdma_vpc_rejected(self): """ Tests that attempting to create an RDMA VPC interface via diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index c0999e485..5c42d2aa4 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -16,6 +16,7 @@ DestinationAuthentication, LogsDestinationDetailsBase, LogsStreamDetails, + LogsStreamQuota, LogsStreamType, ) @@ -653,6 +654,41 @@ def test_delete_stream(self): self.assertEqual(m.call_url, "/monitor/streams/1") + def test_stream_quotas(self): + """ + Test that stream_quotas returns a paginated list of quota objects. + """ + url = "/monitor/streams/quotas" + with self.mock_get(url) as m: + result = self.client.monitor.stream_quotas() + + self.assertEqual(m.call_url, url) + self.assertEqual(len(result), 2) + self.assertIsInstance(result[0], LogsStreamQuota) + self.assertEqual(result[0].quota_id, "aclp_audit_logs_streams") + self.assertEqual(result[0].quota_name, "Number of Audit Logs Streams") + self.assertEqual( + result[0].description, + "Current number of audit logs streams per account", + ) + self.assertEqual(result[0].quota_limit, 5) + self.assertEqual( + result[0].quota_type, "logs_streams_aclp_audit_logs_streams" + ) + self.assertIsInstance(result[1], LogsStreamQuota) + self.assertEqual(result[1].quota_id, "aclp_lke_audit_logs_streams") + self.assertEqual( + result[1].quota_name, "Number of LKE Audit Logs Streams" + ) + self.assertEqual( + result[1].description, + "Current number of LKE audit logs streams per account", + ) + self.assertEqual(result[1].quota_limit, 10) + self.assertEqual( + result[1].quota_type, "logs_streams_aclp_lke_audit_logs_streams" + ) + class LkeAuditLogsStreamTest(ClientBaseCase): """