Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions linode_api4/groups/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 *
Expand Down
72 changes: 72 additions & 0 deletions linode_api4/groups/lock.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
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: TBD
:param filters: Any number of filters to apply to this query.
See :doc:`Filtering Collections</linode_api4/objects/filtering>`
for more details on filtering.
:returns: A list of 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: TBD
: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 | str
:param lock_type: The type of lock to create. Defaults to "cannot_delete".
:type lock_type: LockType | 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)
4 changes: 4 additions & 0 deletions linode_api4/linode_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
ImageGroup,
LinodeGroup,
LKEGroup,
LockGroup,
LongviewGroup,
MaintenanceGroup,
MetricsGroup,
Expand Down Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions linode_api4/objects/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,3 +24,4 @@
from .placement import *
from .monitor import *
from .monitor_api import *
from .lock import *
1 change: 1 addition & 0 deletions linode_api4/objects/linode.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
47 changes: 47 additions & 0 deletions linode_api4/objects/lock.py
Original file line number Diff line number Diff line change
@@ -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),
}
27 changes: 27 additions & 0 deletions test/fixtures/locks.json
Original file line number Diff line number Diff line change
@@ -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
}
10 changes: 10 additions & 0 deletions test/fixtures/locks_1.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{
"id": 1,
"lock_type": "cannot_delete",
"entity": {
"id": 123,
"type": "linode",
"label": "test-linode",
"url": "/v4/linode/instances/123"
}
}
1 change: 1 addition & 0 deletions test/integration/models/lock/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# This file is intentionally left empty to make the directory a Python package.
151 changes: 151 additions & 0 deletions test/integration/models/lock/test_lock.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
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 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 = [lk.id for lk in locks]
assert lock_id not in lock_ids
Loading
Loading