diff --git a/linode_api4/groups/__init__.py b/linode_api4/groups/__init__.py index 6f87eeb65..3c1bc9a7f 100644 --- a/linode_api4/groups/__init__.py +++ b/linode_api4/groups/__init__.py @@ -9,6 +9,7 @@ from .linode import * from .lke import * from .lke_tier import * +from .lock import * from .longview import * from .maintenance import * from .monitor import * diff --git a/linode_api4/groups/account.py b/linode_api4/groups/account.py index 6f8c6528e..da356e999 100644 --- a/linode_api4/groups/account.py +++ b/linode_api4/groups/account.py @@ -11,6 +11,7 @@ ChildAccount, Event, Invoice, + Lock, Login, MappedObject, OAuthClient, @@ -510,3 +511,58 @@ def child_accounts(self, *filters): :rtype: PaginatedList of ChildAccount """ return self.client._get_and_filter(ChildAccount, *filters) + + def locks(self, *filters): + """ + Returns a list of all resource locks on the account. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-locks + + :param filters: Any number of filters to apply to this query. + See :doc:`Filtering Collections` + for more details on filtering. + + :returns: A list of resource locks on the account. + :rtype: PaginatedList of Lock + """ + return self.client._get_and_filter(Lock, *filters) + + def lock_create(self, entity_type, entity_id, lock_type, **kwargs): + """ + Creates a resource lock to prevent deletion or modification. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-lock + + :param entity_type: The type of entity to lock (e.g., "linode"). + :type entity_type: str + :param entity_id: The ID of the entity to lock. + :type entity_id: int + :param lock_type: The type of lock (e.g., "cannot_delete"). + :type lock_type: str or LockType + + :returns: The created resource lock. + :rtype: Lock + """ + from linode_api4.objects.lock import ( # pylint: disable=import-outside-toplevel + LockType, + ) + + params = { + "entity_type": entity_type, + "entity_id": entity_id, + "lock_type": ( + lock_type.value + if isinstance(lock_type, LockType) + else lock_type + ), + } + params.update(kwargs) + + result = self.client.post("/locks", data=params) + + if "id" not in result: + raise UnexpectedResponseError( + "Unexpected response when creating lock!", json=result + ) + + return Lock(self.client, result["id"], result) diff --git a/linode_api4/groups/lock.py b/linode_api4/groups/lock.py new file mode 100644 index 000000000..c288da9b6 --- /dev/null +++ b/linode_api4/groups/lock.py @@ -0,0 +1,89 @@ +from typing import Union + +from linode_api4.errors import UnexpectedResponseError +from linode_api4.groups import Group +from linode_api4.objects import Lock, LockType + +__all__ = ["LockGroup"] + + +class LockGroup(Group): + """ + Encapsulates methods for interacting with Resource Locks. + + Resource locks prevent deletion or modification of resources. + Currently, only Linode instances can be locked. + """ + + def __call__(self, *filters): + """ + Returns a list of all Resource Locks on the account. + + This is intended to be called off of the :any:`LinodeClient` + class, like this:: + + locks = client.locks() + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-locks + + :param filters: Any number of filters to apply to this query. + See :doc:`Filtering Collections` + for more details on filtering. + + :returns: A list of Resource Locks on the account. + :rtype: PaginatedList of Lock + """ + return self.client._get_and_filter(Lock, *filters) + + def create( + self, + entity_type: str, + entity_id: Union[int, str], + lock_type: Union[LockType, str] = LockType.cannot_delete, + ) -> Lock: + """ + Creates a new Resource Lock for the specified entity. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-lock + + :param entity_type: The type of entity to lock (e.g., "linode"). + :type entity_type: str + :param entity_id: The ID of the entity to lock. + :type entity_id: int or str + :param lock_type: The type of lock to create. Defaults to "cannot_delete". + :type lock_type: LockType or str + + :returns: The newly created Resource Lock. + :rtype: Lock + """ + params = { + "entity_type": entity_type, + "entity_id": entity_id, + "lock_type": lock_type, + } + + result = self.client.post("/locks", data=params) + + if "id" not in result: + raise UnexpectedResponseError( + "Unexpected response when creating lock!", json=result + ) + + return Lock(self.client, result["id"], result) + + def delete(self, lock: Union[Lock, int]) -> bool: + """ + Deletes a Resource Lock. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/delete-lock + + :param lock: The Lock to delete, or its ID. + :type lock: Lock or int + + :returns: True if the lock was successfully deleted. + :rtype: bool + """ + lock_id = lock.id if isinstance(lock, Lock) else lock + + self.client.delete(f"/locks/{lock_id}") + return True diff --git a/linode_api4/linode_client.py b/linode_api4/linode_client.py index 1d9f0bba4..73a33e6a4 100644 --- a/linode_api4/linode_client.py +++ b/linode_api4/linode_client.py @@ -18,6 +18,7 @@ ImageGroup, LinodeGroup, LKEGroup, + LockGroup, LongviewGroup, MaintenanceGroup, MetricsGroup, @@ -454,6 +455,9 @@ def __init__( self.monitor = MonitorGroup(self) + #: Access methods related to Resource Locks - See :any:`LockGroup` for more information. + self.locks = LockGroup(self) + super().__init__( token=token, base_url=base_url, diff --git a/linode_api4/objects/__init__.py b/linode_api4/objects/__init__.py index 9f120310c..98d1c7a7d 100644 --- a/linode_api4/objects/__init__.py +++ b/linode_api4/objects/__init__.py @@ -24,3 +24,4 @@ from .placement import * from .monitor import * from .monitor_api import * +from .lock import * diff --git a/linode_api4/objects/linode.py b/linode_api4/objects/linode.py index df2694f66..fae0926d5 100644 --- a/linode_api4/objects/linode.py +++ b/linode_api4/objects/linode.py @@ -803,6 +803,7 @@ class Instance(Base): "maintenance_policy": Property( mutable=True ), # Note: This field is only available when using v4beta. + "locks": Property(unordered=True), } @property diff --git a/linode_api4/objects/lock.py b/linode_api4/objects/lock.py new file mode 100644 index 000000000..b6552da7b --- /dev/null +++ b/linode_api4/objects/lock.py @@ -0,0 +1,47 @@ +from dataclasses import dataclass + +from linode_api4.objects.base import Base, Property +from linode_api4.objects.serializable import JSONObject, StrEnum + +__all__ = ["LockType", "LockEntity", "Lock"] + + +class LockType(StrEnum): + """ + LockType defines valid values for resource lock types. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-lock + """ + + cannot_delete = "cannot_delete" + cannot_delete_with_subresources = "cannot_delete_with_subresources" + + +@dataclass +class LockEntity(JSONObject): + """ + Represents the entity that is locked. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-lock + """ + + id: int = 0 + type: str = "" + label: str = "" + url: str = "" + + +class Lock(Base): + """ + A resource lock that prevents deletion or modification of a resource. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-lock + """ + + api_endpoint = "/locks/{id}" + + properties = { + "id": Property(identifier=True), + "lock_type": Property(), + "entity": Property(json_object=LockEntity), + } diff --git a/test/fixtures/locks.json b/test/fixtures/locks.json new file mode 100644 index 000000000..b84056b6b --- /dev/null +++ b/test/fixtures/locks.json @@ -0,0 +1,27 @@ +{ + "data": [ + { + "id": 1, + "lock_type": "cannot_delete", + "entity": { + "id": 123, + "type": "linode", + "label": "test-linode", + "url": "/v4/linode/instances/123" + } + }, + { + "id": 2, + "lock_type": "cannot_delete_with_subresources", + "entity": { + "id": 456, + "type": "linode", + "label": "another-linode", + "url": "/v4/linode/instances/456" + } + } + ], + "page": 1, + "pages": 1, + "results": 2 +} diff --git a/test/fixtures/locks_1.json b/test/fixtures/locks_1.json new file mode 100644 index 000000000..ed7a802bf --- /dev/null +++ b/test/fixtures/locks_1.json @@ -0,0 +1,10 @@ +{ + "id": 1, + "lock_type": "cannot_delete", + "entity": { + "id": 123, + "type": "linode", + "label": "test-linode", + "url": "/v4/linode/instances/123" + } +} diff --git a/test/integration/models/lock/__init__.py b/test/integration/models/lock/__init__.py new file mode 100644 index 000000000..1e07a34ee --- /dev/null +++ b/test/integration/models/lock/__init__.py @@ -0,0 +1 @@ +# This file is intentionally left empty to make the directory a Python package. diff --git a/test/integration/models/lock/test_lock.py b/test/integration/models/lock/test_lock.py new file mode 100644 index 000000000..5c2876693 --- /dev/null +++ b/test/integration/models/lock/test_lock.py @@ -0,0 +1,199 @@ +from test.integration.conftest import get_region +from test.integration.helpers import ( + get_test_label, + send_request_when_resource_available, +) + +import pytest + +from linode_api4.objects import Lock, LockType + + +@pytest.fixture(scope="function") +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") + label = get_test_label(length=8) + + linode_instance, _ = client.linode.instance_create( + "g6-nanode-1", + region, + image="linode/debian12", + label=label, + firewall=e2e_test_firewall, + ) + + yield linode_instance + + # Clean up any locks on the Linode before deleting it + locks = client.locks() + for lock in locks: + if ( + lock.entity.id == linode_instance.id + and lock.entity.type == "linode" + ): + lock.delete() + + send_request_when_resource_available( + timeout=100, func=linode_instance.delete + ) + + +@pytest.fixture(scope="function") +def test_lock(test_linode_client, linode_for_lock): + """ + Create a lock for testing. + """ + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete, + ) + + yield lock + + # Clean up lock if it still exists + try: + lock.delete() + except Exception: + pass # Lock may have been deleted by the test + + +@pytest.mark.smoke +def test_get_lock(test_linode_client, test_lock): + """ + Test that a lock can be retrieved by ID. + """ + lock = test_linode_client.load(Lock, test_lock.id) + + assert lock.id == test_lock.id + assert lock.lock_type == "cannot_delete" + assert lock.entity is not None + assert lock.entity.type == "linode" + + +def test_list_locks(test_linode_client, test_lock): + """ + Test that locks can be listed. + """ + locks = test_linode_client.locks() + + assert len(locks) > 0 + + # Verify our test lock is in the list + lock_ids = [lock.id for lock in locks] + assert test_lock.id in lock_ids + + +def test_create_lock_cannot_delete(test_linode_client, linode_for_lock): + """ + Test creating a cannot_delete lock. + """ + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete, + ) + + assert lock.id is not None + assert lock.lock_type == "cannot_delete" + assert lock.entity.id == linode_for_lock.id + assert lock.entity.type == "linode" + assert lock.entity.label == linode_for_lock.label + + # Clean up + lock.delete() + + +def test_create_lock_cannot_delete_with_subresources( + test_linode_client, linode_for_lock +): + """ + Test creating a cannot_delete_with_subresources lock. + """ + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete_with_subresources, + ) + + assert lock.id is not None + assert lock.lock_type == "cannot_delete_with_subresources" + assert lock.entity.id == linode_for_lock.id + assert lock.entity.type == "linode" + + # Clean up + lock.delete() + + +def test_delete_lock(test_linode_client, linode_for_lock): + """ + Test that a lock can be deleted. + """ + # Create a lock + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete, + ) + + lock_id = lock.id + + # Delete the lock using the group method + result = test_linode_client.locks.delete(lock) + + assert result is True + + # Verify the lock no longer exists + locks = test_linode_client.locks() + lock_ids = [l.id for l in locks] + assert lock_id not in lock_ids + + +def test_delete_lock_by_id(test_linode_client, linode_for_lock): + """ + Test that a lock can be deleted by ID. + """ + # Create a lock + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete, + ) + + lock_id = lock.id + + # Delete the lock by ID + result = test_linode_client.locks.delete(lock_id) + + assert result is True + + # Verify the lock no longer exists + locks = test_linode_client.locks() + lock_ids = [l.id for l in locks] + assert lock_id not in lock_ids + + +def test_lock_object_delete(test_linode_client, linode_for_lock): + """ + Test that a lock can be deleted using the Lock object's delete method. + """ + # Create a lock + lock = test_linode_client.locks.create( + entity_type="linode", + entity_id=linode_for_lock.id, + lock_type=LockType.cannot_delete, + ) + + lock_id = lock.id + + # Delete the lock using the object method + lock.delete() + + # Verify the lock no longer exists + locks = test_linode_client.locks() + lock_ids = [l.id for l in locks] + assert lock_id not in lock_ids diff --git a/test/unit/groups/lock_test.py b/test/unit/groups/lock_test.py new file mode 100644 index 000000000..158525317 --- /dev/null +++ b/test/unit/groups/lock_test.py @@ -0,0 +1,97 @@ +from test.unit.base import ClientBaseCase + +from linode_api4.objects import Lock, LockType + + +class LockGroupTest(ClientBaseCase): + """ + Tests methods of the LockGroup class + """ + + def test_list_locks(self): + """ + Tests that locks can be retrieved using client.locks() + """ + locks = self.client.locks() + + self.assertEqual(len(locks), 2) + self.assertEqual(locks[0].id, 1) + self.assertEqual(locks[0].lock_type, "cannot_delete") + self.assertEqual(locks[0].entity.id, 123) + self.assertEqual(locks[0].entity.type, "linode") + self.assertEqual(locks[1].id, 2) + self.assertEqual(locks[1].lock_type, "cannot_delete_with_subresources") + self.assertEqual(locks[1].entity.id, 456) + + def test_create_lock(self): + """ + Tests that a lock can be created using client.locks.create() + """ + with self.mock_post("/locks/1") as m: + lock = self.client.locks.create( + entity_type="linode", + entity_id=123, + lock_type=LockType.cannot_delete, + ) + + self.assertEqual(m.call_url, "/locks") + self.assertEqual(m.call_data["entity_type"], "linode") + self.assertEqual(m.call_data["entity_id"], 123) + self.assertEqual(m.call_data["lock_type"], "cannot_delete") + + self.assertEqual(lock.id, 1) + self.assertEqual(lock.lock_type, "cannot_delete") + self.assertIsNotNone(lock.entity) + self.assertEqual(lock.entity.id, 123) + + def test_create_lock_with_subresources(self): + """ + Tests that a lock with subresources can be created + """ + with self.mock_post("/locks/1") as m: + lock = self.client.locks.create( + entity_type="linode", + entity_id=456, + lock_type=LockType.cannot_delete_with_subresources, + ) + + self.assertEqual(m.call_url, "/locks") + self.assertEqual(m.call_data["entity_type"], "linode") + self.assertEqual(m.call_data["entity_id"], 456) + self.assertEqual( + m.call_data["lock_type"], "cannot_delete_with_subresources" + ) + + def test_create_lock_default_type(self): + """ + Tests that creating a lock without specifying lock_type uses cannot_delete + """ + with self.mock_post("/locks/1") as m: + self.client.locks.create( + entity_type="linode", + entity_id=123, + ) + + self.assertEqual(m.call_data["lock_type"], "cannot_delete") + + def test_delete_lock_by_object(self): + """ + Tests that a lock can be deleted using client.locks.delete() with a Lock object + """ + lock = Lock(self.client, 1) + + with self.mock_delete() as m: + result = self.client.locks.delete(lock) + + self.assertTrue(result) + self.assertEqual(m.call_url, "/locks/1") + + def test_delete_lock_by_id(self): + """ + Tests that a lock can be deleted using client.locks.delete() with an ID + """ + with self.mock_delete() as m: + result = self.client.locks.delete(123) + + self.assertTrue(result) + self.assertEqual(m.call_url, "/locks/123") diff --git a/test/unit/objects/lock_test.py b/test/unit/objects/lock_test.py new file mode 100644 index 000000000..b50b12d13 --- /dev/null +++ b/test/unit/objects/lock_test.py @@ -0,0 +1,63 @@ +from test.unit.base import ClientBaseCase + +from linode_api4.objects.lock import Lock, LockEntity, LockType + + +class LockTest(ClientBaseCase): + """ + Tests methods of the Lock class + """ + + def test_get_lock(self): + """ + Tests that a lock is loaded correctly by ID + """ + lock = Lock(self.client, 1) + + self.assertEqual(lock.id, 1) + self.assertEqual(lock.lock_type, "cannot_delete") + self.assertIsInstance(lock.entity, LockEntity) + self.assertEqual(lock.entity.id, 123) + self.assertEqual(lock.entity.type, "linode") + self.assertEqual(lock.entity.label, "test-linode") + self.assertEqual(lock.entity.url, "/v4/linode/instances/123") + + def test_get_locks(self): + """ + Tests that locks can be retrieved + """ + locks = self.client.account.locks() + + self.assertEqual(len(locks), 2) + self.assertEqual(locks[0].id, 1) + self.assertEqual(locks[0].lock_type, "cannot_delete") + self.assertEqual(locks[1].id, 2) + self.assertEqual(locks[1].lock_type, "cannot_delete_with_subresources") + + def test_create_lock(self): + """ + Tests that a lock can be created + """ + with self.mock_post("/locks/1") as m: + lock = self.client.account.lock_create( + "linode", 123, LockType.cannot_delete + ) + + self.assertEqual(m.call_url, "/locks") + self.assertEqual(m.call_data["entity_type"], "linode") + self.assertEqual(m.call_data["entity_id"], 123) + self.assertEqual(m.call_data["lock_type"], "cannot_delete") + + self.assertEqual(lock.id, 1) + self.assertEqual(lock.lock_type, "cannot_delete") + + def test_delete_lock(self): + """ + Tests that a lock can be deleted + """ + lock = Lock(self.client, 1) + + with self.mock_delete() as m: + lock.delete() + + self.assertEqual(m.call_url, "/locks/1")