From b6118203fb4948e2deb1c9a7a1f1623f5d194914 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Tue, 28 Oct 2025 17:36:18 +0100 Subject: [PATCH 01/19] Added functionality to clean-up stale (e.g. longer than 30 days) Object Storage keys created by linode-cli. --- linodecli/configuration/config.py | 65 +++++- linodecli/plugins/obj/__init__.py | 255 ++++++++++++++++++++++- pyproject.toml | 7 +- tests/integration/obj/test_obj_plugin.py | 251 ++++++++++++++++++++++ 4 files changed, 554 insertions(+), 24 deletions(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index 700e82910..b708312aa 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -5,7 +5,7 @@ import argparse import os import sys -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Type, TypeVar, cast from linodecli.exit_codes import ExitCodes @@ -27,6 +27,8 @@ ENV_TOKEN_NAME = "LINODE_CLI_TOKEN" +T = TypeVar("T") + class CLIConfig: """ @@ -216,13 +218,8 @@ def plugin_set_value(self, key: str, value: Any): :param value: The value to set for this key :type value: any """ - if self.running_plugin is None: - raise RuntimeError( - "No running plugin to retrieve configuration for!" - ) - username = self.username or self.default_username() - self.config.set(username, f"plugin-{self.running_plugin}-{key}", value) + self.config.set(username, self._get_plugin_key(key), value) def plugin_get_value(self, key: str) -> Optional[Any]: """ @@ -238,15 +235,52 @@ def plugin_get_value(self, key: str) -> Optional[Any]: :returns: The value for this plugin for this key, or None if not set :rtype: any """ + username = self.username or self.default_username() or "DEFAULT" + return self.config.get( + username, self._get_plugin_key(key), fallback=None + ) + + def plugin_get_config_value_or_set_default( + self, key: str, default: T, value_type: Type[T] = str + ) -> T: + """ + Retrieves a plugin option value of the given type from the config. If the + value is not set, sets it to the provided default value and returns that. + """ + value = self.plugin_get_value(key) + + if value is None: + # option not set - set to default and store it in the config file + value_as_str = ( + ("yes" if default else "no") + if value_type is bool + else str(default) + ) + self.plugin_set_value(key, value_as_str) + self.write_config() + return default + + if value_type is bool: + return self.parse_boolean(value) + + return cast(T, value_type(value)) + + def plugin_remove_option(self, key: str): + """ + Removes a plugin configuration option. + + :param key: The key of the option to remove + """ + username = self.username or self.default_username() + self.config.remove_option(username, self._get_plugin_key(key)) + + def _get_plugin_key(self, key: str) -> str: if self.running_plugin is None: raise RuntimeError( "No running plugin to retrieve configuration for!" ) - username = self.username or self.default_username() or "DEFAULT" - full_key = f"plugin-{self.running_plugin}-{key}" - - return self.config.get(username, full_key, fallback=None) + return f"plugin-{self.running_plugin}-{key}" # TODO: this is more of an argparsing function than it is a config function # might be better to move this to argparsing during refactor and just have @@ -654,3 +688,12 @@ def get_custom_aliases(self) -> Dict[str, str]: if (self.config.has_section("custom_aliases")) else {} ) + + def parse_boolean(self, value: str) -> Optional[bool]: + """ + Parses a string config value into a boolean. + + :param value: The string value to parse. + :return: The parsed boolean value. + """ + return self.config.BOOLEAN_STATES.get(value.lower(), None) diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index 45c552db1..dcc210f57 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -4,6 +4,7 @@ """ import getpass import os +import re import socket import sys import time @@ -11,13 +12,13 @@ from contextlib import suppress from datetime import datetime from math import ceil -from typing import List +from typing import List, Optional +from pytimeparse import parse as parse_time from rich import print as rprint from rich.table import Table from linodecli.cli import CLI -from linodecli.configuration import _do_get_request from linodecli.configuration.helpers import _default_text_input from linodecli.exit_codes import ExitCodes from linodecli.help_formatter import SortingHelpFormatter @@ -65,6 +66,17 @@ HAS_BOTO = False +CLUSTER_KEY = "cluster" +PERFORM_KEY_CLEANUP_KEY = "perform-key-cleanup" +KEY_LIFESPAN_KEY = "key-lifespan" +KEY_ROTATION_PERIOD_KEY = "key-rotation-period" +KEY_CLEANUP_BATCH_SIZE_KEY = "key-cleanup-batch-size" +LAST_KEY_CLEANUP_TIMESTAMP_KEY = "last-key-cleanup-timestamp" +ACCESS_KEY_KEY = "access-key" +SECRET_KEY_KEY = "secret-key" +TOKEN_KEY = "token" + + def generate_url(get_client, args, **kwargs): # pylint: disable=unused-argument """ Generates a URL to an object @@ -322,6 +334,16 @@ def get_obj_args_parser(): type=str, help="The cluster to use for the operation", ) + parser.add_argument( + "--skip-key-cleanup", + action="store_true", + help="Skip cleanup of old linode-cli generated Object Storage keys", + ) + parser.add_argument( + "--force-key-cleanup", + action="store_true", + help="Force cleanup of old linode-cli generated Object Storage keys", + ) return parser @@ -400,8 +422,10 @@ def call( access_key = None secret_key = None - # make a client, but only if we weren't printing help + # make a client and clean-up keys, but only if we weren't printing help if not is_help: + if not parsed.skip_key_cleanup: + _cleanup_keys(context.client, parsed.force_key_cleanup) access_key, secret_key = get_credentials(context.client) cluster = parsed.cluster @@ -497,8 +521,8 @@ def _get_s3_creds(client: CLI, force: bool = False): :returns: The access key and secret key for this user :rtype: tuple(str, str) """ - access_key = client.config.plugin_get_value("access-key") - secret_key = client.config.plugin_get_value("secret-key") + access_key = client.config.plugin_get_value(ACCESS_KEY_KEY) + secret_key = client.config.plugin_get_value(SECRET_KEY_KEY) if force or access_key is None: # this means there are no stored s3 creds for this user - set them up @@ -507,7 +531,7 @@ def _get_s3_creds(client: CLI, force: bool = False): # being provided by the environment, but if the CLI is running without a # config, we shouldn't generate new keys (or we'd end up doing so with each # request) - instead ask for them to be set up. - if client.config.get_value("token") is None: + if client.config.get_value(TOKEN_KEY) is None: print( "You are running the Linode CLI without a configuration file, but " "object storage keys were not configured. " @@ -571,8 +595,8 @@ def _get_s3_creds(client: CLI, force: bool = False): access_key = resp["access_key"] secret_key = resp["secret_key"] - client.config.plugin_set_value("access-key", access_key) - client.config.plugin_set_value("secret-key", secret_key) + client.config.plugin_set_value(ACCESS_KEY_KEY, access_key) + client.config.plugin_set_value(SECRET_KEY_KEY, secret_key) client.config.write_config() return access_key, secret_key @@ -580,7 +604,7 @@ def _get_s3_creds(client: CLI, force: bool = False): def _configure_plugin(client: CLI): """ - Configures a default cluster value. + Configures Object Storage plugin. """ cluster = _default_text_input( # pylint: disable=protected-access @@ -589,5 +613,216 @@ def _configure_plugin(client: CLI): ) if cluster: - client.config.plugin_set_value("cluster", cluster) + client.config.plugin_set_value(CLUSTER_KEY, cluster) + + perform_key_cleanup = _default_text_input( # pylint: disable=protected-access + "Perform automatic cleanup of old linode-cli generated Object Storage keys?" + " The cleanup will occur before any Object Storage operation," + " at most once every 24 hours. (Y/n)", + default="y", + validator=lambda x: ( + "Please enter 'y' or 'n'" + if x.lower() not in ("y", "n", "yes", "no") + else None + ), + ) + perform_key_cleanup = ( + "yes" if perform_key_cleanup.lower() in ("y", "yes") else "no" + ) + client.config.plugin_set_value(PERFORM_KEY_CLEANUP_KEY, perform_key_cleanup) + if perform_key_cleanup == "yes": + key_lifespan = _default_text_input( # pylint: disable=protected-access + "Linode-cli Object Storage key lifespan (e.g. `30d` for 30 days)", + default="30d", + validator=lambda x: ( + "Please enter a valid time duration" + if parse_time(x) is None + else None + ), + ) + client.config.plugin_set_value(KEY_LIFESPAN_KEY, key_lifespan) + + key_rotation_period = _default_text_input( # pylint: disable=protected-access + "Linode-cli Object Storage key rotation period (e.g. `10d` for 10 days)", + default="10d", + validator=lambda x: ( + "Please enter a valid time duration" + if parse_time(x) is None + else None + ), + ) + client.config.plugin_set_value( + KEY_ROTATION_PERIOD_KEY, key_rotation_period + ) + + cleanup_batch_size = _default_text_input( # pylint: disable=protected-access + "Number of old linode-cli Object Storage keys to clean up at once", + default="10", + validator=lambda x: ( + "Please enter a valid integer" if not x.isdigit() else None + ), + ) + client.config.plugin_set_value( + KEY_CLEANUP_BATCH_SIZE_KEY, str(int(cleanup_batch_size)) + ) + client.config.write_config() + + +def _cleanup_keys(client: CLI, force_key_cleanup: bool) -> None: + """ + Cleans up stale linode-cli generated object storage keys. + """ + try: + if not _should_perform_key_cleanup(client): + return + + current_timestamp = int(time.time()) + last_cleanup = client.config.plugin_get_value( + LAST_KEY_CLEANUP_TIMESTAMP_KEY + ) + if ( + not force_key_cleanup + and last_cleanup + and int(last_cleanup) > current_timestamp - 24 * 60 * 60 + ): + # if we did a cleanup in the last 24 hours, skip + return + + print( + "Cleaning up old linode-cli Object Storage keys." + " To disable this, use the --skip-key-cleanup option.", + file=sys.stderr, + ) + status, keys = client.call_operation("object-storage", "keys-list") + if status != 200: + print( + "Failed to list object storage keys for cleanup", + file=sys.stderr, + ) + return + + key_lifespan = _get_key_lifespan(client) + key_rotation_period = _get_key_rotation_period(client) + cleanup_batch_size = _get_cleanup_batch_size(client) + + linode_cli_keys = _get_linode_cli_keys( + keys["data"], key_lifespan, key_rotation_period, current_timestamp + ) + _rotate_current_key_if_needed(client, linode_cli_keys) + _delete_stale_keys(client, linode_cli_keys, cleanup_batch_size) + + client.config.plugin_set_value( + LAST_KEY_CLEANUP_TIMESTAMP_KEY, str(current_timestamp) + ) + client.config.write_config() + + except Exception as e: + print( + "Unable to clean up stale linode-cli Object Storage keys", + e, + file=sys.stderr, + ) + + +def _should_perform_key_cleanup(client: CLI) -> bool: + return client.config.plugin_get_config_value_or_set_default( + PERFORM_KEY_CLEANUP_KEY, True, bool + ) + + +def _get_key_lifespan(client) -> str: + return client.config.plugin_get_config_value_or_set_default( + KEY_LIFESPAN_KEY, "30d" + ) + + +def _get_key_rotation_period(client) -> str: + return client.config.plugin_get_config_value_or_set_default( + KEY_ROTATION_PERIOD_KEY, "10d" + ) + + +def _get_cleanup_batch_size(client) -> int: + return client.config.plugin_get_config_value_or_set_default( + KEY_CLEANUP_BATCH_SIZE_KEY, 10, int + ) + + +def _get_linode_cli_keys( + keys_data: list, + key_lifespan: str, + key_rotation_period: str, + current_timestamp: int, +) -> list: + + stale_threshold = current_timestamp - parse_time(key_lifespan) + rotation_threshold = current_timestamp - parse_time(key_rotation_period) + + def extract_key_info(key: dict) -> Optional[dict]: + match = re.match(r"^linode-cli-.+@.+-(\d{10,})$", key["label"]) + if not match: + return None + created_timestamp = int(match.group(1)) + is_stale = created_timestamp <= stale_threshold + needs_rotation = is_stale or created_timestamp <= rotation_threshold + return { + "id": key["id"], + "label": key["label"], + "access_key": key["access_key"], + "created_timestamp": created_timestamp, + "is_stale": is_stale, + "needs_rotation": needs_rotation, + } + + return sorted( + [info for key in keys_data if (info := extract_key_info(key))], + key=lambda k: k["created_timestamp"], + ) + + +def _rotate_current_key_if_needed(client: CLI, linode_cli_keys: list) -> None: + current_access_key = client.config.plugin_get_value(ACCESS_KEY_KEY) + key_to_rotate = next( + ( + key_info + for key_info in linode_cli_keys + if key_info["access_key"] == current_access_key + and key_info["needs_rotation"] + ), + None, + ) + if key_to_rotate: + _delete_key(client, key_to_rotate["id"], key_to_rotate["label"]) + linode_cli_keys.remove(key_to_rotate) + client.config.plugin_remove_option(ACCESS_KEY_KEY) + client.config.plugin_remove_option(SECRET_KEY_KEY) + client.config.write_config() + + +def _delete_stale_keys( + client: CLI, linode_cli_keys: list, batch_size: int +) -> None: + stale_keys = [k for k in linode_cli_keys if k["is_stale"]] + for key_info in stale_keys[:batch_size]: + _delete_key(client, key_info["id"], key_info["label"]) + + +def _delete_key(client: CLI, key_id: str, label: str) -> None: + try: + print( + f"Deleting linode-cli Object Storage key: {label}", file=sys.stderr + ) + status, _ = client.call_operation( + "object-storage", "keys-delete", [str(key_id)] + ) + if status != 200: + print( + f"Failed to delete key: {label}; status {status}", + file=sys.stderr, + ) + except Exception as e: + print( + f"Exception occurred while deleting key: {label}; {e}", + file=sys.stderr, + ) diff --git a/pyproject.toml b/pyproject.toml index 27a8aea5a..40ac8ea05 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,10 +14,11 @@ dependencies = [ "openapi3", "requests", "PyYAML", - "packaging", - "rich", + "packaging<25", + "rich<14", "urllib3<3", - "linode-metadata>=0.3.0" + "linode-metadata>=0.3.0", + "pytimeparse" ] dynamic = ["version"] diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 3960ba81d..ce8f24ffa 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -1,5 +1,7 @@ +import time from concurrent.futures import ThreadPoolExecutor, wait from typing import Callable, Optional +from unittest.mock import patch import pytest import requests @@ -347,3 +349,252 @@ def test_generate_url( response = requests.get(url.strip("\n")) assert response.text == content assert response.status_code == 200 + + +def test_obj_action_triggers_key_cleanup_and_deletes_stale_key( + keys: Keys, + monkeypatch: MonkeyPatch, + create_bucket: callable, +): + patch_keys(keys, monkeypatch) + bucket = create_bucket() + + now = int(time.time()) + stale_timestamp = ( + now - 31 * 24 * 60 * 60 + ) # 31 days ago (assuming 30d lifespan) + fresh_timestamp = now + + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + fresh_key = { + "id": "fresh-id", + "label": f"linode-cli-testuser@localhost-{fresh_timestamp}", + "access_key": "FRESHKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key, fresh_key]} + if resource == "object-storage" and action == "keys-delete": + return 200, {} + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + if resource == "account" and action == "view": + return 200, {} + return 200, {} + + with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: + mock_client = MockCLI.return_value + mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[ + k + ] + mock_client.config.plugin_get_value.side_effect = lambda k: None + mock_client.call_operation.side_effect = call_operation_side_effect + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + # Execute the ls command + exec_test_command(BASE_CMD + ["ls", bucket]) + + # Check that keys-delete was called for the stale key only + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert any( + c[1][2][0] == "stale-id" for c in delete_calls + ), "Stale key was not deleted" + assert not any( + c[1][2][0] == "fresh-id" for c in delete_calls + ), "Fresh key should not be deleted" + + +def test_obj_action_triggers_key_rotation( + keys: Keys, + monkeypatch: MonkeyPatch, + create_bucket: callable, +): + patch_keys(keys, monkeypatch) + bucket = create_bucket() + + now = int(time.time()) + # Key created 31 days ago, rotation period is 30 days + old_timestamp = now - 60 * 60 * 24 * 31 + + key_due_for_rotation = { + "id": "rotate-id", + "label": f"linode-cli-testuser@localhost-{old_timestamp}", + "access_key": "ROTATEKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [key_due_for_rotation]} + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + if resource == "object-storage" and action == "keys-delete": + return 200, {} + if resource == "account" and action == "view": + return 200, {} + return 200, {} + + with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: + mock_client = MockCLI.return_value + mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "90d", + "key-rotation-period-days": "30d", + "key-cleanup-batch-size": 10, + }[ + k + ] + mock_client.config.plugin_get_value.side_effect = lambda k: None + mock_client.call_operation.side_effect = call_operation_side_effect + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + exec_test_command(BASE_CMD + ["ls", bucket]) + + # Check that keys-create (rotation) was called + create_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-create" + ] + assert create_calls, "Key rotation (keys-create) was not triggered" + + # Check that keys-delete was called for the old key + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert any( + c[1][2][0] == "rotate-id" for c in delete_calls + ), "Old key was not deleted after rotation" + + +def test_obj_action_does_not_trigger_cleanup_if_recent( + keys: Keys, + monkeypatch: MonkeyPatch, + create_bucket: callable, +): + patch_keys(keys, monkeypatch) + bucket = create_bucket() + + now = int(time.time()) + # Set last cleanup to 1 hour ago (less than 24h) + last_cleanup = now - 60 * 60 + + stale_timestamp = now - 31 * 24 * 60 * 60 + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key]} + if resource == "object-storage" and action == "keys-delete": + return 200, {} + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + if resource == "account" and action == "view": + return 200, {} + return 200, {} + + with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: + mock_client = MockCLI.return_value + mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[ + k + ] + # Return a recent last-key-cleanup-timestamp + mock_client.config.plugin_get_value.side_effect = lambda k: ( + str(last_cleanup) if k == "last-key-cleanup-timestamp" else None + ) + mock_client.call_operation.side_effect = call_operation_side_effect + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + exec_test_command(BASE_CMD + ["ls", bucket]) + + # Check that keys-delete was NOT called + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert ( + not delete_calls + ), "Cleanup should not be performed if it was done in the last 24 hours" + + +def test_obj_action_does_not_trigger_cleanup_if_disabled( + keys: Keys, + monkeypatch: MonkeyPatch, + create_bucket: callable, +): + patch_keys(keys, monkeypatch) + bucket = create_bucket() + + now = int(time.time()) + stale_timestamp = now - 31 * 24 * 60 * 60 + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key]} + if resource == "object-storage" and action == "keys-delete": + return 200, {} + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + if resource == "account" and action == "view": + return 200, {} + return 200, {} + + with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: + mock_client = MockCLI.return_value + mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { + "perform-key-cleanup": False, # Clean-up disabled + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[ + k + ] + mock_client.config.plugin_get_value.side_effect = lambda k: None + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + mock_client.call_operation.side_effect = call_operation_side_effect + + exec_test_command(BASE_CMD + ["ls", bucket]) + + # Check that keys-delete was NOT called + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert ( + not delete_calls + ), "Cleanup should not be performed when perform-key-cleanup is False" From bea9b7db5000986884bfa54567ee1325b137152b Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Wed, 26 Nov 2025 15:02:45 +0100 Subject: [PATCH 02/19] Code review changes: removed functionality to set the cleanup options with `linode-obj obj configure` and storing the default values; added CLI cleanup options instead; some refactoring. --- linodecli/configuration/config.py | 42 +++--- linodecli/plugins/obj/__init__.py | 161 ++++++++++------------- tests/integration/obj/test_obj_plugin.py | 64 ++++----- 3 files changed, 123 insertions(+), 144 deletions(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index b708312aa..c431ea457 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -221,7 +221,9 @@ def plugin_set_value(self, key: str, value: Any): username = self.username or self.default_username() self.config.set(username, self._get_plugin_key(key), value) - def plugin_get_value(self, key: str) -> Optional[Any]: + def plugin_get_value( + self, key: str, default: T = None, value_type: Type[T] = str + ) -> Optional[T]: """ Retrieves and returns a config value previously set for a plugin. Your plugin should have set this value in the past. If this value does not @@ -232,38 +234,36 @@ def plugin_get_value(self, key: str) -> Optional[Any]: :param key: The key of the value to return :type key: str + :param default: The default value to return if the key is not set + :type default: T + + :param value_type: The type to which the value should be cast + :type value_type: Type[T] + :returns: The value for this plugin for this key, or None if not set :rtype: any """ username = self.username or self.default_username() or "DEFAULT" - return self.config.get( + value = self.config.get( username, self._get_plugin_key(key), fallback=None ) - - def plugin_get_config_value_or_set_default( - self, key: str, default: T, value_type: Type[T] = str - ) -> T: - """ - Retrieves a plugin option value of the given type from the config. If the - value is not set, sets it to the provided default value and returns that. - """ - value = self.plugin_get_value(key) - if value is None: - # option not set - set to default and store it in the config file - value_as_str = ( - ("yes" if default else "no") - if value_type is bool - else str(default) - ) - self.plugin_set_value(key, value_as_str) - self.write_config() return default + if value_type is str: + return value + if value_type is bool: return self.parse_boolean(value) - return cast(T, value_type(value)) + try: + return cast(T, value_type(value)) + except (ValueError, TypeError): + print( + "" f"Could not cast config value {value} to {value_type}.", + file=sys.stderr, + ) + return default def plugin_remove_option(self, key: str): """ diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index dcc210f57..de6c9bb9d 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -67,7 +67,7 @@ CLUSTER_KEY = "cluster" -PERFORM_KEY_CLEANUP_KEY = "perform-key-cleanup" +KEY_CLEANUP_ENABLED_KEY = "key-cleanup-enabled" KEY_LIFESPAN_KEY = "key-lifespan" KEY_ROTATION_PERIOD_KEY = "key-rotation-period" KEY_CLEANUP_BATCH_SIZE_KEY = "key-cleanup-batch-size" @@ -326,23 +326,44 @@ def get_obj_args_parser(): metavar="COMMAND", nargs="?", type=str, - help="The command to execute in object storage", + help="The command to execute in object storage.", ) parser.add_argument( "--cluster", metavar="CLUSTER", type=str, - help="The cluster to use for the operation", + help="The cluster to use for the operation.", ) parser.add_argument( - "--skip-key-cleanup", + "--force-key-cleanup", action="store_true", - help="Skip cleanup of old linode-cli generated Object Storage keys", + help="Performs cleanup of old linode-cli generated Object Storage keys" + " before executing the Object Storage command. It overrides" + " the --perform-key-cleanup option.", ) parser.add_argument( - "--force-key-cleanup", - action="store_true", - help="Force cleanup of old linode-cli generated Object Storage keys", + "--key-cleanup-enabled", + choices=["yes", "no"], + help="If set to 'yes', performs cleanup of old linode-cli generated Object Storage" + " keys before executing the Object Storage command. Cleanup occurs" + " at most once every 24 hours.", + ) + parser.add_argument( + "--key-lifespan", + type=str, + help="Specifies the lifespan of linode-cli generated Object Storage keys" + " (e.g. 30d for 30 days). Used only during key cleanup.", + ) + parser.add_argument( + "--key-rotation-period", + type=str, + help="Specifies the period after which the linode-cli generated Object Storage" + " key must be rotated (e.g. 10d for 10 days). Used only during key cleanup.", + ) + parser.add_argument( + "--key-cleanup-batch-size", + type=int, + help="Number of old linode-cli generated Object Storage keys to clean up at once.", ) return parser @@ -424,8 +445,7 @@ def call( # make a client and clean-up keys, but only if we weren't printing help if not is_help: - if not parsed.skip_key_cleanup: - _cleanup_keys(context.client, parsed.force_key_cleanup) + _cleanup_keys(context.client, parsed) access_key, secret_key = get_credentials(context.client) cluster = parsed.cluster @@ -606,7 +626,6 @@ def _configure_plugin(client: CLI): """ Configures Object Storage plugin. """ - cluster = _default_text_input( # pylint: disable=protected-access "Default cluster for operations (e.g. `us-mia-1`)", optional=False, @@ -615,85 +634,27 @@ def _configure_plugin(client: CLI): if cluster: client.config.plugin_set_value(CLUSTER_KEY, cluster) - perform_key_cleanup = _default_text_input( # pylint: disable=protected-access - "Perform automatic cleanup of old linode-cli generated Object Storage keys?" - " The cleanup will occur before any Object Storage operation," - " at most once every 24 hours. (Y/n)", - default="y", - validator=lambda x: ( - "Please enter 'y' or 'n'" - if x.lower() not in ("y", "n", "yes", "no") - else None - ), - ) - perform_key_cleanup = ( - "yes" if perform_key_cleanup.lower() in ("y", "yes") else "no" - ) - client.config.plugin_set_value(PERFORM_KEY_CLEANUP_KEY, perform_key_cleanup) - if perform_key_cleanup == "yes": - key_lifespan = _default_text_input( # pylint: disable=protected-access - "Linode-cli Object Storage key lifespan (e.g. `30d` for 30 days)", - default="30d", - validator=lambda x: ( - "Please enter a valid time duration" - if parse_time(x) is None - else None - ), - ) - client.config.plugin_set_value(KEY_LIFESPAN_KEY, key_lifespan) - - key_rotation_period = _default_text_input( # pylint: disable=protected-access - "Linode-cli Object Storage key rotation period (e.g. `10d` for 10 days)", - default="10d", - validator=lambda x: ( - "Please enter a valid time duration" - if parse_time(x) is None - else None - ), - ) - client.config.plugin_set_value( - KEY_ROTATION_PERIOD_KEY, key_rotation_period - ) - - cleanup_batch_size = _default_text_input( # pylint: disable=protected-access - "Number of old linode-cli Object Storage keys to clean up at once", - default="10", - validator=lambda x: ( - "Please enter a valid integer" if not x.isdigit() else None - ), - ) - client.config.plugin_set_value( - KEY_CLEANUP_BATCH_SIZE_KEY, str(int(cleanup_batch_size)) - ) - client.config.write_config() -def _cleanup_keys(client: CLI, force_key_cleanup: bool) -> None: +def _cleanup_keys(client: CLI, options) -> None: """ Cleans up stale linode-cli generated object storage keys. """ try: - if not _should_perform_key_cleanup(client): - return - current_timestamp = int(time.time()) - last_cleanup = client.config.plugin_get_value( - LAST_KEY_CLEANUP_TIMESTAMP_KEY - ) - if ( - not force_key_cleanup - and last_cleanup - and int(last_cleanup) > current_timestamp - 24 * 60 * 60 - ): - # if we did a cleanup in the last 24 hours, skip + if not _should_perform_key_cleanup(client, options, current_timestamp): return - print( - "Cleaning up old linode-cli Object Storage keys." - " To disable this, use the --skip-key-cleanup option.", - file=sys.stderr, + cleanup_message = ( + "Cleaning up old linode-cli generated Object Storage keys." ) + if not options.force_key_cleanup: + cleanup_message += ( + " To disable this, use the '--key-cleanup-enabled no' option." + ) + print(cleanup_message, file=sys.stderr) + status, keys = client.call_operation("object-storage", "keys-list") if status != 200: print( @@ -702,9 +663,9 @@ def _cleanup_keys(client: CLI, force_key_cleanup: bool) -> None: ) return - key_lifespan = _get_key_lifespan(client) - key_rotation_period = _get_key_rotation_period(client) - cleanup_batch_size = _get_cleanup_batch_size(client) + key_lifespan = _get_key_lifespan(client, options) + key_rotation_period = _get_key_rotation_period(client, options) + cleanup_batch_size = _get_cleanup_batch_size(client, options) linode_cli_keys = _get_linode_cli_keys( keys["data"], key_lifespan, key_rotation_period, current_timestamp @@ -725,26 +686,44 @@ def _cleanup_keys(client: CLI, force_key_cleanup: bool) -> None: ) -def _should_perform_key_cleanup(client: CLI) -> bool: - return client.config.plugin_get_config_value_or_set_default( - PERFORM_KEY_CLEANUP_KEY, True, bool +def _should_perform_key_cleanup( + client: CLI, options, current_timestamp +) -> bool: + if options.force_key_cleanup: + return True + if not _is_key_cleanup_enabled(client, options): + return False + last_cleanup = client.config.plugin_get_value( + LAST_KEY_CLEANUP_TIMESTAMP_KEY ) + # if we did a cleanup in the last 24 hours, skip it this time + return ( + last_cleanup is None + or int(last_cleanup) <= current_timestamp - 24 * 60 * 60 + ) + + +def _is_key_cleanup_enabled(client, options) -> bool: + if options.key_cleanup_enabled in ["yes", "no"]: + return options.key_cleanup_enabled == "yes" + return client.config.plugin_get_value(KEY_CLEANUP_ENABLED_KEY, True, bool) + -def _get_key_lifespan(client) -> str: - return client.config.plugin_get_config_value_or_set_default( +def _get_key_lifespan(client, options) -> str: + return options.key_lifespan or client.config.plugin_get_value( KEY_LIFESPAN_KEY, "30d" ) -def _get_key_rotation_period(client) -> str: - return client.config.plugin_get_config_value_or_set_default( +def _get_key_rotation_period(client, options) -> str: + return options.key_rotation_period or client.config.plugin_get_value( KEY_ROTATION_PERIOD_KEY, "10d" ) -def _get_cleanup_batch_size(client) -> int: - return client.config.plugin_get_config_value_or_set_default( +def _get_cleanup_batch_size(client, options) -> int: + return options.key_cleanup_batch_size or client.config.plugin_get_value( KEY_CLEANUP_BATCH_SIZE_KEY, 10, int ) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index ce8f24ffa..e2c1f81b5 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -389,14 +389,14 @@ def call_operation_side_effect(resource, action, *args, **kwargs): with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: mock_client = MockCLI.return_value - mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { - "perform-key-cleanup": True, - "key-lifespan": "30d", - "key-rotation-period-days": "10d", - "key-cleanup-batch-size": 10, - }[ - k - ] + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[k] + ) mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.call_operation.side_effect = call_operation_side_effect mock_client.config.plugin_set_value.return_value = None @@ -450,14 +450,14 @@ def call_operation_side_effect(resource, action, *args, **kwargs): with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: mock_client = MockCLI.return_value - mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { - "perform-key-cleanup": True, - "key-lifespan": "90d", - "key-rotation-period-days": "30d", - "key-cleanup-batch-size": 10, - }[ - k - ] + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "90d", + "key-rotation-period-days": "30d", + "key-cleanup-batch-size": 10, + }[k] + ) mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.call_operation.side_effect = call_operation_side_effect mock_client.config.plugin_set_value.return_value = None @@ -516,14 +516,14 @@ def call_operation_side_effect(resource, action, *args, **kwargs): with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: mock_client = MockCLI.return_value - mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { - "perform-key-cleanup": True, - "key-lifespan": "30d", - "key-rotation-period-days": "10d", - "key-cleanup-batch-size": 10, - }[ - k - ] + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "perform-key-cleanup": True, + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[k] + ) # Return a recent last-key-cleanup-timestamp mock_client.config.plugin_get_value.side_effect = lambda k: ( str(last_cleanup) if k == "last-key-cleanup-timestamp" else None @@ -574,14 +574,14 @@ def call_operation_side_effect(resource, action, *args, **kwargs): with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: mock_client = MockCLI.return_value - mock_client.config.plugin_get_config_value_or_set_default.side_effect = lambda k, d=None, t=None: { - "perform-key-cleanup": False, # Clean-up disabled - "key-lifespan": "30d", - "key-rotation-period-days": "10d", - "key-cleanup-batch-size": 10, - }[ - k - ] + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "perform-key-cleanup": False, # Clean-up disabled + "key-lifespan": "30d", + "key-rotation-period-days": "10d", + "key-cleanup-batch-size": 10, + }[k] + ) mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.config.plugin_set_value.return_value = None mock_client.config.write_config.return_value = None From faa787613cf73f95a55517adaed463937851a898 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 09:26:33 +0100 Subject: [PATCH 03/19] Code review changes: reverting package version restriction --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 40ac8ea05..e4fdba13d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,8 +14,8 @@ dependencies = [ "openapi3", "requests", "PyYAML", - "packaging<25", - "rich<14", + "packaging", + "rich", "urllib3<3", "linode-metadata>=0.3.0", "pytimeparse" From 93442907cbeb8a369ed725b9eaf887f2a0264839 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 16:27:46 +0100 Subject: [PATCH 04/19] Code review changes: improving cleanup info message --- linodecli/plugins/obj/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index de6c9bb9d..a8ac6534b 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -649,7 +649,7 @@ def _cleanup_keys(client: CLI, options) -> None: cleanup_message = ( "Cleaning up old linode-cli generated Object Storage keys." ) - if not options.force_key_cleanup: + if not options.force_key_cleanup and not options.key_cleanup_enabled: cleanup_message += ( " To disable this, use the '--key-cleanup-enabled no' option." ) From 6a2c9fb80df25450782865510be0871fb432a072 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 16:41:48 +0100 Subject: [PATCH 05/19] Code review changes: tests improved and minor corrections --- linodecli/configuration/config.py | 2 +- linodecli/plugins/obj/__init__.py | 2 +- tests/integration/obj/test_obj_plugin.py | 8 +------- 3 files changed, 3 insertions(+), 9 deletions(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index c431ea457..f98f44def 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -260,7 +260,7 @@ def plugin_get_value( return cast(T, value_type(value)) except (ValueError, TypeError): print( - "" f"Could not cast config value {value} to {value_type}.", + f"Could not cast config value {value} to {value_type}.", file=sys.stderr, ) return default diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index a8ac6534b..51333fc18 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -743,7 +743,7 @@ def extract_key_info(key: dict) -> Optional[dict]: if not match: return None created_timestamp = int(match.group(1)) - is_stale = created_timestamp <= stale_threshold + is_stale = created_timestamp < stale_threshold needs_rotation = is_stale or created_timestamp <= rotation_threshold return { "id": key["id"], diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index e2c1f81b5..c331a9d46 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -397,7 +397,6 @@ def call_operation_side_effect(resource, action, *args, **kwargs): "key-cleanup-batch-size": 10, }[k] ) - mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.call_operation.side_effect = call_operation_side_effect mock_client.config.plugin_set_value.return_value = None mock_client.config.write_config.return_value = None @@ -458,7 +457,6 @@ def call_operation_side_effect(resource, action, *args, **kwargs): "key-cleanup-batch-size": 10, }[k] ) - mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.call_operation.side_effect = call_operation_side_effect mock_client.config.plugin_set_value.return_value = None mock_client.config.write_config.return_value = None @@ -522,12 +520,9 @@ def call_operation_side_effect(resource, action, *args, **kwargs): "key-lifespan": "30d", "key-rotation-period-days": "10d", "key-cleanup-batch-size": 10, + "last-key-cleanup-timestamp": str(last_cleanup), }[k] ) - # Return a recent last-key-cleanup-timestamp - mock_client.config.plugin_get_value.side_effect = lambda k: ( - str(last_cleanup) if k == "last-key-cleanup-timestamp" else None - ) mock_client.call_operation.side_effect = call_operation_side_effect mock_client.config.plugin_set_value.return_value = None mock_client.config.write_config.return_value = None @@ -582,7 +577,6 @@ def call_operation_side_effect(resource, action, *args, **kwargs): "key-cleanup-batch-size": 10, }[k] ) - mock_client.config.plugin_get_value.side_effect = lambda k: None mock_client.config.plugin_set_value.return_value = None mock_client.config.write_config.return_value = None mock_client.call_operation.side_effect = call_operation_side_effect From 20ab1cd33d4fd9d92ad099ebc1d204538bc1f7a8 Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 16:54:01 +0100 Subject: [PATCH 06/19] Update tests/integration/obj/test_obj_plugin.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- tests/integration/obj/test_obj_plugin.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index c331a9d46..7aa1a8471 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -451,9 +451,9 @@ def call_operation_side_effect(resource, action, *args, **kwargs): mock_client = MockCLI.return_value mock_client.config.plugin_get_value.side_effect = ( lambda k, d=None, t=None: { - "perform-key-cleanup": True, + "key-cleanup-enabled": True, "key-lifespan": "90d", - "key-rotation-period-days": "30d", + "key-rotation-period": "30d", "key-cleanup-batch-size": 10, }[k] ) From 3f9286a64da041815c8b90088f92b39c94d4f1d0 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 16:56:19 +0100 Subject: [PATCH 07/19] Code review changes: tests updated --- tests/integration/obj/test_obj_plugin.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 7aa1a8471..67a0f7c38 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -516,9 +516,9 @@ def call_operation_side_effect(resource, action, *args, **kwargs): mock_client = MockCLI.return_value mock_client.config.plugin_get_value.side_effect = ( lambda k, d=None, t=None: { - "perform-key-cleanup": True, + "key-cleanup-enabled": True, "key-lifespan": "30d", - "key-rotation-period-days": "10d", + "key-rotation-period": "10d", "key-cleanup-batch-size": 10, "last-key-cleanup-timestamp": str(last_cleanup), }[k] @@ -571,9 +571,9 @@ def call_operation_side_effect(resource, action, *args, **kwargs): mock_client = MockCLI.return_value mock_client.config.plugin_get_value.side_effect = ( lambda k, d=None, t=None: { - "perform-key-cleanup": False, # Clean-up disabled + "key-cleanup-enabled": False, # Clean-up disabled "key-lifespan": "30d", - "key-rotation-period-days": "10d", + "key-rotation-period": "10d", "key-cleanup-batch-size": 10, }[k] ) From 63644a552b58e887ccd1d530de5e22f095aee741 Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:08:14 +0100 Subject: [PATCH 08/19] Update linodecli/plugins/obj/__init__.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- linodecli/plugins/obj/__init__.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index 51333fc18..2262bef7b 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -680,8 +680,7 @@ def _cleanup_keys(client: CLI, options) -> None: except Exception as e: print( - "Unable to clean up stale linode-cli Object Storage keys", - e, + f"Unable to clean up stale linode-cli Object Storage keys: {e}", file=sys.stderr, ) From 408e171cf55b2843bbaf0fc6cad8f2c4bc341c89 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 17:14:14 +0100 Subject: [PATCH 09/19] Code review changes: small improvements suggested by Copilot --- linodecli/configuration/config.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index f98f44def..9d84e4133 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -254,7 +254,8 @@ def plugin_get_value( return value if value_type is bool: - return self.parse_boolean(value) + bool_value = self.parse_boolean(value) + return bool_value if bool_value is not None else default try: return cast(T, value_type(value)) @@ -691,7 +692,8 @@ def get_custom_aliases(self) -> Dict[str, str]: def parse_boolean(self, value: str) -> Optional[bool]: """ - Parses a string config value into a boolean. + Parses a string config value into a boolean. Returns None if the value + cannot be parsed as a boolean. :param value: The string value to parse. :return: The parsed boolean value. From 9e2e2c9c1662b7ea764b70633f526cf39e2af8aa Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:18:17 +0100 Subject: [PATCH 10/19] Update linodecli/configuration/config.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- linodecli/configuration/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index 9d84e4133..20c13be7d 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -222,7 +222,7 @@ def plugin_set_value(self, key: str, value: Any): self.config.set(username, self._get_plugin_key(key), value) def plugin_get_value( - self, key: str, default: T = None, value_type: Type[T] = str + self, key: str, default: Optional[T] = None, value_type: Type[T] = str ) -> Optional[T]: """ Retrieves and returns a config value previously set for a plugin. Your From fb071f6d5f7449113550ab0ce9ac4a49ba449ab5 Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:22:41 +0100 Subject: [PATCH 11/19] Update linodecli/configuration/config.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- linodecli/configuration/config.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index 20c13be7d..9dc85d9dc 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -250,10 +250,10 @@ def plugin_get_value( if value is None: return default - if value_type is str: + if value_type == str: return value - if value_type is bool: + if value_type == bool: bool_value = self.parse_boolean(value) return bool_value if bool_value is not None else default From a1ac3e1ab3f46ded853a4e147cdd829af43189c6 Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:23:01 +0100 Subject: [PATCH 12/19] Update tests/integration/obj/test_obj_plugin.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- tests/integration/obj/test_obj_plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 67a0f7c38..87c99f40e 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -571,7 +571,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): mock_client = MockCLI.return_value mock_client.config.plugin_get_value.side_effect = ( lambda k, d=None, t=None: { - "key-cleanup-enabled": False, # Clean-up disabled + "key-cleanup-enabled": False, # Cleanup disabled "key-lifespan": "30d", "key-rotation-period": "10d", "key-cleanup-batch-size": 10, From 9f64ccb4ba74b964e8712d9b0f8cade9b6a478ec Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:26:15 +0100 Subject: [PATCH 13/19] Update tests/integration/obj/test_obj_plugin.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- tests/integration/obj/test_obj_plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 87c99f40e..86ee4eb87 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -391,7 +391,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): mock_client = MockCLI.return_value mock_client.config.plugin_get_value.side_effect = ( lambda k, d=None, t=None: { - "perform-key-cleanup": True, + "key-cleanup-enabled": True, "key-lifespan": "30d", "key-rotation-period-days": "10d", "key-cleanup-batch-size": 10, From 0a7f4db173754d0c82ba69d07504925df57ae366 Mon Sep 17 00:00:00 2001 From: skulpok-akamai Date: Thu, 27 Nov 2025 17:26:46 +0100 Subject: [PATCH 14/19] Update tests/integration/obj/test_obj_plugin.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- tests/integration/obj/test_obj_plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 86ee4eb87..6a38db85e 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -393,7 +393,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): lambda k, d=None, t=None: { "key-cleanup-enabled": True, "key-lifespan": "30d", - "key-rotation-period-days": "10d", + "key-rotation-period": "10d", "key-cleanup-batch-size": 10, }[k] ) From df1209f601c6e4882dc48d2d0475131e00652931 Mon Sep 17 00:00:00 2001 From: Sebastian Kulpok Date: Thu, 27 Nov 2025 17:27:45 +0100 Subject: [PATCH 15/19] Code review changes: test updated --- tests/integration/obj/test_obj_plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 6a38db85e..2193b2470 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -591,4 +591,4 @@ def call_operation_side_effect(resource, action, *args, **kwargs): ] assert ( not delete_calls - ), "Cleanup should not be performed when perform-key-cleanup is False" + ), "Cleanup should not be performed when key-cleanup-enabled is False" From 9eb85c88732073dd49f448241f1e7844bffca75d Mon Sep 17 00:00:00 2001 From: Lena Garber Date: Thu, 11 Dec 2025 12:44:37 -0500 Subject: [PATCH 16/19] Migrate integration tests to unit tests --- linodecli/plugins/obj/__init__.py | 7 +- tests/integration/obj/test_obj_plugin.py | 245 ------------------ tests/unit/test_plugin_obj.py | 311 ++++++++++++++++++++++- 3 files changed, 315 insertions(+), 248 deletions(-) diff --git a/linodecli/plugins/obj/__init__.py b/linodecli/plugins/obj/__init__.py index 2262bef7b..2d5761340 100644 --- a/linodecli/plugins/obj/__init__.py +++ b/linodecli/plugins/obj/__init__.py @@ -641,6 +641,7 @@ def _cleanup_keys(client: CLI, options) -> None: """ Cleans up stale linode-cli generated object storage keys. """ + try: current_timestamp = int(time.time()) if not _should_perform_key_cleanup(client, options, current_timestamp): @@ -670,6 +671,7 @@ def _cleanup_keys(client: CLI, options) -> None: linode_cli_keys = _get_linode_cli_keys( keys["data"], key_lifespan, key_rotation_period, current_timestamp ) + _rotate_current_key_if_needed(client, linode_cli_keys) _delete_stale_keys(client, linode_cli_keys, cleanup_batch_size) @@ -692,6 +694,7 @@ def _should_perform_key_cleanup( return True if not _is_key_cleanup_enabled(client, options): return False + last_cleanup = client.config.plugin_get_value( LAST_KEY_CLEANUP_TIMESTAMP_KEY ) @@ -733,7 +736,6 @@ def _get_linode_cli_keys( key_rotation_period: str, current_timestamp: int, ) -> list: - stale_threshold = current_timestamp - parse_time(key_lifespan) rotation_threshold = current_timestamp - parse_time(key_rotation_period) @@ -741,9 +743,11 @@ def extract_key_info(key: dict) -> Optional[dict]: match = re.match(r"^linode-cli-.+@.+-(\d{10,})$", key["label"]) if not match: return None + created_timestamp = int(match.group(1)) is_stale = created_timestamp < stale_threshold needs_rotation = is_stale or created_timestamp <= rotation_threshold + return { "id": key["id"], "label": key["label"], @@ -761,6 +765,7 @@ def extract_key_info(key: dict) -> Optional[dict]: def _rotate_current_key_if_needed(client: CLI, linode_cli_keys: list) -> None: current_access_key = client.config.plugin_get_value(ACCESS_KEY_KEY) + key_to_rotate = next( ( key_info diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 2193b2470..3960ba81d 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -1,7 +1,5 @@ -import time from concurrent.futures import ThreadPoolExecutor, wait from typing import Callable, Optional -from unittest.mock import patch import pytest import requests @@ -349,246 +347,3 @@ def test_generate_url( response = requests.get(url.strip("\n")) assert response.text == content assert response.status_code == 200 - - -def test_obj_action_triggers_key_cleanup_and_deletes_stale_key( - keys: Keys, - monkeypatch: MonkeyPatch, - create_bucket: callable, -): - patch_keys(keys, monkeypatch) - bucket = create_bucket() - - now = int(time.time()) - stale_timestamp = ( - now - 31 * 24 * 60 * 60 - ) # 31 days ago (assuming 30d lifespan) - fresh_timestamp = now - - stale_key = { - "id": "stale-id", - "label": f"linode-cli-testuser@localhost-{stale_timestamp}", - "access_key": "STALEKEY", - } - fresh_key = { - "id": "fresh-id", - "label": f"linode-cli-testuser@localhost-{fresh_timestamp}", - "access_key": "FRESHKEY", - } - - def call_operation_side_effect(resource, action, *args, **kwargs): - if resource == "object-storage" and action == "keys-list": - return 200, {"data": [stale_key, fresh_key]} - if resource == "object-storage" and action == "keys-delete": - return 200, {} - if resource == "object-storage" and action == "keys-create": - return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} - if resource == "account" and action == "view": - return 200, {} - return 200, {} - - with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: - mock_client = MockCLI.return_value - mock_client.config.plugin_get_value.side_effect = ( - lambda k, d=None, t=None: { - "key-cleanup-enabled": True, - "key-lifespan": "30d", - "key-rotation-period": "10d", - "key-cleanup-batch-size": 10, - }[k] - ) - mock_client.call_operation.side_effect = call_operation_side_effect - mock_client.config.plugin_set_value.return_value = None - mock_client.config.write_config.return_value = None - - # Execute the ls command - exec_test_command(BASE_CMD + ["ls", bucket]) - - # Check that keys-delete was called for the stale key only - delete_calls = [ - c - for c in mock_client.call_operation.mock_calls - if c[1][1] == "keys-delete" - ] - assert any( - c[1][2][0] == "stale-id" for c in delete_calls - ), "Stale key was not deleted" - assert not any( - c[1][2][0] == "fresh-id" for c in delete_calls - ), "Fresh key should not be deleted" - - -def test_obj_action_triggers_key_rotation( - keys: Keys, - monkeypatch: MonkeyPatch, - create_bucket: callable, -): - patch_keys(keys, monkeypatch) - bucket = create_bucket() - - now = int(time.time()) - # Key created 31 days ago, rotation period is 30 days - old_timestamp = now - 60 * 60 * 24 * 31 - - key_due_for_rotation = { - "id": "rotate-id", - "label": f"linode-cli-testuser@localhost-{old_timestamp}", - "access_key": "ROTATEKEY", - } - - def call_operation_side_effect(resource, action, *args, **kwargs): - if resource == "object-storage" and action == "keys-list": - return 200, {"data": [key_due_for_rotation]} - if resource == "object-storage" and action == "keys-create": - return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} - if resource == "object-storage" and action == "keys-delete": - return 200, {} - if resource == "account" and action == "view": - return 200, {} - return 200, {} - - with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: - mock_client = MockCLI.return_value - mock_client.config.plugin_get_value.side_effect = ( - lambda k, d=None, t=None: { - "key-cleanup-enabled": True, - "key-lifespan": "90d", - "key-rotation-period": "30d", - "key-cleanup-batch-size": 10, - }[k] - ) - mock_client.call_operation.side_effect = call_operation_side_effect - mock_client.config.plugin_set_value.return_value = None - mock_client.config.write_config.return_value = None - - exec_test_command(BASE_CMD + ["ls", bucket]) - - # Check that keys-create (rotation) was called - create_calls = [ - c - for c in mock_client.call_operation.mock_calls - if c[1][1] == "keys-create" - ] - assert create_calls, "Key rotation (keys-create) was not triggered" - - # Check that keys-delete was called for the old key - delete_calls = [ - c - for c in mock_client.call_operation.mock_calls - if c[1][1] == "keys-delete" - ] - assert any( - c[1][2][0] == "rotate-id" for c in delete_calls - ), "Old key was not deleted after rotation" - - -def test_obj_action_does_not_trigger_cleanup_if_recent( - keys: Keys, - monkeypatch: MonkeyPatch, - create_bucket: callable, -): - patch_keys(keys, monkeypatch) - bucket = create_bucket() - - now = int(time.time()) - # Set last cleanup to 1 hour ago (less than 24h) - last_cleanup = now - 60 * 60 - - stale_timestamp = now - 31 * 24 * 60 * 60 - stale_key = { - "id": "stale-id", - "label": f"linode-cli-testuser@localhost-{stale_timestamp}", - "access_key": "STALEKEY", - } - - def call_operation_side_effect(resource, action, *args, **kwargs): - if resource == "object-storage" and action == "keys-list": - return 200, {"data": [stale_key]} - if resource == "object-storage" and action == "keys-delete": - return 200, {} - if resource == "object-storage" and action == "keys-create": - return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} - if resource == "account" and action == "view": - return 200, {} - return 200, {} - - with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: - mock_client = MockCLI.return_value - mock_client.config.plugin_get_value.side_effect = ( - lambda k, d=None, t=None: { - "key-cleanup-enabled": True, - "key-lifespan": "30d", - "key-rotation-period": "10d", - "key-cleanup-batch-size": 10, - "last-key-cleanup-timestamp": str(last_cleanup), - }[k] - ) - mock_client.call_operation.side_effect = call_operation_side_effect - mock_client.config.plugin_set_value.return_value = None - mock_client.config.write_config.return_value = None - - exec_test_command(BASE_CMD + ["ls", bucket]) - - # Check that keys-delete was NOT called - delete_calls = [ - c - for c in mock_client.call_operation.mock_calls - if c[1][1] == "keys-delete" - ] - assert ( - not delete_calls - ), "Cleanup should not be performed if it was done in the last 24 hours" - - -def test_obj_action_does_not_trigger_cleanup_if_disabled( - keys: Keys, - monkeypatch: MonkeyPatch, - create_bucket: callable, -): - patch_keys(keys, monkeypatch) - bucket = create_bucket() - - now = int(time.time()) - stale_timestamp = now - 31 * 24 * 60 * 60 - stale_key = { - "id": "stale-id", - "label": f"linode-cli-testuser@localhost-{stale_timestamp}", - "access_key": "STALEKEY", - } - - def call_operation_side_effect(resource, action, *args, **kwargs): - if resource == "object-storage" and action == "keys-list": - return 200, {"data": [stale_key]} - if resource == "object-storage" and action == "keys-delete": - return 200, {} - if resource == "object-storage" and action == "keys-create": - return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} - if resource == "account" and action == "view": - return 200, {} - return 200, {} - - with patch("linodecli.plugins.obj.__init__.CLI") as MockCLI: - mock_client = MockCLI.return_value - mock_client.config.plugin_get_value.side_effect = ( - lambda k, d=None, t=None: { - "key-cleanup-enabled": False, # Cleanup disabled - "key-lifespan": "30d", - "key-rotation-period": "10d", - "key-cleanup-batch-size": 10, - }[k] - ) - mock_client.config.plugin_set_value.return_value = None - mock_client.config.write_config.return_value = None - mock_client.call_operation.side_effect = call_operation_side_effect - - exec_test_command(BASE_CMD + ["ls", bucket]) - - # Check that keys-delete was NOT called - delete_calls = [ - c - for c in mock_client.call_operation.mock_calls - if c[1][1] == "keys-delete" - ] - assert ( - not delete_calls - ), "Cleanup should not be performed when key-cleanup-enabled is False" diff --git a/tests/unit/test_plugin_obj.py b/tests/unit/test_plugin_obj.py index 17ae5c425..bf3db9b4c 100644 --- a/tests/unit/test_plugin_obj.py +++ b/tests/unit/test_plugin_obj.py @@ -1,6 +1,7 @@ -from pytest import CaptureFixture +from pytest import CaptureFixture, MonkeyPatch -from linodecli import CLI +from linodecli import CLI, plugins +from linodecli.plugins import obj from linodecli.plugins.obj import get_obj_args_parser, helpers, print_help @@ -34,3 +35,309 @@ def test_helpers_denominate(): assert helpers._denominate(123456789) == "117.74 MB" assert helpers._denominate(1e23) == "90949470177.29 TB" + + +def test_obj_action_triggers_key_cleanup_and_deletes_stale_key( + monkeypatch: MonkeyPatch, +): + now = int(time.time()) + stale_timestamp = ( + now - 31 * 24 * 60 * 60 + ) # 31 days ago (assuming 30d lifespan) + fresh_timestamp = now + + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + fresh_key = { + "id": "fresh-id", + "label": f"linode-cli-testuser@localhost-{fresh_timestamp}", + "access_key": "FRESHKEY", + } + + # Mocks for Linode CLI commands + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key, fresh_key]} + + if resource == "object-storage" and action == "keys-delete": + return 200, {} + + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + + if resource == "account" and action == "view": + return 200, {} + + return 200, {} + + # OBJ plugin & CLI config mocks + with ( + patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch.dict( + obj.COMMAND_MAP, + { + # We don't want to actually execute any S3 operations + "ls": lambda *args, **kwargs: None, + }, + ), + ): + mock_client = MockCLI.return_value + mock_client.call_operation.side_effect = call_operation_side_effect + + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "key-cleanup-enabled": True, + "key-lifespan": "30d", + "key-rotation-period": "10d", + "key-cleanup-batch-size": 10, + }.get(k, None) + ) + + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + # Execute the ls command + obj.call( + ["ls", "bucket"], + plugins.PluginContext("12345", mock_client), + ) + + # Check that keys-delete was called for the stale key only + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert any( + c[1][2][0] == "stale-id" for c in delete_calls + ), "Stale key was not deleted" + assert not any( + c[1][2][0] == "fresh-id" for c in delete_calls + ), "Fresh key should not be deleted" + + +def test_obj_action_triggers_key_rotation( + monkeypatch: MonkeyPatch, +): + now = int(time.time()) + # Key created 31 days ago, rotation period is 30 days + old_timestamp = now - 60 * 60 * 24 * 31 + + key_due_for_rotation = { + "id": "rotate-id", + "label": f"linode-cli-testuser@localhost-{old_timestamp}", + "access_key": "ROTATEKEY", + } + + # Mocks for Linode CLI commands + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [key_due_for_rotation]} + + if resource == "object-storage" and action == "keys-delete": + return 200, {} + + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + + if resource == "account" and action == "view": + return 200, {} + + return 200, {} + + # OBJ plugin & CLI config mocks + with ( + patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch.dict( + obj.COMMAND_MAP, + { + # We don't want to actually execute any S3 operations + "ls": lambda *args, **kwargs: None, + }, + ), + ): + mock_client = MockCLI.return_value + mock_client.call_operation.side_effect = call_operation_side_effect + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + mock_options = { + "access-key": "ROTATEKEY", + "secret-key": "12345", + "key-cleanup-enabled": True, + "key-lifespan": "90d", + "key-rotation-period": "30d", + "key-cleanup-batch-size": 10, + } + + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: mock_options.get(k, None) + ) + + # This test relies on the plugin updating the config + mock_client.config.plugin_remove_option.side_effect = ( + lambda k, d=None, t=None: mock_options.pop(k, None) + ) + + obj.call( + ["ls", "bucket"], + plugins.PluginContext("12345", mock_client), + ) + + # Check that keys-create (rotation) was called + create_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-create" + ] + assert create_calls, "Key rotation (keys-create) was not triggered" + + # Check that keys-delete was called for the old key + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert any( + c[1][2][0] == "rotate-id" for c in delete_calls + ), "Old key was not deleted after rotation" + + +def test_obj_action_does_not_trigger_cleanup_if_recent( + monkeypatch: MonkeyPatch, +): + now = int(time.time()) + # Set last cleanup to 1 hour ago (less than 24h) + last_cleanup = now - 60 * 60 + + stale_timestamp = now - 31 * 24 * 60 * 60 + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key]} + + if resource == "object-storage" and action == "keys-delete": + return 200, {} + + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + + if resource == "account" and action == "view": + return 200, {} + + return 200, {} + + with ( + patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch.dict( + obj.COMMAND_MAP, + { + # We don't want to actually execute any S3 operations + "ls": lambda *args, **kwargs: None, + }, + ), + ): + mock_client = MockCLI.return_value + + mock_client.call_operation.side_effect = call_operation_side_effect + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "cluster": "us-mia-1", + "key-cleanup-enabled": True, + "key-lifespan": "30d", + "key-rotation-period": "10d", + "key-cleanup-batch-size": 10, + "last-key-cleanup-timestamp": str(last_cleanup), + }.get(k, None) + ) + + obj.call( + ["ls", "bucket"], + plugins.PluginContext("12345", mock_client), + ) + + # Check that keys-delete was NOT called + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert ( + not delete_calls + ), "Cleanup should not be performed if it was done in the last 24 hours" + + +def test_obj_action_does_not_trigger_cleanup_if_disabled( + monkeypatch: MonkeyPatch, +): + now = int(time.time()) + stale_timestamp = now - 31 * 24 * 60 * 60 + stale_key = { + "id": "stale-id", + "label": f"linode-cli-testuser@localhost-{stale_timestamp}", + "access_key": "STALEKEY", + } + + def call_operation_side_effect(resource, action, *args, **kwargs): + if resource == "object-storage" and action == "keys-list": + return 200, {"data": [stale_key]} + + if resource == "object-storage" and action == "keys-delete": + return 200, {} + + if resource == "object-storage" and action == "keys-create": + return 200, {"access_key": "NEWKEY", "secret_key": "NEWSECRET"} + + if resource == "account" and action == "view": + return 200, {} + + return 200, {} + + with ( + patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch.dict( + obj.COMMAND_MAP, + { + # We don't want to actually execute any S3 operations + "ls": lambda *args, **kwargs: None, + }, + ), + ): + mock_client = MockCLI.return_value + mock_client.config.plugin_get_value.side_effect = ( + lambda k, d=None, t=None: { + "key-cleanup-enabled": False, # Cleanup disabled + "key-lifespan": "30d", + "key-rotation-period": "10d", + "key-cleanup-batch-size": 10, + }.get(k, None) + ) + mock_client.config.plugin_set_value.return_value = None + mock_client.config.write_config.return_value = None + mock_client.call_operation.side_effect = call_operation_side_effect + + obj.call( + ["ls", "bucket"], + plugins.PluginContext("12345", mock_client), + ) + + # Check that keys-delete was NOT called + delete_calls = [ + c + for c in mock_client.call_operation.mock_calls + if c[1][1] == "keys-delete" + ] + assert ( + not delete_calls + ), "Cleanup should not be performed when key-cleanup-enabled is False" From 3f9ccc462a93b4c2d6f83e43074aa2fd1ef45cba Mon Sep 17 00:00:00 2001 From: Lena Garber Date: Thu, 11 Dec 2025 12:46:39 -0500 Subject: [PATCH 17/19] fix imports --- tests/unit/test_plugin_obj.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/unit/test_plugin_obj.py b/tests/unit/test_plugin_obj.py index bf3db9b4c..647422457 100644 --- a/tests/unit/test_plugin_obj.py +++ b/tests/unit/test_plugin_obj.py @@ -1,3 +1,6 @@ +import time +from unittest.mock import patch + from pytest import CaptureFixture, MonkeyPatch from linodecli import CLI, plugins From 732d773304f655b90c5f84a6579dcddd42c0057c Mon Sep 17 00:00:00 2001 From: Lena Garber Date: Thu, 11 Dec 2025 13:03:57 -0500 Subject: [PATCH 18/19] fix python 3.10 compat --- tests/unit/test_plugin_obj.py | 24 ++++++++---------------- 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/tests/unit/test_plugin_obj.py b/tests/unit/test_plugin_obj.py index 647422457..d407be3ac 100644 --- a/tests/unit/test_plugin_obj.py +++ b/tests/unit/test_plugin_obj.py @@ -40,9 +40,7 @@ def test_helpers_denominate(): assert helpers._denominate(1e23) == "90949470177.29 TB" -def test_obj_action_triggers_key_cleanup_and_deletes_stale_key( - monkeypatch: MonkeyPatch, -): +def test_obj_action_triggers_key_cleanup_and_deletes_stale_key(): now = int(time.time()) stale_timestamp = ( now - 31 * 24 * 60 * 60 @@ -78,7 +76,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): # OBJ plugin & CLI config mocks with ( - patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch("linodecli.plugins.obj.CLI") as MockCLI, patch.dict( obj.COMMAND_MAP, { @@ -122,9 +120,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): ), "Fresh key should not be deleted" -def test_obj_action_triggers_key_rotation( - monkeypatch: MonkeyPatch, -): +def test_obj_action_triggers_key_rotation(): now = int(time.time()) # Key created 31 days ago, rotation period is 30 days old_timestamp = now - 60 * 60 * 24 * 31 @@ -153,7 +149,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): # OBJ plugin & CLI config mocks with ( - patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch("linodecli.plugins.obj.CLI") as MockCLI, patch.dict( obj.COMMAND_MAP, { @@ -209,9 +205,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): ), "Old key was not deleted after rotation" -def test_obj_action_does_not_trigger_cleanup_if_recent( - monkeypatch: MonkeyPatch, -): +def test_obj_action_does_not_trigger_cleanup_if_recent(): now = int(time.time()) # Set last cleanup to 1 hour ago (less than 24h) last_cleanup = now - 60 * 60 @@ -239,7 +233,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): return 200, {} with ( - patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch("linodecli.plugins.obj.CLI") as MockCLI, patch.dict( obj.COMMAND_MAP, { @@ -281,9 +275,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): ), "Cleanup should not be performed if it was done in the last 24 hours" -def test_obj_action_does_not_trigger_cleanup_if_disabled( - monkeypatch: MonkeyPatch, -): +def test_obj_action_does_not_trigger_cleanup_if_disabled(): now = int(time.time()) stale_timestamp = now - 31 * 24 * 60 * 60 stale_key = { @@ -308,7 +300,7 @@ def call_operation_side_effect(resource, action, *args, **kwargs): return 200, {} with ( - patch("linodecli.plugins.obj.__init__.CLI") as MockCLI, + patch("linodecli.plugins.obj.CLI") as MockCLI, patch.dict( obj.COMMAND_MAP, { From 9eb1f510243f8ca42b2e59017d7ec9b7dda6a5ca Mon Sep 17 00:00:00 2001 From: Lena Garber Date: Thu, 11 Dec 2025 13:40:04 -0500 Subject: [PATCH 19/19] fix lint --- .pylintrc | 4 ---- tests/unit/test_plugin_obj.py | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/.pylintrc b/.pylintrc index 03ddde567..1329f16e7 100644 --- a/.pylintrc +++ b/.pylintrc @@ -89,10 +89,6 @@ py-version=3.9 # Discover python modules and packages in the file system subtree. recursive=no -# When enabled, pylint would attempt to guess common misconfiguration and emit -# user-friendly hints instead of false-positive error messages. -suggestion-mode=yes - # Allow loading of arbitrary C extensions. Extensions are imported into the # active Python interpreter and may run arbitrary code. unsafe-load-any-extension=no diff --git a/tests/unit/test_plugin_obj.py b/tests/unit/test_plugin_obj.py index d407be3ac..608508099 100644 --- a/tests/unit/test_plugin_obj.py +++ b/tests/unit/test_plugin_obj.py @@ -1,7 +1,7 @@ import time from unittest.mock import patch -from pytest import CaptureFixture, MonkeyPatch +from pytest import CaptureFixture from linodecli import CLI, plugins from linodecli.plugins import obj