Migrate integration tests to unit tests

Author: lgarber-akamai

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,17 +736,18 @@ 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)
 
     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

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"