TPT-4472: Create tests for stream commands

tests/integration/helpers.py

@@ -54,6 +54,7 @@
     "regions",
     "ssh",
     "stackscripts",
+    "streams",
     "tickets",
     "tags",
     "users",

tests/integration/streams/fixtures.py

@@ -0,0 +1,179 @@
+import json
+import os
+from typing import Any, Dict
+
+import pytest
+import requests
+
+from linodecli import ExitCodes
+from linodecli.cli import logger
+from tests.integration.helpers import (
+    BASE_CMDS,
+    delete_target_id,
+    exec_test_command,
+    get_random_region_with_caps,
+    get_random_text,
+)
+
+
+def get_object_storage_buckets() -> Dict[str, Any]:
+    token = os.getenv("LINODE_CLI_TOKEN")
+    if token is None:
+        raise ValueError("LINODE_CLI_TOKEN environment variable is not set. ")
+    host = os.getenv("LINODE_CLI_API_HOST")
+    if token is None:
+        raise ValueError(
+            "LINODE_CLI_API_HOST environment variable is not set. "
+        )
+    version = os.getenv("LINODE_CLI_API_VERSION")
+    if token is None:
+        raise ValueError(
+            "LINODE_CLI_API_VERSION environment variable is not set. "
+        )
+
+    url = f"https://{host}/{version}/object-storage/buckets"
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+
+    response = requests.get(url, headers=headers, timeout=30)
+    response.raise_for_status()
+
+    return response.json()
+
+
+@pytest.fixture(scope="function")
+def create_stream(create_destination_akamai_object_storage_type):
+    label = get_random_text(8) + "_stream_cli_test"
+    result_create = exec_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "create",
+            "--label",
+            label,
+            "--type",
+            "audit_logs",
+            "--destinations",
+            str(create_destination_akamai_object_storage_type["id"]),
+            "--json",
+        ],
+    )
+    result_create_json = json.loads(result_create)[0]
+    yield result_create_json
+    wait_for_stream_status(str(result_create_json["id"]), "active")
+    delete_target_id(
+        "streams",
+        str(result_create_json["id"]),
+    )
+
+
+@pytest.fixture
+def create_object_storage_keys():
+    label = get_random_text(8) + "_destination_cli_test"
+    test_bucket = get_object_storage_buckets()["data"][0]
+    result_keys = exec_test_command(
+        BASE_CMDS["object-storage"]
+        + [
+            "keys-create",
+            "--label",
+            label,
+            "--bucket_access",
+            '[{"region": "'
+            + test_bucket["region"]
+            + '", "bucket_name": "'
+            + test_bucket["label"]
+            + '", "permissions": "read_write" }]',
+            "--text",
+            "--no-headers",
+            "--delimiter",
+            ",",
+            "--format",
+            "access_key,secret_key,id",
+        ]
+    ).split(",")
+    yield result_keys[0], result_keys[1], test_bucket["label"], test_bucket[
+        "s3_endpoint"
+    ]
+    delete_target_id("object-storage", str(result_keys[2]), "keys-delete")
+
+
+@pytest.fixture(scope="function")
+def create_destination_akamai_object_storage_type(create_object_storage_keys):
+    get_random_region_with_caps(required_capabilities=["Linodes"])
+
+    label = get_random_text(8) + "_destination_cli_test"
+    result_create_destination = exec_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-create",
+            "--label",
+            label,
+            "--type",
+            "akamai_object_storage",
+            "--details.host",
+            create_object_storage_keys[3],
+            "--details.bucket_name",
+            create_object_storage_keys[2],
+            "--details.path",
+            "audit-logs",
+            "--details.access_key_secret",
+            create_object_storage_keys[1],
+            "--details.access_key_id",
+            create_object_storage_keys[0],
+            "--delimiter",
+            ",",
+            "--json",
+        ]
+    )
+    create_destination_json = json.loads(result_create_destination)
+    yield create_destination_json[0]
+    try:
+        exec_test_command(
+            BASE_CMDS["streams"]
+            + [
+                "destination-view",
+                str(create_destination_json[0]["id"]),
+                "--delimiter",
+                ",",
+                "--text",
+            ],
+        )
+    except RuntimeError:
+        logger.debug("Destination already deleted, skipping deletion.")
+    else:
+        delete_target_id(
+            "streams",
+            str(create_destination_json[0]["id"]),
+            "destination-delete",
+        )
+
+
+def wait_for_stream_status(
+    stream_id: "str", status: "str", timeout=1800, interval=10
+):
+    import time
+
+    must_end = time.time() + timeout
+    while time.time() < must_end:
+        time.sleep(interval)
+        try:
+            result = exec_test_command(
+                BASE_CMDS["streams"]
+                + [
+                    "view",
+                    stream_id,
+                    "--format",
+                    "status",
+                    "--text",
+                    "--no-headers",
+                ]
+            )
+        except RuntimeError as response_error:
+            if "Not found" in str(response_error):
+                continue
+            else:
+                raise RuntimeError(response_error)
+        if status == result:
+            return True
+    return False

tests/integration/streams/test_destination.py

@@ -0,0 +1,178 @@
+import json
+
+import pytest
+
+from linodecli.exit_codes import ExitCodes
+from tests.integration.helpers import (
+    BASE_CMDS,
+    assert_headers_in_lines,
+    exec_failing_test_command,
+    exec_test_command,
+)
+from tests.integration.streams.fixtures import (  # noqa: F401
+    create_destination_akamai_object_storage_type,
+    create_object_storage_keys,
+)
+
+
+@pytest.mark.smoke
+def test_list_destinations():
+    result = exec_test_command(
+        BASE_CMDS["streams"]
+        + ["destinations-list", "--delimiter", ",", "--text"]
+    )
+    lines = result.splitlines()
+    headers = [
+        "created",
+        "created_by",
+        "details",
+        "id",
+        "label",
+        "status",
+        "type",
+        "updated",
+        "updated_by",
+        "version",
+    ]
+    assert_headers_in_lines(headers, lines)
+
+
+def test_create_destination_error():
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-create",
+            "--label",
+            "test",
+            "--type",
+            "custom_https",
+            "--details",
+            "test",
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "ambiguous option: --details could match" in result
+
+
+def test_view_destination_error():
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-view",
+            "1",
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "Request failed: 404" in result
+    assert "Destination not found" in result
+
+
+def test_destination_history_view_error():
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-history-view",
+            "1",
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "Request failed: 404" in result
+    assert "Destination not found" in result
+
+
+def test_update_destination_error():
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-update",
+            "1",
+            "--details",
+            "test",
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "Request failed: 404" in result
+    assert "Destination not found" in result
+
+
+def test_remove_destination_error():
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-delete",
+            "1",
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "Request failed: 404" in result
+    assert "Destination not found" in result
+
+
+def test_create_view_delete_destination(
+    create_destination_akamai_object_storage_type,
+):
+    result_view = exec_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-view",
+            str(create_destination_akamai_object_storage_type["id"]),
+            "--json",
+        ],
+    )
+    result_view_json = json.loads(result_view)[0]
+    assert result_view_json["status"] == "active"
+    assert result_view_json["type"] == "akamai_object_storage"
+    assert result_view_json["version"] == 1
+
+    result_history_view = exec_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-history-view",
+            str(create_destination_akamai_object_storage_type["id"]),
+            "--json",
+        ],
+    )
+    result_history_view_json = json.loads(result_history_view)[0]
+    assert result_history_view_json["status"] == "active"
+    assert result_history_view_json["type"] == "akamai_object_storage"
+    assert result_history_view_json["version"] == 1
+
+    exec_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-delete",
+            str(create_destination_akamai_object_storage_type["id"]),
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+    )
+
+    result = exec_failing_test_command(
+        BASE_CMDS["streams"]
+        + [
+            "destination-view",
+            str(create_destination_akamai_object_storage_type["id"]),
+            "--delimiter",
+            ",",
+            "--text",
+        ],
+        expected_code=ExitCodes.REQUEST_FAILED,
+    )
+    assert "Request failed: 404" in result
+    assert "Destination not found" in result