Hello, World!
" + "" + ) + + +@pytest.fixture +def static_site_error(): + return ( + "" + "" + "Error!
" + "" + ) + + +@pytest.fixture(scope="session") +def keys(): + response = json.loads( + exec_test_command( + CLI_CMD + + [ + "keys-create", + "--label", + "cli-integration-test-obj-key", + "--json", + ], + ).stdout.decode() + )[0] + _keys = Keys( + access_key=response.get("access_key"), + secret_key=response.get("secret_key"), + ) + yield _keys + exec_test_command(CLI_CMD + ["keys-delete", str(response.get("id"))]) + + +@pytest.fixture(scope="session") +def test_key(): + label = get_random_text(10) + key = ( + exec_test_command( + CLI_CMD + + [ + "keys-create", + "--label", + label, + "--text", + "--no-headers", + "--format=id", + ] + ) + .stdout.decode() + .strip() + ) + + yield key + + exec_test_command(CLI_CMD + ["keys-delete", key]) diff --git a/tests/integration/obj/test_obj_bucket.py b/tests/integration/obj/test_obj_bucket.py deleted file mode 100644 index 17f86dae4..000000000 --- a/tests/integration/obj/test_obj_bucket.py +++ /dev/null @@ -1,148 +0,0 @@ -import time - -import pytest - -from tests.integration.helpers import ( - assert_headers_in_lines, - delete_target_id, - exec_test_command, -) - -BASE_CMD = ["linode-cli", "object-storage"] - - -def test_clusters_list(): - res = ( - exec_test_command( - BASE_CMD + ["clusters-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() - ) - lines = res.splitlines() - headers = ["domain", "status", "region"] - assert_headers_in_lines(headers, lines) - - -@pytest.fixture -def get_cluster_id(): - cluster_id = ( - exec_test_command( - BASE_CMD - + [ - "clusters-list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - first_id = cluster_id[0] - yield first_id - - -def test_clusters_view(get_cluster_id): - cluster_id = get_cluster_id - res = ( - exec_test_command( - BASE_CMD + ["clusters-view", cluster_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() - ) - lines = res.splitlines() - headers = ["domain", "status", "region"] - assert_headers_in_lines(headers, lines) - - -def test_create_obj_storage_key(): - new_label = str(time.time_ns()) + "label" - exec_test_command( - BASE_CMD - + [ - "keys-create", - "--label", - new_label, - "--text", - "--no-headers", - ] - ) - - -def test_obj_storage_key_list(): - res = ( - exec_test_command(BASE_CMD + ["keys-list", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() - ) - lines = res.splitlines() - headers = ["label", "access_key", "secret_key"] - assert_headers_in_lines(headers, lines) - - -@pytest.fixture -def get_key_id(): - key_id = ( - exec_test_command( - BASE_CMD - + [ - "keys-list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - first_id = key_id[0] - yield first_id - - -def test_obj_storage_key_view(get_key_id): - key_id = get_key_id - res = ( - exec_test_command( - BASE_CMD + ["keys-view", key_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() - ) - lines = res.splitlines() - headers = ["label", "access_key", "secret_key"] - assert_headers_in_lines(headers, lines) - - -def test_obj_storage_key_update(get_key_id): - key_id = get_key_id - new_label = str(time.time_ns()) + "label" - updated_label = ( - exec_test_command( - BASE_CMD - + [ - "keys-update", - key_id, - "--label", - new_label, - "--text", - "--no-headers", - "--format=label", - ] - ) - .stdout.decode() - .rstrip() - ) - assert new_label == updated_label - delete_target_id( - target="object-storage", delete_command="keys-delete", id=key_id - ) diff --git a/tests/integration/obj/test_obj_plugin.py b/tests/integration/obj/test_obj_plugin.py index 04930cfe9..abaf3fb26 100644 --- a/tests/integration/obj/test_obj_plugin.py +++ b/tests/integration/obj/test_obj_plugin.py @@ -1,110 +1,14 @@ -import json -import logging from concurrent.futures import ThreadPoolExecutor, wait -from dataclasses import dataclass from typing import Callable, Optional import pytest import requests from pytest import MonkeyPatch -from linodecli.plugins.obj import ENV_ACCESS_KEY_NAME, ENV_SECRET_KEY_NAME from linodecli.plugins.obj.list import TRUNCATED_MSG from tests.integration.fixture_types import GetTestFilesType, GetTestFileType from tests.integration.helpers import count_lines, exec_test_command - -REGION = "us-southeast-1" -CLI_CMD = ["linode-cli", "object-storage"] -BASE_CMD = ["linode-cli", "obj", "--cluster", REGION] - - -@dataclass -class Keys: - access_key: str - secret_key: str - - -@pytest.fixture -def static_site_index(): - return ( - "" - "" - "Hello, World!
" - "" - ) - - -@pytest.fixture -def static_site_error(): - return ( - "" - "" - "Error!
" - "" - ) - - -@pytest.fixture(scope="session") -def keys(): - response = json.loads( - exec_test_command( - CLI_CMD - + [ - "keys-create", - "--label", - "cli-integration-test-obj-key", - "--json", - ], - ).stdout.decode() - )[0] - _keys = Keys( - access_key=response.get("access_key"), - secret_key=response.get("secret_key"), - ) - yield _keys - exec_test_command(CLI_CMD + ["keys-delete", str(response.get("id"))]) - - -def patch_keys(keys: Keys, monkeypatch: MonkeyPatch): - assert keys.access_key is not None - assert keys.secret_key is not None - monkeypatch.setenv(ENV_ACCESS_KEY_NAME, keys.access_key) - monkeypatch.setenv(ENV_SECRET_KEY_NAME, keys.secret_key) - - -@pytest.fixture -def create_bucket( - name_generator: Callable, keys: Keys, monkeypatch: MonkeyPatch -): - created_buckets = set() - patch_keys(keys, monkeypatch) - - def _create_bucket(bucket_name: Optional[str] = None): - if not bucket_name: - bucket_name = name_generator("test-bk") - - exec_test_command(BASE_CMD + ["mb", bucket_name]) - created_buckets.add(bucket_name) - return bucket_name - - yield _create_bucket - for bk in created_buckets: - try: - delete_bucket(bk) - except Exception as e: - logging.exception(f"Failed to cleanup bucket: {bk}, {e}") - - -def delete_bucket(bucket_name: str, force: bool = True): - args = BASE_CMD + ["rb", bucket_name] - if force: - args.append("--recursive") - exec_test_command(args) - return bucket_name +from tests.integration.obj.conftest import BASE_CMD, REGION, Keys, patch_keys def test_obj_single_file_single_bucket( @@ -452,6 +356,6 @@ def test_generate_url( BASE_CMD + ["signurl", bucket, test_file.name, "+300"] ) url = process.stdout.decode() - response = requests.get(url) + response = requests.get(url.strip("\n")) assert response.text == content assert response.status_code == 200 diff --git a/tests/integration/obj/test_object_storage.py b/tests/integration/obj/test_object_storage.py new file mode 100644 index 000000000..8f45041fe --- /dev/null +++ b/tests/integration/obj/test_object_storage.py @@ -0,0 +1,307 @@ +import json +from typing import Callable, Optional + +import pytest +from pytest import MonkeyPatch + +from tests.integration.helpers import exec_test_command, get_random_text +from tests.integration.obj.conftest import CLI_CMD, REGION, Keys + + +def test_clusters_list(): + response = ( + exec_test_command(CLI_CMD + ["clusters-list", "--json"]) + .stdout.decode() + .rstrip() + ) + + clusters = json.loads(response) + + assert isinstance(clusters, list) + assert len(clusters) > 0 + + for cluster in clusters: + assert isinstance(cluster, dict) + assert { + "id", + "region", + "status", + "domain", + "static_site_domain", + }.issubset(cluster.keys()) + + assert cluster["id"] + assert cluster["region"] + assert cluster["status"] in {"available", "unavailable"} + assert cluster["domain"].endswith(".linodeobjects.com") + assert cluster["static_site_domain"].startswith("website-") + + +def test_clusters_view(): + response = ( + exec_test_command(CLI_CMD + ["clusters-view", REGION, "--json"]) + .stdout.decode() + .rstrip() + ) + + clusters = json.loads(response) + + assert isinstance(clusters, list) + assert len(clusters) == 1 + + for cluster in clusters: + assert isinstance(cluster, dict) + assert { + "id", + "region", + "status", + "domain", + "static_site_domain", + }.issubset(cluster.keys()) + + assert cluster["id"] == "us-southeast-1" + assert cluster["region"] == "us-southeast" + assert cluster["status"] in {"available", "unavailable"} + assert cluster["domain"].endswith(".linodeobjects.com") + assert cluster["static_site_domain"].startswith("website-") + + +def test_keys_create( + create_bucket: Callable[[Optional[str]], str], + keys: Keys, + monkeypatch: MonkeyPatch, +): + bucket_name = create_bucket() + region = "us-southeast" # Fixed typo + label = get_random_text(10) + response = ( + exec_test_command( + CLI_CMD + + [ + "keys-create", + "--label", + label, + "--bucket_access", + f'[{{"region": "{region}", "bucket_name": "{bucket_name}", "permissions": "read_write"}}]', + "--json", + ] + ) + .stdout.decode() + .rstrip() + ) + + data = json.loads(response) + key = data[0] + + assert key["id"] > 0 + assert key["label"] == label + assert key["access_key"] + assert key["secret_key"] + assert key["limited"] is True + + bucket = key["bucket_access"][0] + assert bucket["cluster"] == "us-southeast-1" + assert bucket["bucket_name"].startswith("test-bk") + assert bucket["permissions"] == "read_write" + assert bucket["region"] == "us-southeast" + + region_info = key["regions"][0] + assert region_info["id"] == "us-southeast" + assert region_info["s3_endpoint"].endswith(".linodeobjects.com") + assert region_info["endpoint_type"] == "E0" + + exec_test_command(CLI_CMD + ["keys-delete", str(key["id"])]) + + +def test_keys_delete(): + label = get_random_text(10) + + key = ( + exec_test_command( + CLI_CMD + + [ + "keys-create", + "--label", + label, + "--text", + "--no-headers", + "--format=id", + ] + ) + .stdout.decode() + .strip() + ) + + assert key, "Key creation failed, received empty key ID" + + # Delete the key + exec_test_command(CLI_CMD + ["keys-delete", key]) + + # Verify deletion by listing keys + keys_list = ( + exec_test_command(CLI_CMD + ["keys-list", "--text"]) + .stdout.decode() + .strip() + ) + + assert key not in keys_list, f"Key {key} still exists after deletion!" + + +def test_keys_list(test_key): + keys_list = ( + exec_test_command(CLI_CMD + ["keys-list", "--text"]) + .stdout.decode() + .strip() + ) + + assert test_key in keys_list + + +def test_keys_update(test_key): + update_label = get_random_text(10) + + updated_key_resp = ( + exec_test_command( + CLI_CMD + + [ + "keys-update", + test_key, + "--label", + update_label, + "--region", + "us-east", + "--json", + ] + ) + .stdout.decode() + .strip() + ) + + assert update_label in updated_key_resp + + +def test_keys_view(test_key): + view_resp = ( + exec_test_command( + CLI_CMD + + [ + "keys-view", + test_key, + "--json", + ] + ) + .stdout.decode() + .strip() + ) + + data = json.loads(view_resp) + + key = data[0] + + assert key["id"] > 0 + assert isinstance(key["label"], str) and key["label"] + assert key["access_key"] + assert key["secret_key"] == "[REDACTED]" + assert isinstance(key["limited"], bool) + + region = key["regions"][0] + assert region["id"] == "us-east" + assert region["s3_endpoint"].endswith(".linodeobjects.com") + assert region["endpoint_type"] in {"E0", "E1", "E2", "E3"} + + +def test_types(): + data = ( + exec_test_command( + CLI_CMD + + [ + "types", + "--json", + ] + ) + .stdout.decode() + .strip() + ) + + types = json.loads(data) + + assert isinstance(types, list) and len(types) > 0 + + for type in types: + assert "id" in type and isinstance(type["id"], str) and type["id"] + assert ( + "label" in type and isinstance(type["label"], str) and type["label"] + ) + assert "price" in type and isinstance(type["price"], dict) + assert "hourly" in type["price"] and isinstance( + type["price"]["hourly"], (int, float) + ) + assert "monthly" in type["price"] and ( + type["price"]["monthly"] is None + or isinstance(type["price"]["monthly"], (int, float)) + ) + assert "transfer" in type and isinstance(type["transfer"], int) + + if "region_prices" in type: + assert isinstance(type["region_prices"], list) + for region_price in type["region_prices"]: + assert ( + "id" in region_price + and isinstance(region_price["id"], str) + and region_price["id"] + ) + assert "hourly" in region_price and isinstance( + region_price["hourly"], (int, float) + ) + assert "monthly" in region_price and ( + region_price["monthly"] is None + or isinstance(region_price["monthly"], (int, float)) + ) + + +def test_endpoints(): + data = ( + exec_test_command( + CLI_CMD + + [ + "endpoints", + "--json", + ] + ) + .stdout.decode() + .strip() + ) + + endpoints = json.loads(data) + + assert isinstance(endpoints, list) + assert all("region" in e for e in endpoints) + assert all("endpoint_type" in e for e in endpoints) + assert all("s3_endpoint" in e for e in endpoints) + + us_east = next(e for e in endpoints if e["region"] == "us-east") + assert us_east["endpoint_type"] == "E0" + assert us_east["s3_endpoint"] == "us-east-1.linodeobjects.com" + + +@pytest.mark.skipif( + reason="Skipping until the command is fixed and aligned with techdocs example. Applicable for spec version after 4.197.1" +) +def test_transfers(): + data = ( + exec_test_command( + CLI_CMD + + [ + "transfers", + "--json", + ] + ) + .stdout.decode() + .strip() + ) + + transfers = json.loads(data) + + assert len(transfers) > 0 + assert "used" in transfers[0] + assert isinstance(transfers[0]["used"], int)