From 4d3c836d0b6d7e55d56c8affe278646c36bb6ca1 Mon Sep 17 00:00:00 2001 From: ykim-1 Date: Thu, 23 Jan 2025 14:38:55 -0800 Subject: [PATCH 1/3] refactor unit test into group and object directories --- test/unit/groups/database_test.py | 185 ++++++++++++++++++ test/unit/groups/image_test.py | 37 ++++ test/unit/groups/linode_test.py | 118 +++++++++++ test/unit/groups/lke_test.py | 43 ++++ test/unit/groups/placement_test.py | 68 +++++++ test/unit/{objects => groups}/polling_test.py | 0 test/unit/groups/region_test.py | 51 +++++ test/unit/groups/vpc_test.py | 107 ++++++++++ test/unit/objects/database_test.py | 170 ---------------- test/unit/objects/image_test.py | 30 --- test/unit/objects/linode_test.py | 111 +---------- test/unit/objects/lke_test.py | 31 --- test/unit/objects/placement_test.py | 42 ---- test/unit/objects/region_test.py | 43 ---- test/unit/objects/vpc_test.py | 75 ------- 15 files changed, 610 insertions(+), 501 deletions(-) create mode 100644 test/unit/groups/database_test.py create mode 100644 test/unit/groups/image_test.py create mode 100644 test/unit/groups/linode_test.py create mode 100644 test/unit/groups/lke_test.py create mode 100644 test/unit/groups/placement_test.py rename test/unit/{objects => groups}/polling_test.py (100%) create mode 100644 test/unit/groups/region_test.py create mode 100644 test/unit/groups/vpc_test.py diff --git a/test/unit/groups/database_test.py b/test/unit/groups/database_test.py new file mode 100644 index 000000000..446ed3617 --- /dev/null +++ b/test/unit/groups/database_test.py @@ -0,0 +1,185 @@ +from test.unit.base import ClientBaseCase + +from linode_api4.objects import MySQLDatabase + + +class DatabaseTest(ClientBaseCase): + """ + Tests methods of the DatabaseGroup class + """ + + def test_get_types(self): + """ + Test that database types are properly handled + """ + types = self.client.database.types() + + self.assertEqual(len(types), 1) + self.assertEqual(types[0].type_class, "nanode") + self.assertEqual(types[0].id, "g6-nanode-1") + self.assertEqual(types[0].engines.mysql[0].price.monthly, 20) + + def test_get_engines(self): + """ + Test that database engines are properly handled + """ + engines = self.client.database.engines() + + self.assertEqual(len(engines), 2) + + self.assertEqual(engines[0].engine, "mysql") + self.assertEqual(engines[0].id, "mysql/8.0.26") + self.assertEqual(engines[0].version, "8.0.26") + + self.assertEqual(engines[1].engine, "postgresql") + self.assertEqual(engines[1].id, "postgresql/10.14") + self.assertEqual(engines[1].version, "10.14") + + def test_get_databases(self): + """ + Test that databases are properly handled + """ + dbs = self.client.database.instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "mysql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-123-456-mysql-mysql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-123-456-mysql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "8.0.26") + + def test_database_instance(self): + """ + Ensures that the .instance attribute properly translates database types + """ + + dbs = self.client.database.instances() + db_translated = dbs[0].instance + + self.assertTrue(isinstance(db_translated, MySQLDatabase)) + self.assertEqual(db_translated.ssl_connection, True) + + +class MySQLDatabaseTest(ClientBaseCase): + """ + Tests methods of the MySQLDatabase class + """ + + def test_get_instances(self): + """ + Test that database types are properly handled + """ + dbs = self.client.database.mysql_instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "mysql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-123-456-mysql-mysql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-123-456-mysql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "8.0.26") + + def test_create(self): + """ + Test that MySQL databases can be created + """ + + with self.mock_post("/databases/mysql/instances") as m: + # We don't care about errors here; we just want to + # validate the request. + try: + self.client.database.mysql_create( + "cool", + "us-southeast", + "mysql/8.0.26", + "g6-standard-1", + cluster_size=3, + ) + except Exception: + pass + + self.assertEqual(m.method, "post") + self.assertEqual(m.call_url, "/databases/mysql/instances") + self.assertEqual(m.call_data["label"], "cool") + self.assertEqual(m.call_data["region"], "us-southeast") + self.assertEqual(m.call_data["engine"], "mysql/8.0.26") + self.assertEqual(m.call_data["type"], "g6-standard-1") + self.assertEqual(m.call_data["cluster_size"], 3) + + +class PostgreSQLDatabaseTest(ClientBaseCase): + """ + Tests methods of the PostgreSQLDatabase class + """ + + def test_get_instances(self): + """ + Test that database types are properly handled + """ + dbs = self.client.database.postgresql_instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "postgresql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-0000-000-pgsql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-0000-000-pgsql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "13.2") + + def test_create(self): + """ + Test that PostgreSQL databases can be created + """ + + with self.mock_post("/databases/postgresql/instances") as m: + # We don't care about errors here; we just want to + # validate the request. + try: + self.client.database.postgresql_create( + "cool", + "us-southeast", + "postgresql/13.2", + "g6-standard-1", + cluster_size=3, + ) + except Exception: + pass + + self.assertEqual(m.method, "post") + self.assertEqual(m.call_url, "/databases/postgresql/instances") + self.assertEqual(m.call_data["label"], "cool") + self.assertEqual(m.call_data["region"], "us-southeast") + self.assertEqual(m.call_data["engine"], "postgresql/13.2") + self.assertEqual(m.call_data["type"], "g6-standard-1") + self.assertEqual(m.call_data["cluster_size"], 3) diff --git a/test/unit/groups/image_test.py b/test/unit/groups/image_test.py new file mode 100644 index 000000000..e2aab386b --- /dev/null +++ b/test/unit/groups/image_test.py @@ -0,0 +1,37 @@ +from test.unit.base import ClientBaseCase + + +class ImageTest(ClientBaseCase): + """ + Tests methods of the Image class + """ + + def test_image_create_cloud_init(self): + """ + Test that an image can be created successfully with cloud-init. + """ + + with self.mock_post("images/private/123") as m: + self.client.images.create( + "Test Image", + "us-southeast", + description="very real image upload.", + cloud_init=True, + ) + + self.assertTrue(m.call_data["cloud_init"]) + + def test_image_create_upload_cloud_init(self): + """ + Test that an image upload URL can be created successfully with cloud-init. + """ + + with self.mock_post("images/upload") as m: + self.client.images.create_upload( + "Test Image", + "us-southeast", + description="very real image upload.", + cloud_init=True, + ) + + self.assertTrue(m.call_data["cloud_init"]) diff --git a/test/unit/groups/linode_test.py b/test/unit/groups/linode_test.py new file mode 100644 index 000000000..8112a5d93 --- /dev/null +++ b/test/unit/groups/linode_test.py @@ -0,0 +1,118 @@ +from test.unit.base import ClientBaseCase + +from linode_api4 import InstancePlacementGroupAssignment +from linode_api4.objects import ConfigInterface + + +class LinodeTest(ClientBaseCase): + """ + Tests methods of the Linode class + """ + + def test_instance_create_with_user_data(self): + """ + Tests that the metadata field is populated on Linode create. + """ + + with self.mock_post("linode/instances/123") as m: + self.client.linode.instance_create( + "g6-nanode-1", + "us-southeast", + metadata=self.client.linode.build_instance_metadata( + user_data="cool" + ), + ) + + self.assertEqual( + m.call_data, + { + "region": "us-southeast", + "type": "g6-nanode-1", + "metadata": {"user_data": "Y29vbA=="}, + }, + ) + + def test_instance_create_with_interfaces(self): + """ + Tests that user can pass a list of interfaces on Linode create. + """ + interfaces = [ + {"purpose": "public"}, + ConfigInterface( + purpose="vlan", label="cool-vlan", ipam_address="10.0.0.4/32" + ), + ] + with self.mock_post("linode/instances/123") as m: + self.client.linode.instance_create( + "us-southeast", + "g6-nanode-1", + interfaces=interfaces, + ) + + self.assertEqual( + m.call_data["interfaces"], + [ + {"purpose": "public"}, + { + "purpose": "vlan", + "label": "cool-vlan", + "ipam_address": "10.0.0.4/32", + }, + ], + ) + + def test_build_instance_metadata(self): + """ + Tests that the metadata field is built correctly. + """ + self.assertEqual( + self.client.linode.build_instance_metadata(user_data="cool"), + {"user_data": "Y29vbA=="}, + ) + + self.assertEqual( + self.client.linode.build_instance_metadata( + user_data="cool", encode_user_data=False + ), + {"user_data": "cool"}, + ) + + def test_create_with_placement_group(self): + """ + Tests that you can create a Linode with a Placement Group + """ + + with self.mock_post("linode/instances/123") as m: + self.client.linode.instance_create( + "g6-nanode-1", + "eu-west", + placement_group=InstancePlacementGroupAssignment( + id=123, + compliant_only=True, + ), + ) + + self.assertEqual( + m.call_data["placement_group"], {"id": 123, "compliant_only": True} + ) + + +class TypeTest(ClientBaseCase): + def test_get_types(self): + """ + Tests that Linode types can be returned + """ + types = self.client.linode.types() + + self.assertEqual(len(types), 5) + for t in types: + self.assertTrue(t._populated) + self.assertIsNotNone(t.id) + self.assertIsNotNone(t.label) + self.assertIsNotNone(t.disk) + self.assertIsNotNone(t.type_class) + self.assertIsNotNone(t.gpus) + self.assertIsNone(t.successor) + self.assertIsNotNone(t.region_prices) + self.assertIsNotNone(t.addons.backups.region_prices) + self.assertIsNotNone(t.accelerated_devices) diff --git a/test/unit/groups/lke_test.py b/test/unit/groups/lke_test.py new file mode 100644 index 000000000..a39db81a6 --- /dev/null +++ b/test/unit/groups/lke_test.py @@ -0,0 +1,43 @@ +from test.unit.base import ClientBaseCase + +from linode_api4.objects import ( + LKEClusterControlPlaneACLAddressesOptions, + LKEClusterControlPlaneACLOptions, + LKEClusterControlPlaneOptions, +) + + +class LKETest(ClientBaseCase): + """ + Tests methods of the LKE class + """ + + def test_cluster_create_with_acl(self): + """ + Tests that an LKE cluster can be created with a control plane ACL configuration. + """ + + with self.mock_post("lke/clusters") as m: + self.client.lke.cluster_create( + "us-mia", + "test-acl-cluster", + [self.client.lke.node_pool("g6-nanode-1", 3)], + "1.29", + control_plane=LKEClusterControlPlaneOptions( + acl=LKEClusterControlPlaneACLOptions( + enabled=True, + addresses=LKEClusterControlPlaneACLAddressesOptions( + ipv4=["10.0.0.1/32"], ipv6=["1234::5678"] + ), + ) + ), + ) + + assert "high_availability" not in m.call_data["control_plane"] + assert m.call_data["control_plane"]["acl"]["enabled"] + assert m.call_data["control_plane"]["acl"]["addresses"]["ipv4"] == [ + "10.0.0.1/32" + ] + assert m.call_data["control_plane"]["acl"]["addresses"]["ipv6"] == [ + "1234::5678" + ] diff --git a/test/unit/groups/placement_test.py b/test/unit/groups/placement_test.py new file mode 100644 index 000000000..3c8337845 --- /dev/null +++ b/test/unit/groups/placement_test.py @@ -0,0 +1,68 @@ +from test.unit.base import ClientBaseCase + +from linode_api4 import PlacementGroupPolicy +from linode_api4.objects import ( + MigratedInstance, + PlacementGroup, + PlacementGroupMember, + PlacementGroupType, +) + + +class PlacementTest(ClientBaseCase): + """ + Tests methods of the Placement Group + """ + + def test_list_pgs(self): + """ + Tests that you can list PGs. + """ + + pgs = self.client.placement.groups() + + self.validate_pg_123(pgs[0]) + assert pgs[0]._populated + + def test_create_pg(self): + """ + Tests that you can create a Placement Group. + """ + + with self.mock_post("/placement/groups/123") as m: + pg = self.client.placement.group_create( + "test", + "eu-west", + PlacementGroupType.anti_affinity_local, + PlacementGroupPolicy.strict, + ) + + assert m.call_url == "/placement/groups" + + self.assertEqual( + m.call_data, + { + "label": "test", + "region": "eu-west", + "placement_group_type": str( + PlacementGroupType.anti_affinity_local + ), + "placement_group_policy": PlacementGroupPolicy.strict, + }, + ) + + assert pg._populated + self.validate_pg_123(pg) + + def validate_pg_123(self, pg: PlacementGroup): + assert pg.id == 123 + assert pg.label == "test" + assert pg.region.id == "eu-west" + assert pg.placement_group_type == "anti_affinity:local" + assert pg.placement_group_policy == "strict" + assert pg.is_compliant + assert pg.members[0] == PlacementGroupMember( + linode_id=123, is_compliant=True + ) + assert pg.migrations.inbound[0] == MigratedInstance(linode_id=123) + assert pg.migrations.outbound[0] == MigratedInstance(linode_id=456) diff --git a/test/unit/objects/polling_test.py b/test/unit/groups/polling_test.py similarity index 100% rename from test/unit/objects/polling_test.py rename to test/unit/groups/polling_test.py diff --git a/test/unit/groups/region_test.py b/test/unit/groups/region_test.py new file mode 100644 index 000000000..fe44c13ab --- /dev/null +++ b/test/unit/groups/region_test.py @@ -0,0 +1,51 @@ +import json +from test.unit.base import ClientBaseCase + +from linode_api4.objects.region import RegionAvailabilityEntry + + +class RegionTest(ClientBaseCase): + """ + Tests methods of the Region class + """ + + def test_list_availability(self): + """ + Tests that region availability can be listed and filtered on. + """ + + with self.mock_get("/regions/availability") as m: + avail_entries = self.client.regions.availability( + RegionAvailabilityEntry.filters.region == "us-east", + RegionAvailabilityEntry.filters.plan == "premium4096.7", + ) + + assert len(avail_entries) > 0 + + for entry in avail_entries: + assert entry.region is not None + assert len(entry.region) > 0 + + assert entry.plan is not None + assert len(entry.plan) > 0 + + assert entry.available is not None + + # Ensure all three pages are read + assert m.call_count == 3 + assert m.mock.call_args_list[0].args[0] == "//regions/availability" + + assert ( + m.mock.call_args_list[1].args[0] + == "//regions/availability?page=2&page_size=100" + ) + assert ( + m.mock.call_args_list[2].args[0] + == "//regions/availability?page=3&page_size=100" + ) + + # Ensure the filter headers are correct + for k, call in m.mock.call_args_list: + assert json.loads(call.get("headers").get("X-Filter")) == { + "+and": [{"region": "us-east"}, {"plan": "premium4096.7"}] + } diff --git a/test/unit/groups/vpc_test.py b/test/unit/groups/vpc_test.py new file mode 100644 index 000000000..b63fc6d01 --- /dev/null +++ b/test/unit/groups/vpc_test.py @@ -0,0 +1,107 @@ +import datetime +from test.unit.base import ClientBaseCase + +from linode_api4 import DATE_FORMAT, VPC, VPCSubnet + + +class VPCTest(ClientBaseCase): + """ + Tests methods of the VPC Group + """ + + def test_create_vpc(self): + """ + Tests that you can create a VPC. + """ + + with self.mock_post("/vpcs/123456") as m: + vpc = self.client.vpcs.create("test-vpc", "us-southeast") + + self.assertEqual(m.call_url, "/vpcs") + + self.assertEqual( + m.call_data, + { + "label": "test-vpc", + "region": "us-southeast", + }, + ) + + self.assertEqual(vpc._populated, True) + self.validate_vpc_123456(vpc) + + def test_create_vpc_with_subnet(self): + """ + Tests that you can create a VPC. + """ + + with self.mock_post("/vpcs/123456") as m: + vpc = self.client.vpcs.create( + "test-vpc", + "us-southeast", + subnets=[{"label": "test-subnet", "ipv4": "10.0.0.0/24"}], + ) + + self.assertEqual(m.call_url, "/vpcs") + + self.assertEqual( + m.call_data, + { + "label": "test-vpc", + "region": "us-southeast", + "subnets": [ + {"label": "test-subnet", "ipv4": "10.0.0.0/24"} + ], + }, + ) + + self.assertEqual(vpc._populated, True) + self.validate_vpc_123456(vpc) + + def test_list_ips(self): + """ + Validates that all VPC IPs can be listed. + """ + + with self.mock_get("/vpcs/ips") as m: + result = self.client.vpcs.ips() + + assert m.call_url == "/vpcs/ips" + assert len(result) == 1 + + ip = result[0] + assert ip.address == "10.0.0.2" + assert ip.address_range == None + assert ip.vpc_id == 123 + assert ip.subnet_id == 456 + assert ip.region == "us-mia" + assert ip.linode_id == 123 + assert ip.config_id == 456 + assert ip.interface_id == 789 + assert ip.active + assert ip.nat_1_1 == "172.233.179.133" + assert ip.gateway == "10.0.0.1" + assert ip.prefix == 24 + assert ip.subnet_mask == "255.255.255.0" + + def validate_vpc_123456(self, vpc: VPC): + expected_dt = datetime.datetime.strptime( + "2018-01-01T00:01:01", DATE_FORMAT + ) + + self.assertEqual(vpc.label, "test-vpc") + self.assertEqual(vpc.description, "A very real VPC.") + self.assertEqual(vpc.region.id, "us-southeast") + self.assertEqual(vpc.created, expected_dt) + self.assertEqual(vpc.updated, expected_dt) + + def validate_vpc_subnet_789(self, subnet: VPCSubnet): + expected_dt = datetime.datetime.strptime( + "2018-01-01T00:01:01", DATE_FORMAT + ) + + self.assertEqual(subnet.label, "test-subnet") + self.assertEqual(subnet.ipv4, "10.0.0.0/24") + self.assertEqual(subnet.linodes[0].id, 12345) + self.assertEqual(subnet.created, expected_dt) + self.assertEqual(subnet.updated, expected_dt) diff --git a/test/unit/objects/database_test.py b/test/unit/objects/database_test.py index d5b84cebb..816e1cd95 100644 --- a/test/unit/objects/database_test.py +++ b/test/unit/objects/database_test.py @@ -4,130 +4,11 @@ from linode_api4.objects import MySQLDatabase -class DatabaseTest(ClientBaseCase): - """ - Tests methods of the DatabaseGroup class - """ - - def test_get_types(self): - """ - Test that database types are properly handled - """ - types = self.client.database.types() - - self.assertEqual(len(types), 1) - self.assertEqual(types[0].type_class, "nanode") - self.assertEqual(types[0].id, "g6-nanode-1") - self.assertEqual(types[0].engines.mysql[0].price.monthly, 20) - - def test_get_engines(self): - """ - Test that database engines are properly handled - """ - engines = self.client.database.engines() - - self.assertEqual(len(engines), 2) - - self.assertEqual(engines[0].engine, "mysql") - self.assertEqual(engines[0].id, "mysql/8.0.26") - self.assertEqual(engines[0].version, "8.0.26") - - self.assertEqual(engines[1].engine, "postgresql") - self.assertEqual(engines[1].id, "postgresql/10.14") - self.assertEqual(engines[1].version, "10.14") - - def test_get_databases(self): - """ - Test that databases are properly handled - """ - dbs = self.client.database.instances() - - self.assertEqual(len(dbs), 1) - self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") - self.assertEqual(dbs[0].cluster_size, 3) - self.assertEqual(dbs[0].encrypted, False) - self.assertEqual(dbs[0].engine, "mysql") - self.assertEqual( - dbs[0].hosts.primary, - "lin-123-456-mysql-mysql-primary.servers.linodedb.net", - ) - self.assertEqual( - dbs[0].hosts.secondary, - "lin-123-456-mysql-primary-private.servers.linodedb.net", - ) - self.assertEqual(dbs[0].id, 123) - self.assertEqual(dbs[0].region, "us-east") - self.assertEqual(dbs[0].updates.duration, 3) - self.assertEqual(dbs[0].version, "8.0.26") - - def test_database_instance(self): - """ - Ensures that the .instance attribute properly translates database types - """ - - dbs = self.client.database.instances() - db_translated = dbs[0].instance - - self.assertTrue(isinstance(db_translated, MySQLDatabase)) - self.assertEqual(db_translated.ssl_connection, True) - - class MySQLDatabaseTest(ClientBaseCase): """ Tests methods of the MySQLDatabase class """ - def test_get_instances(self): - """ - Test that database types are properly handled - """ - dbs = self.client.database.mysql_instances() - - self.assertEqual(len(dbs), 1) - self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") - self.assertEqual(dbs[0].cluster_size, 3) - self.assertEqual(dbs[0].encrypted, False) - self.assertEqual(dbs[0].engine, "mysql") - self.assertEqual( - dbs[0].hosts.primary, - "lin-123-456-mysql-mysql-primary.servers.linodedb.net", - ) - self.assertEqual( - dbs[0].hosts.secondary, - "lin-123-456-mysql-primary-private.servers.linodedb.net", - ) - self.assertEqual(dbs[0].id, 123) - self.assertEqual(dbs[0].region, "us-east") - self.assertEqual(dbs[0].updates.duration, 3) - self.assertEqual(dbs[0].version, "8.0.26") - - def test_create(self): - """ - Test that MySQL databases can be created - """ - - with self.mock_post("/databases/mysql/instances") as m: - # We don't care about errors here; we just want to - # validate the request. - try: - self.client.database.mysql_create( - "cool", - "us-southeast", - "mysql/8.0.26", - "g6-standard-1", - cluster_size=3, - ) - except Exception: - pass - - self.assertEqual(m.method, "post") - self.assertEqual(m.call_url, "/databases/mysql/instances") - self.assertEqual(m.call_data["label"], "cool") - self.assertEqual(m.call_data["region"], "us-southeast") - self.assertEqual(m.call_data["engine"], "mysql/8.0.26") - self.assertEqual(m.call_data["type"], "g6-standard-1") - self.assertEqual(m.call_data["cluster_size"], 3) - def test_update(self): """ Test that the MySQL database can be updated @@ -260,57 +141,6 @@ class PostgreSQLDatabaseTest(ClientBaseCase): Tests methods of the PostgreSQLDatabase class """ - def test_get_instances(self): - """ - Test that database types are properly handled - """ - dbs = self.client.database.postgresql_instances() - - self.assertEqual(len(dbs), 1) - self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") - self.assertEqual(dbs[0].cluster_size, 3) - self.assertEqual(dbs[0].encrypted, False) - self.assertEqual(dbs[0].engine, "postgresql") - self.assertEqual( - dbs[0].hosts.primary, - "lin-0000-000-pgsql-primary.servers.linodedb.net", - ) - self.assertEqual( - dbs[0].hosts.secondary, - "lin-0000-000-pgsql-primary-private.servers.linodedb.net", - ) - self.assertEqual(dbs[0].id, 123) - self.assertEqual(dbs[0].region, "us-east") - self.assertEqual(dbs[0].updates.duration, 3) - self.assertEqual(dbs[0].version, "13.2") - - def test_create(self): - """ - Test that PostgreSQL databases can be created - """ - - with self.mock_post("/databases/postgresql/instances") as m: - # We don't care about errors here; we just want to - # validate the request. - try: - self.client.database.postgresql_create( - "cool", - "us-southeast", - "postgresql/13.2", - "g6-standard-1", - cluster_size=3, - ) - except Exception: - pass - - self.assertEqual(m.method, "post") - self.assertEqual(m.call_url, "/databases/postgresql/instances") - self.assertEqual(m.call_data["label"], "cool") - self.assertEqual(m.call_data["region"], "us-southeast") - self.assertEqual(m.call_data["engine"], "postgresql/13.2") - self.assertEqual(m.call_data["type"], "g6-standard-1") - self.assertEqual(m.call_data["cluster_size"], 3) - def test_update(self): """ Test that the PostgreSQL database can be updated diff --git a/test/unit/objects/image_test.py b/test/unit/objects/image_test.py index 5d1ce42d5..0f4713bae 100644 --- a/test/unit/objects/image_test.py +++ b/test/unit/objects/image_test.py @@ -114,36 +114,6 @@ def put_mock(url: str, data: Optional[BinaryIO] = None, **kwargs): self.assertEqual(image.tags[0], "test_tag") self.assertEqual(image.tags[1], "test2") - def test_image_create_cloud_init(self): - """ - Test that an image can be created successfully with cloud-init. - """ - - with self.mock_post("images/private/123") as m: - self.client.images.create( - "Test Image", - "us-southeast", - description="very real image upload.", - cloud_init=True, - ) - - self.assertTrue(m.call_data["cloud_init"]) - - def test_image_create_upload_cloud_init(self): - """ - Test that an image upload URL can be created successfully with cloud-init. - """ - - with self.mock_post("images/upload") as m: - self.client.images.create_upload( - "Test Image", - "us-southeast", - description="very real image upload.", - cloud_init=True, - ) - - self.assertTrue(m.call_data["cloud_init"]) - def test_image_replication(self): """ Test that image can be replicated. diff --git a/test/unit/objects/linode_test.py b/test/unit/objects/linode_test.py index 44fea4f36..6016d2776 100644 --- a/test/unit/objects/linode_test.py +++ b/test/unit/objects/linode_test.py @@ -1,11 +1,7 @@ from datetime import datetime from test.unit.base import ClientBaseCase -from linode_api4 import ( - InstanceDiskEncryptionType, - InstancePlacementGroupAssignment, - NetworkInterface, -) +from linode_api4 import InstanceDiskEncryptionType, NetworkInterface from linode_api4.objects import ( Config, ConfigInterface, @@ -444,74 +440,6 @@ def test_create_disk(self): assert disk.id == 12345 assert disk.disk_encryption == InstanceDiskEncryptionType.disabled - def test_instance_create_with_user_data(self): - """ - Tests that the metadata field is populated on Linode create. - """ - - with self.mock_post("linode/instances/123") as m: - self.client.linode.instance_create( - "g6-nanode-1", - "us-southeast", - metadata=self.client.linode.build_instance_metadata( - user_data="cool" - ), - ) - - self.assertEqual( - m.call_data, - { - "region": "us-southeast", - "type": "g6-nanode-1", - "metadata": {"user_data": "Y29vbA=="}, - }, - ) - - def test_instance_create_with_interfaces(self): - """ - Tests that user can pass a list of interfaces on Linode create. - """ - interfaces = [ - {"purpose": "public"}, - ConfigInterface( - purpose="vlan", label="cool-vlan", ipam_address="10.0.0.4/32" - ), - ] - with self.mock_post("linode/instances/123") as m: - self.client.linode.instance_create( - "us-southeast", - "g6-nanode-1", - interfaces=interfaces, - ) - - self.assertEqual( - m.call_data["interfaces"], - [ - {"purpose": "public"}, - { - "purpose": "vlan", - "label": "cool-vlan", - "ipam_address": "10.0.0.4/32", - }, - ], - ) - - def test_build_instance_metadata(self): - """ - Tests that the metadata field is built correctly. - """ - self.assertEqual( - self.client.linode.build_instance_metadata(user_data="cool"), - {"user_data": "Y29vbA=="}, - ) - - self.assertEqual( - self.client.linode.build_instance_metadata( - user_data="cool", encode_user_data=False - ), - {"user_data": "cool"}, - ) - def test_get_placement_group(self): """ Tests that you can get the placement group for a Linode @@ -535,25 +463,6 @@ def test_get_placement_group(self): assert pg.label == "test" assert pg.placement_group_type == "anti_affinity:local" - def test_create_with_placement_group(self): - """ - Tests that you can create a Linode with a Placement Group - """ - - with self.mock_post("linode/instances/123") as m: - self.client.linode.instance_create( - "g6-nanode-1", - "eu-west", - placement_group=InstancePlacementGroupAssignment( - id=123, - compliant_only=True, - ), - ) - - self.assertEqual( - m.call_data["placement_group"], {"id": 123, "compliant_only": True} - ) - class DiskTest(ClientBaseCase): """ @@ -663,24 +572,6 @@ def test_get_stackscript(self): class TypeTest(ClientBaseCase): - def test_get_types(self): - """ - Tests that Linode types can be returned - """ - types = self.client.linode.types() - - self.assertEqual(len(types), 5) - for t in types: - self.assertTrue(t._populated) - self.assertIsNotNone(t.id) - self.assertIsNotNone(t.label) - self.assertIsNotNone(t.disk) - self.assertIsNotNone(t.type_class) - self.assertIsNotNone(t.gpus) - self.assertIsNone(t.successor) - self.assertIsNotNone(t.region_prices) - self.assertIsNotNone(t.addons.backups.region_prices) - self.assertIsNotNone(t.accelerated_devices) def test_get_type_by_id(self): """ diff --git a/test/unit/objects/lke_test.py b/test/unit/objects/lke_test.py index 100f36487..6541f3826 100644 --- a/test/unit/objects/lke_test.py +++ b/test/unit/objects/lke_test.py @@ -7,7 +7,6 @@ LKECluster, LKEClusterControlPlaneACLAddressesOptions, LKEClusterControlPlaneACLOptions, - LKEClusterControlPlaneOptions, LKENodePool, ) from linode_api4.objects.lke import LKENodePoolNode, LKENodePoolTaint @@ -166,36 +165,6 @@ def test_load_node_pool(self): self.assertIsNotNone(pool.autoscaler) self.assertIsNotNone(pool.tags) - def test_cluster_create_with_acl(self): - """ - Tests that an LKE cluster can be created with a control plane ACL configuration. - """ - - with self.mock_post("lke/clusters") as m: - self.client.lke.cluster_create( - "us-mia", - "test-acl-cluster", - [self.client.lke.node_pool("g6-nanode-1", 3)], - "1.29", - control_plane=LKEClusterControlPlaneOptions( - acl=LKEClusterControlPlaneACLOptions( - enabled=True, - addresses=LKEClusterControlPlaneACLAddressesOptions( - ipv4=["10.0.0.1/32"], ipv6=["1234::5678"] - ), - ) - ), - ) - - assert "high_availability" not in m.call_data["control_plane"] - assert m.call_data["control_plane"]["acl"]["enabled"] - assert m.call_data["control_plane"]["acl"]["addresses"]["ipv4"] == [ - "10.0.0.1/32" - ] - assert m.call_data["control_plane"]["acl"]["addresses"]["ipv6"] == [ - "1234::5678" - ] - def test_cluster_get_acl(self): """ Tests that an LKE cluster can be created with a control plane ACL configuration. diff --git a/test/unit/objects/placement_test.py b/test/unit/objects/placement_test.py index 4e5960e7b..08fcdc1e4 100644 --- a/test/unit/objects/placement_test.py +++ b/test/unit/objects/placement_test.py @@ -1,11 +1,9 @@ from test.unit.base import ClientBaseCase -from linode_api4 import PlacementGroupPolicy from linode_api4.objects import ( MigratedInstance, PlacementGroup, PlacementGroupMember, - PlacementGroupType, ) @@ -25,46 +23,6 @@ def test_get_placement_group(self): self.validate_pg_123(pg) assert pg._populated - def test_list_pgs(self): - """ - Tests that you can list PGs. - """ - - pgs = self.client.placement.groups() - - self.validate_pg_123(pgs[0]) - assert pgs[0]._populated - - def test_create_pg(self): - """ - Tests that you can create a Placement Group. - """ - - with self.mock_post("/placement/groups/123") as m: - pg = self.client.placement.group_create( - "test", - "eu-west", - PlacementGroupType.anti_affinity_local, - PlacementGroupPolicy.strict, - ) - - assert m.call_url == "/placement/groups" - - self.assertEqual( - m.call_data, - { - "label": "test", - "region": "eu-west", - "placement_group_type": str( - PlacementGroupType.anti_affinity_local - ), - "placement_group_policy": PlacementGroupPolicy.strict, - }, - ) - - assert pg._populated - self.validate_pg_123(pg) - def test_pg_assign(self): """ Tests that you can assign to a PG. diff --git a/test/unit/objects/region_test.py b/test/unit/objects/region_test.py index a7fcc2694..0bc1afa9e 100644 --- a/test/unit/objects/region_test.py +++ b/test/unit/objects/region_test.py @@ -1,8 +1,6 @@ -import json from test.unit.base import ClientBaseCase from linode_api4.objects import Region -from linode_api4.objects.region import RegionAvailabilityEntry class RegionTest(ClientBaseCase): @@ -30,47 +28,6 @@ def test_get_region(self): region.placement_group_limits.maximum_linodes_per_pg, 5 ) - def test_list_availability(self): - """ - Tests that region availability can be listed and filtered on. - """ - - with self.mock_get("/regions/availability") as m: - avail_entries = self.client.regions.availability( - RegionAvailabilityEntry.filters.region == "us-east", - RegionAvailabilityEntry.filters.plan == "premium4096.7", - ) - - assert len(avail_entries) > 0 - - for entry in avail_entries: - assert entry.region is not None - assert len(entry.region) > 0 - - assert entry.plan is not None - assert len(entry.plan) > 0 - - assert entry.available is not None - - # Ensure all three pages are read - assert m.call_count == 3 - assert m.mock.call_args_list[0].args[0] == "//regions/availability" - - assert ( - m.mock.call_args_list[1].args[0] - == "//regions/availability?page=2&page_size=100" - ) - assert ( - m.mock.call_args_list[2].args[0] - == "//regions/availability?page=3&page_size=100" - ) - - # Ensure the filter headers are correct - for k, call in m.mock.call_args_list: - assert json.loads(call.get("headers").get("X-Filter")) == { - "+and": [{"region": "us-east"}, {"plan": "premium4096.7"}] - } - def test_region_availability(self): """ Tests that availability for a specific region can be listed and filtered on. diff --git a/test/unit/objects/vpc_test.py b/test/unit/objects/vpc_test.py index 4d80716d4..55ebe395a 100644 --- a/test/unit/objects/vpc_test.py +++ b/test/unit/objects/vpc_test.py @@ -30,55 +30,6 @@ def test_list_vpcs(self): self.validate_vpc_123456(vpcs[0]) self.assertEqual(vpcs[0]._populated, True) - def test_create_vpc(self): - """ - Tests that you can create a VPC. - """ - - with self.mock_post("/vpcs/123456") as m: - vpc = self.client.vpcs.create("test-vpc", "us-southeast") - - self.assertEqual(m.call_url, "/vpcs") - - self.assertEqual( - m.call_data, - { - "label": "test-vpc", - "region": "us-southeast", - }, - ) - - self.assertEqual(vpc._populated, True) - self.validate_vpc_123456(vpc) - - def test_create_vpc_with_subnet(self): - """ - Tests that you can create a VPC. - """ - - with self.mock_post("/vpcs/123456") as m: - vpc = self.client.vpcs.create( - "test-vpc", - "us-southeast", - subnets=[{"label": "test-subnet", "ipv4": "10.0.0.0/24"}], - ) - - self.assertEqual(m.call_url, "/vpcs") - - self.assertEqual( - m.call_data, - { - "label": "test-vpc", - "region": "us-southeast", - "subnets": [ - {"label": "test-subnet", "ipv4": "10.0.0.0/24"} - ], - }, - ) - - self.assertEqual(vpc._populated, True) - self.validate_vpc_123456(vpc) - def test_get_subnet(self): """ Tests that you can list VPCs. @@ -125,32 +76,6 @@ def test_create_subnet(self): self.validate_vpc_subnet_789(subnet) - def test_list_ips(self): - """ - Validates that all VPC IPs can be listed. - """ - - with self.mock_get("/vpcs/ips") as m: - result = self.client.vpcs.ips() - - assert m.call_url == "/vpcs/ips" - assert len(result) == 1 - - ip = result[0] - assert ip.address == "10.0.0.2" - assert ip.address_range == None - assert ip.vpc_id == 123 - assert ip.subnet_id == 456 - assert ip.region == "us-mia" - assert ip.linode_id == 123 - assert ip.config_id == 456 - assert ip.interface_id == 789 - assert ip.active - assert ip.nat_1_1 == "172.233.179.133" - assert ip.gateway == "10.0.0.1" - assert ip.prefix == 24 - assert ip.subnet_mask == "255.255.255.0" - def validate_vpc_123456(self, vpc: VPC): expected_dt = datetime.datetime.strptime( "2018-01-01T00:01:01", DATE_FORMAT From e72c643cd382d2fa62b829513edd6f77067ec6f2 Mon Sep 17 00:00:00 2001 From: ykim-1 Date: Mon, 27 Jan 2025 12:25:33 -0800 Subject: [PATCH 2/3] address codeQL warnings --- test/unit/groups/database_test.py | 15 ++- test/unit/groups/vpc_test.py | 2 +- test/unit/objects/database_test.py | 185 ++++++++++++++++++++++++++++- 3 files changed, 193 insertions(+), 9 deletions(-) diff --git a/test/unit/groups/database_test.py b/test/unit/groups/database_test.py index 446ed3617..09d842b77 100644 --- a/test/unit/groups/database_test.py +++ b/test/unit/groups/database_test.py @@ -1,7 +1,10 @@ +import logging from test.unit.base import ClientBaseCase from linode_api4.objects import MySQLDatabase +logger = logging.getLogger(__name__) + class DatabaseTest(ClientBaseCase): """ @@ -116,8 +119,10 @@ def test_create(self): "g6-standard-1", cluster_size=3, ) - except Exception: - pass + except Exception as e: + logger.warning( + "An error occurred while validating the request: %s", e + ) self.assertEqual(m.method, "post") self.assertEqual(m.call_url, "/databases/mysql/instances") @@ -173,8 +178,10 @@ def test_create(self): "g6-standard-1", cluster_size=3, ) - except Exception: - pass + except Exception as e: + logger.warning( + "An error occurred while validating the request: %s", e + ) self.assertEqual(m.method, "post") self.assertEqual(m.call_url, "/databases/postgresql/instances") diff --git a/test/unit/groups/vpc_test.py b/test/unit/groups/vpc_test.py index b63fc6d01..7b8c985d2 100644 --- a/test/unit/groups/vpc_test.py +++ b/test/unit/groups/vpc_test.py @@ -71,7 +71,7 @@ def test_list_ips(self): ip = result[0] assert ip.address == "10.0.0.2" - assert ip.address_range == None + assert ip.address_range is None assert ip.vpc_id == 123 assert ip.subnet_id == 456 assert ip.region == "us-mia" diff --git a/test/unit/objects/database_test.py b/test/unit/objects/database_test.py index 816e1cd95..f875a0ed3 100644 --- a/test/unit/objects/database_test.py +++ b/test/unit/objects/database_test.py @@ -1,14 +1,136 @@ +import logging from test.unit.base import ClientBaseCase from linode_api4 import PostgreSQLDatabase from linode_api4.objects import MySQLDatabase +logger = logging.getLogger(__name__) + + +class DatabaseTest(ClientBaseCase): + """ + Tests methods of the DatabaseGroup class + """ + + def test_get_types(self): + """ + Test that database types are properly handled + """ + types = self.client.database.types() + + self.assertEqual(len(types), 1) + self.assertEqual(types[0].type_class, "nanode") + self.assertEqual(types[0].id, "g6-nanode-1") + self.assertEqual(types[0].engines.mysql[0].price.monthly, 20) + + def test_get_engines(self): + """ + Test that database engines are properly handled + """ + engines = self.client.database.engines() + + self.assertEqual(len(engines), 2) + + self.assertEqual(engines[0].engine, "mysql") + self.assertEqual(engines[0].id, "mysql/8.0.26") + self.assertEqual(engines[0].version, "8.0.26") + + self.assertEqual(engines[1].engine, "postgresql") + self.assertEqual(engines[1].id, "postgresql/10.14") + self.assertEqual(engines[1].version, "10.14") + + def test_get_databases(self): + """ + Test that databases are properly handled + """ + dbs = self.client.database.instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "mysql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-123-456-mysql-mysql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-123-456-mysql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "8.0.26") + + def test_database_instance(self): + """ + Ensures that the .instance attribute properly translates database types + """ + + dbs = self.client.database.instances() + db_translated = dbs[0].instance + + self.assertTrue(isinstance(db_translated, MySQLDatabase)) + self.assertEqual(db_translated.ssl_connection, True) + class MySQLDatabaseTest(ClientBaseCase): """ Tests methods of the MySQLDatabase class """ + def test_get_instances(self): + """ + Test that database types are properly handled + """ + dbs = self.client.database.mysql_instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "mysql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-123-456-mysql-mysql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-123-456-mysql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "8.0.26") + + def test_create(self): + """ + Test that MySQL databases can be created + """ + + with self.mock_post("/databases/mysql/instances") as m: + # We don't care about errors here; we just want to + # validate the request. + try: + self.client.database.mysql_create( + "cool", + "us-southeast", + "mysql/8.0.26", + "g6-standard-1", + cluster_size=3, + ) + except Exception: + pass + + self.assertEqual(m.method, "post") + self.assertEqual(m.call_url, "/databases/mysql/instances") + self.assertEqual(m.call_data["label"], "cool") + self.assertEqual(m.call_data["region"], "us-southeast") + self.assertEqual(m.call_data["engine"], "mysql/8.0.26") + self.assertEqual(m.call_data["type"], "g6-standard-1") + self.assertEqual(m.call_data["cluster_size"], 3) + def test_update(self): """ Test that the MySQL database can be updated @@ -59,8 +181,10 @@ def test_create_backup(self): # validate the request. try: db.backup_create("mybackup", target="secondary") - except Exception: - pass + except Exception as e: + logger.warning( + "An error occurred while validating the request: %s", e + ) self.assertEqual(m.method, "post") self.assertEqual( @@ -141,6 +265,57 @@ class PostgreSQLDatabaseTest(ClientBaseCase): Tests methods of the PostgreSQLDatabase class """ + def test_get_instances(self): + """ + Test that database types are properly handled + """ + dbs = self.client.database.postgresql_instances() + + self.assertEqual(len(dbs), 1) + self.assertEqual(dbs[0].allow_list[1], "192.0.1.0/24") + self.assertEqual(dbs[0].cluster_size, 3) + self.assertEqual(dbs[0].encrypted, False) + self.assertEqual(dbs[0].engine, "postgresql") + self.assertEqual( + dbs[0].hosts.primary, + "lin-0000-000-pgsql-primary.servers.linodedb.net", + ) + self.assertEqual( + dbs[0].hosts.secondary, + "lin-0000-000-pgsql-primary-private.servers.linodedb.net", + ) + self.assertEqual(dbs[0].id, 123) + self.assertEqual(dbs[0].region, "us-east") + self.assertEqual(dbs[0].updates.duration, 3) + self.assertEqual(dbs[0].version, "13.2") + + def test_create(self): + """ + Test that PostgreSQL databases can be created + """ + + with self.mock_post("/databases/postgresql/instances") as m: + # We don't care about errors here; we just want to + # validate the request. + try: + self.client.database.postgresql_create( + "cool", + "us-southeast", + "postgresql/13.2", + "g6-standard-1", + cluster_size=3, + ) + except Exception: + pass + + self.assertEqual(m.method, "post") + self.assertEqual(m.call_url, "/databases/postgresql/instances") + self.assertEqual(m.call_data["label"], "cool") + self.assertEqual(m.call_data["region"], "us-southeast") + self.assertEqual(m.call_data["engine"], "postgresql/13.2") + self.assertEqual(m.call_data["type"], "g6-standard-1") + self.assertEqual(m.call_data["cluster_size"], 3) + def test_update(self): """ Test that the PostgreSQL database can be updated @@ -191,8 +366,10 @@ def test_create_backup(self): # validate the request. try: db.backup_create("mybackup", target="secondary") - except Exception: - pass + except Exception as e: + logger.warning( + "An error occurred while validating the request: %s", e + ) self.assertEqual(m.method, "post") self.assertEqual( From b9eb75b3d502c046225e5a999e8cf0bf3eda52b4 Mon Sep 17 00:00:00 2001 From: ykim-1 Date: Tue, 4 Feb 2025 09:54:59 -0800 Subject: [PATCH 3/3] fix import --- test/unit/objects/lke_test.py | 1 + 1 file changed, 1 insertion(+) diff --git a/test/unit/objects/lke_test.py b/test/unit/objects/lke_test.py index a74b817bf..1a39b69bc 100644 --- a/test/unit/objects/lke_test.py +++ b/test/unit/objects/lke_test.py @@ -7,6 +7,7 @@ LKECluster, LKEClusterControlPlaneACLAddressesOptions, LKEClusterControlPlaneACLOptions, + LKEClusterControlPlaneOptions, LKENodePool, ) from linode_api4.objects.lke import LKENodePoolNode, LKENodePoolTaint