diff --git a/tests/integration/account/test_account.py b/tests/integration/account/test_account.py index fb173d2a2..f1fae5dca 100644 --- a/tests/integration/account/test_account.py +++ b/tests/integration/account/test_account.py @@ -1,15 +1,15 @@ import pytest -from tests.integration.helpers import assert_headers_in_lines, exec_test_command - -BASE_CMD = ["linode-cli", "account"] +from tests.integration.helpers import ( + BASE_CMDS, + assert_headers_in_lines, + exec_test_command, +) def test_account_transfer(): - res = ( - exec_test_command(BASE_CMD + ["transfer", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["transfer", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -18,12 +18,8 @@ def test_account_transfer(): def test_available_service(): - res = ( - exec_test_command( - BASE_CMD + ["get-availability", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["get-availability", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -32,13 +28,9 @@ def test_available_service(): def test_region_availability(): - res = ( - exec_test_command( - BASE_CMD - + ["get-account-availability", "us-east", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + + ["get-account-availability", "us-east", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -47,12 +39,8 @@ def test_region_availability(): def test_event_list(): - res = ( - exec_test_command( - ["linode-cli", "events", "list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + ["linode-cli", "events", "list", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["entity.label", "username"] @@ -61,43 +49,34 @@ def test_event_list(): @pytest.fixture def get_event_id(): - event_id = ( - exec_test_command( - [ - "linode-cli", - "events", - "list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + event_id = exec_test_command( + [ + "linode-cli", + "events", + "list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() first_id = event_id[0].split(",")[0] yield first_id def test_event_view(get_event_id): event_id = get_event_id - res = ( - exec_test_command( - [ - "linode-cli", - "events", - "view", - event_id, - "--text", - "--delimiter=,", - ] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + [ + "linode-cli", + "events", + "view", + event_id, + "--text", + "--delimiter=,", + ] ) lines = res.splitlines() @@ -117,7 +96,6 @@ def test_event_read(get_event_id): "--delimiter=,", ] ) - assert process.returncode == 0 def test_event_seen(get_event_id): @@ -132,16 +110,11 @@ def test_event_seen(get_event_id): "--delimiter=,", ] ) - assert process.returncode == 0 def test_account_invoice_list(): - res = ( - exec_test_command( - BASE_CMD + ["invoices-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["invoices-list", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["billing_source", "tax", "subtotal"] @@ -150,35 +123,27 @@ def test_account_invoice_list(): @pytest.fixture def get_invoice_id(): - invoice_id = ( - exec_test_command( - BASE_CMD - + [ - "invoices-list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + invoice_id = exec_test_command( + BASE_CMDS["account"] + + [ + "invoices-list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() first_id = invoice_id[0] yield first_id def test_account_invoice_view(get_invoice_id): invoice_id = get_invoice_id - res = ( - exec_test_command( - BASE_CMD + ["invoice-view", invoice_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + + ["invoice-view", invoice_id, "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -188,12 +153,9 @@ def test_account_invoice_view(get_invoice_id): def test_account_invoice_items(get_invoice_id): invoice_id = get_invoice_id - res = ( - exec_test_command( - BASE_CMD + ["invoice-items", invoice_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + + ["invoice-items", invoice_id, "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -202,10 +164,8 @@ def test_account_invoice_items(get_invoice_id): def test_account_logins_list(): - res = ( - exec_test_command(BASE_CMD + ["logins-list", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["logins-list", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["ip", "username", "status"] @@ -214,35 +174,27 @@ def test_account_logins_list(): @pytest.fixture def get_login_id(): - login_id = ( - exec_test_command( - BASE_CMD - + [ - "logins-list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + login_id = exec_test_command( + BASE_CMDS["account"] + + [ + "logins-list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() first_id = login_id[0] yield first_id def test_account_login_view(get_login_id): login_id = get_login_id - res = ( - exec_test_command( - BASE_CMD + ["login-view", login_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + + ["login-view", login_id, "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -251,10 +203,8 @@ def test_account_login_view(get_login_id): def test_account_setting_view(): - res = ( - exec_test_command(BASE_CMD + ["settings", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["settings", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -263,12 +213,8 @@ def test_account_setting_view(): def test_user_list(): - res = ( - exec_test_command( - ["linode-cli", "users", "list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + ["linode-cli", "users", "list", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["email", "username"] @@ -277,35 +223,26 @@ def test_user_list(): @pytest.fixture def username(): - user_id = ( - exec_test_command( - [ - "linode-cli", - "users", - "list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "username", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + user_id = exec_test_command( + [ + "linode-cli", + "users", + "list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "username", + ] + ).splitlines() first_id = user_id[0].split(",")[0] yield first_id def test_user_view(username: str): - res = ( - exec_test_command( - ["linode-cli", "users", "view", username, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + ["linode-cli", "users", "view", username, "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -314,12 +251,8 @@ def test_user_view(username: str): def test_payment_method_list(): - res = ( - exec_test_command( - ["linode-cli", "payment-methods", "list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + ["linode-cli", "payment-methods", "list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -328,12 +261,8 @@ def test_payment_method_list(): def test_payment_list(): - res = ( - exec_test_command( - BASE_CMD + ["payments-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["payments-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -342,18 +271,14 @@ def test_payment_list(): def test_service_transfers(): - res = ( - exec_test_command( - [ - "linode-cli", - "service-transfers", - "list", - "--text", - "--delimiter=,", - ] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + [ + "linode-cli", + "service-transfers", + "list", + "--text", + "--delimiter=,", + ] ) lines = res.splitlines() @@ -362,12 +287,8 @@ def test_service_transfers(): def test_maintenance_list(): - res = ( - exec_test_command( - BASE_CMD + ["maintenance-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["maintenance-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -376,12 +297,8 @@ def test_maintenance_list(): def test_notifications_list(): - res = ( - exec_test_command( - BASE_CMD + ["notifications-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["notifications-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -390,12 +307,8 @@ def test_notifications_list(): def test_clients_list(): - res = ( - exec_test_command( - BASE_CMD + ["clients-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["account"] + ["clients-list", "--text", "--delimiter=,"] ) lines = res.splitlines() diff --git a/tests/integration/beta/helpers.py b/tests/integration/beta/helpers.py new file mode 100644 index 000000000..adcae9604 --- /dev/null +++ b/tests/integration/beta/helpers.py @@ -0,0 +1,23 @@ +from tests.integration.helpers import ( + BASE_CMDS, + exec_test_command, +) + + +def get_beta_id(): + beta_ids = exec_test_command( + BASE_CMDS["betas"] + + [ + "list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() + if not beta_ids or beta_ids == [""]: + pytest.skip("No betas available to test.") + + return beta_ids[0] if beta_ids else None diff --git a/tests/integration/beta/test_beta_program.py b/tests/integration/beta/test_beta_program.py index 240272dff..5dbe7da08 100644 --- a/tests/integration/beta/test_beta_program.py +++ b/tests/integration/beta/test_beta_program.py @@ -1,15 +1,16 @@ import pytest -from tests.integration.helpers import assert_headers_in_lines, exec_test_command - -BASE_CMD = ["linode-cli", "betas"] +from tests.integration.beta.helpers import get_beta_id +from tests.integration.helpers import ( + BASE_CMDS, + assert_headers_in_lines, + exec_test_command, +) def test_beta_list(): - res = ( - exec_test_command(BASE_CMD + ["list", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["betas"] + ["list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -20,43 +21,13 @@ def test_beta_list(): assert_headers_in_lines(headers, lines) -@pytest.fixture -def get_beta_id(): - beta_ids = ( - exec_test_command( - BASE_CMD - + [ - "list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - if not beta_ids or beta_ids == [""]: - pytest.skip("No betas available to test.") - - first_id = beta_ids[0] - yield first_id - - -def test_beta_view(get_beta_id): - beta_id = get_beta_id +def test_beta_view(): + beta_id = get_beta_id() if beta_id is None: pytest.skip("No beta program available to test") else: - res = ( - exec_test_command( - BASE_CMD + ["view", beta_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["betas"] + ["view", beta_id, "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["label", "description"] @@ -64,10 +35,8 @@ def test_beta_view(get_beta_id): def test_beta_enrolled(): - res = ( - exec_test_command(BASE_CMD + ["enrolled", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["betas"] + ["enrolled", "--text", "--delimiter=,"] ) lines = res.splitlines() diff --git a/tests/integration/cli/test_help.py b/tests/integration/cli/test_help.py index 5b7108cfe..50b266f19 100644 --- a/tests/integration/cli/test_help.py +++ b/tests/integration/cli/test_help.py @@ -11,8 +11,7 @@ @pytest.mark.smoke def test_help_page_for_non_aliased_actions(): - process = exec_test_command(["linode-cli", "linodes", "list", "--help"]) - output = process.stdout.decode() + output = exec_test_command(["linode-cli", "linodes", "list", "--help"]) wrapped_output = textwrap.fill(output, width=180).replace("\n", "") assert contains_at_least_one_of( @@ -22,8 +21,8 @@ def test_help_page_for_non_aliased_actions(): assert contains_at_least_one_of( wrapped_output, [ - "API Documentation: https://www.linode.com/docs/api/linode-instances/#linodes-list", - "API Documentation: https://techdocs.akamai.com/linode-api/reference/get-linode-instances", + "API Documentation:", + "https://techdocs.akamai.com/linode-api/reference/", ], ) @@ -33,8 +32,7 @@ def test_help_page_for_non_aliased_actions(): @pytest.mark.smoke def test_help_page_for_aliased_actions(): - process = exec_test_command(["linode-cli", "linodes", "ls", "--help"]) - output = process.stdout.decode() + output = exec_test_command(["linode-cli", "linodes", "ls", "--help"]) wrapped_output = textwrap.fill(output, width=180).replace("\n", "") assert contains_at_least_one_of( @@ -44,8 +42,8 @@ def test_help_page_for_aliased_actions(): assert contains_at_least_one_of( wrapped_output, [ - "API Documentation: https://www.linode.com/docs/api/linode-instances/#linodes-list", - "API Documentation: https://techdocs.akamai.com/linode-api/reference/get-linode-instances", + "API Documentation:", + "https://techdocs.akamai.com/linode-api/reference/", ], ) @@ -72,7 +70,7 @@ def test_debug_output_contains_request_url(monkeypatch: pytest.MonkeyPatch): "12345", "--debug", ] - ).stderr.decode() + ) wrapped_output = textwrap.fill(output, width=180).replace("\n", "") assert ( diff --git a/tests/integration/cli/test_host_overrides.py b/tests/integration/cli/test_host_overrides.py index 9ccbf41da..421dcc192 100644 --- a/tests/integration/cli/test_host_overrides.py +++ b/tests/integration/cli/test_host_overrides.py @@ -9,10 +9,9 @@ def test_cli_command_fails_to_access_invalid_host(monkeypatch: MonkeyPatch): monkeypatch.setenv("LINODE_CLI_API_HOST", INVALID_HOST) - process = exec_failing_test_command( + output = exec_failing_test_command( ["linode-cli", "linodes", "ls"], ExitCodes.UNRECOGNIZED_COMMAND ) - output = process.stderr.decode() expected_output = ["Max retries exceeded with url:", "wrongapi.linode.com"] @@ -32,9 +31,8 @@ def test_cli_command_fails_to_access_invalid_api_scheme( monkeypatch: MonkeyPatch, ): monkeypatch.setenv("LINODE_CLI_API_SCHEME", "ssh") - process = exec_failing_test_command( + output = exec_failing_test_command( ["linode-cli", "linodes", "ls"], ExitCodes.UNRECOGNIZED_COMMAND ) - output = process.stderr.decode() assert "ssh://" in output diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 5e5ab8a96..9ee4d6c76 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -106,7 +106,7 @@ def create_inbound_rule(ipv4_address, ipv6_address): if is_valid_ipv4(ipv4_address) or is_valid_ipv6(ipv6_address): command.extend(["--rules.inbound", inbound_rule]) - firewall_id = exec_test_command(command).stdout.decode().rstrip() + firewall_id = exec_test_command(command) yield firewall_id @@ -201,68 +201,6 @@ def _generate_test_files( return _generate_test_files -# test helper specific to Domains test suite -@pytest.fixture -def master_domain(): - timestamp = str(time.time_ns()) - - domain_id = ( - exec_test_command( - DOMAIN_BASE_CMD - + [ - "create", - "--type", - "master", - "--domain", - timestamp + "example.com", - "--soa_email", - "pthiel_test@linode.com", - "--text", - "--no-header", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield domain_id - - delete_target_id("domains", id=domain_id) - - -@pytest.fixture -def slave_domain(): - timestamp = str(time.time_ns()) - - domain_id = ( - exec_test_command( - DOMAIN_BASE_CMD - + [ - "create", - "--type", - "slave", - "--domain", - timestamp + "-example.com", - "--master_ips", - "1.1.1.1", - "--text", - "--no-header", - "--delimiter", - ",", - "--format=id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield domain_id - - delete_target_id("domains", domain_id) - - # Test helpers specific to Linodes test suite @pytest.fixture def linode_with_label(linode_cloud_firewall): diff --git a/tests/integration/database/fixtures.py b/tests/integration/database/fixtures.py new file mode 100644 index 000000000..40212682d --- /dev/null +++ b/tests/integration/database/fixtures.py @@ -0,0 +1,78 @@ +import pytest + +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, + get_random_text, +) + + +@pytest.fixture(scope="module") +def postgresql_cluster(): + postgresql_database_label = get_random_text(5) + "_postgresql" + + database_id = exec_test_command( + BASE_CMDS["databases"] + + [ + "postgresql-create", + "--type", + "g6-nanode-1", + "--region", + "us-ord", + "--label", + postgresql_database_label, + "--engine", + "postgresql/16", + "--text", + "--delimiter", + ",", + "--no-headers", + "--format", + "id", + "--no-defaults", + "--format", + "id", + ] + ) + + yield database_id + + delete_target_id( + target="databases", delete_command="postgresql-delete", id=database_id + ) + + +@pytest.fixture(scope="module") +def mysql_cluster(): + mysql_database_label = get_random_text(5) + "_mysql" + + database_id = exec_test_command( + BASE_CMDS["databases"] + + [ + "mysql-create", + "--type", + "g6-nanode-1", + "--region", + "us-ord", + "--label", + mysql_database_label, + "--engine", + "mysql/8", + "--text", + "--delimiter", + ",", + "--no-headers", + "--format", + "id", + "--no-defaults", + "--format", + "id", + ] + ) + + yield database_id + + delete_target_id( + target="databases", delete_command="mysql-delete", id=database_id + ) diff --git a/tests/integration/database/helpers.py b/tests/integration/database/helpers.py new file mode 100644 index 000000000..d15cbe4ce --- /dev/null +++ b/tests/integration/database/helpers.py @@ -0,0 +1,36 @@ +from tests.integration.helpers import ( + BASE_CMDS, + exec_test_command, +) + + +def get_db_type_id(): + db_type_ids = exec_test_command( + BASE_CMDS["databases"] + + [ + "types", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() + return db_type_ids[0] if db_type_ids else None + + +def get_engine_id(): + engine_ids = exec_test_command( + BASE_CMDS["databases"] + + [ + "engines", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() + return engine_ids[0] if engine_ids else None diff --git a/tests/integration/database/test_database.py b/tests/integration/database/test_database.py index 3fc9843bb..a80a4fe05 100644 --- a/tests/integration/database/test_database.py +++ b/tests/integration/database/test_database.py @@ -1,165 +1,59 @@ -import time - -import pytest - +from tests.integration.database.fixtures import ( # noqa: F401 + mysql_cluster, + postgresql_cluster, +) +from tests.integration.database.helpers import get_db_type_id, get_engine_id from tests.integration.helpers import ( + BASE_CMDS, assert_headers_in_lines, - delete_target_id, exec_test_command, ) -from tests.integration.linodes.helpers_linodes import DEFAULT_LABEL - -BASE_CMD = ["linode-cli", "databases"] def test_engines_list(): - res = ( - exec_test_command(BASE_CMD + ["engines", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["databases"] + ["engines", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["id", "engine", "version"] assert_headers_in_lines(headers, lines) -timestamp = str(time.time_ns()) -mysql_database_label = DEFAULT_LABEL + "-mysql-" + timestamp -postgresql_database_label = DEFAULT_LABEL + "-postgresql-" + timestamp - - -@pytest.fixture(scope="package", autouse=True) -def test_mysql_cluster(): - database_id = ( - exec_test_command( - BASE_CMD - + [ - "mysql-create", - "--type", - "g6-nanode-1", - "--region", - "us-ord", - "--label", - mysql_database_label, - "--engine", - "mysql/8", - "--text", - "--delimiter", - ",", - "--no-headers", - "--format", - "id", - "--no-defaults", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield database_id - - delete_target_id( - target="databases", delete_command="mysql-delete", id=database_id - ) - - -def test_mysql_suspend_resume(test_mysql_cluster): - database_id = test_mysql_cluster +def test_mysql_suspend_resume(mysql_cluster): + database_id = mysql_cluster res = exec_test_command( - BASE_CMD + ["mysql-suspend", database_id, "--text", "--delimiter=,"] - ).stdout.decode() + BASE_CMDS["databases"] + + ["mysql-suspend", database_id, "--text", "--delimiter=,"] + ) assert "Request failed: 400" not in res res = exec_test_command( - BASE_CMD + ["mysql-resume", database_id, "--text", "--delimiter=,"] - ).stdout.decode() - assert "Request failed: 400" not in res - - -@pytest.fixture(scope="package", autouse=True) -def test_postgresql_cluster(): - database_id = ( - exec_test_command( - BASE_CMD - + [ - "postgresql-create", - "--type", - "g6-nanode-1", - "--region", - "us-ord", - "--label", - postgresql_database_label, - "--engine", - "postgresql/16", - "--text", - "--delimiter", - ",", - "--no-headers", - "--format", - "id", - "--no-defaults", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield database_id - - delete_target_id( - target="databases", delete_command="postgresql-delete", id=database_id + BASE_CMDS["databases"] + + ["mysql-resume", database_id, "--text", "--delimiter=,"] ) + assert "Request failed: 400" not in res -def test_postgresql_suspend_resume(test_postgresql_cluster): - database_id = test_postgresql_cluster +def test_postgresql_suspend_resume(postgresql_cluster): + database_id = postgresql_cluster res = exec_test_command( - BASE_CMD + BASE_CMDS["databases"] + ["postgresql-suspend", database_id, "--text", "--delimiter=,"] - ).stdout.decode() + ) assert "Request failed: 400" not in res res = exec_test_command( - BASE_CMD + ["postgresql-resume", database_id, "--text", "--delimiter=,"] - ).stdout.decode() - assert "Request failed: 400" not in res - - -@pytest.fixture -def get_engine_id(): - engine_id = ( - exec_test_command( - BASE_CMD - + [ - "engines", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() + BASE_CMDS["databases"] + + ["postgresql-resume", database_id, "--text", "--delimiter=,"] ) - first_id = engine_id[0] - yield first_id + assert "Request failed: 400" not in res -def test_engines_view(get_engine_id): - engine_id = get_engine_id - res = ( - exec_test_command( - BASE_CMD + ["engine-view", engine_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() +def test_engines_view(): + engine_id = get_engine_id() + res = exec_test_command( + BASE_CMDS["databases"] + + ["engine-view", engine_id, "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -169,10 +63,8 @@ def test_engines_view(get_engine_id): def test_databases_list(): - res = ( - exec_test_command(BASE_CMD + ["list", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["databases"] + ["list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -182,10 +74,8 @@ def test_databases_list(): def test_mysql_list(): - res = ( - exec_test_command(BASE_CMD + ["mysql-list", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["databases"] + ["mysql-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -196,12 +86,8 @@ def test_mysql_list(): def test_postgresql_list(): - res = ( - exec_test_command( - BASE_CMD + ["postgresql-list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["databases"] + ["postgresql-list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -212,47 +98,19 @@ def test_postgresql_list(): def test_databases_types(): - res = ( - exec_test_command(BASE_CMD + ["types", "--text", "--delimiter=,"]) - .stdout.decode() - .rstrip() + res = exec_test_command( + BASE_CMDS["databases"] + ["types", "--text", "--delimiter=,"] ) lines = res.splitlines() headers = ["id", "label", "_split"] assert_headers_in_lines(headers, lines) -@pytest.fixture -def get_node_id(): - node_id = ( - exec_test_command( - BASE_CMD - + [ - "types", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - first_id = node_id[0] - yield first_id - - -def test_databases_type_view(get_node_id): - node_id = get_node_id - res = ( - exec_test_command( - BASE_CMD + ["type-view", node_id, "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() +def test_databases_type_view(): + db_type_id = get_db_type_id() + res = exec_test_command( + BASE_CMDS["databases"] + + ["type-view", db_type_id, "--text", "--delimiter=,"] ) lines = res.splitlines() diff --git a/tests/integration/domains/fixtures.py b/tests/integration/domains/fixtures.py new file mode 100644 index 000000000..476e38042 --- /dev/null +++ b/tests/integration/domains/fixtures.py @@ -0,0 +1,100 @@ +import pytest + +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, + get_random_text, +) + + +@pytest.fixture(scope="function") +def master_domain(): + domain_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "create", + "--type", + "master", + "--domain", + get_random_text(5) + "-example.com", + "--soa_email", + "pthiel_test@linode.com", + "--text", + "--no-header", + "--format", + "id", + ] + ) + + yield domain_id + + delete_target_id("domains", id=domain_id) + + +@pytest.fixture(scope="function") +def slave_domain(): + domain_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "create", + "--type", + "slave", + "--domain", + get_random_text(5) + "-example.com", + "--master_ips", + "1.1.1.1", + "--text", + "--no-header", + "--delimiter", + ",", + "--format=id", + ] + ) + + yield domain_id + + delete_target_id("domains", domain_id) + + +@pytest.fixture(scope="function") +def domain_and_record(): + # Create domain + domain_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "create", + "--type", + "master", + "--domain", + get_random_text(5) + "-example.com", + "--soa_email=pthiel@linode.com", + "--text", + "--no-header", + "--format=id", + ] + ) + + # Create record + record_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "records-create", + "--protocol=tcp", + "--type=SRV", + "--port=23", + "--priority=4", + "--service=telnet", + "--target=record-setup", + "--weight=4", + "--text", + "--no-header", + "--delimiter=,", + "--format=id", + domain_id, + ] + ) + + yield domain_id, record_id + + delete_target_id(target="domains", id=domain_id) diff --git a/tests/integration/domains/test_master_domains.py b/tests/integration/domains/test_domain_master.py similarity index 64% rename from tests/integration/domains/test_master_domains.py rename to tests/integration/domains/test_domain_master.py index ee990b1ab..67649d528 100644 --- a/tests/integration/domains/test_master_domains.py +++ b/tests/integration/domains/test_domain_master.py @@ -4,44 +4,15 @@ import pytest from linodecli.exit_codes import ExitCodes +from tests.integration.domains.fixtures import ( # noqa: F401 + master_domain, +) from tests.integration.helpers import ( - delete_target_id, + BASE_CMDS, exec_failing_test_command, exec_test_command, ) -BASE_CMD = ["linode-cli", "domains"] - - -@pytest.fixture -def master_test_domain(): - timestamp = str(time.time_ns()) - # Create domain - master_domain_id = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--type", - "master", - "--domain", - "BC" + timestamp + "-example.com", - "--soa_email=pthiel" + timestamp + "@linode.com", - "--text", - "--no-header", - "--delimiter", - ",", - "--format=id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield master_domain_id - - delete_target_id(target="domains", id=master_domain_id) - def test_create_domain_fails_without_spcified_type(): timestamp = str(time.time_ns()) @@ -50,7 +21,7 @@ def test_create_domain_fails_without_spcified_type(): # not all output from the linode-cli goes to stdout, stderr result = exec_failing_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "create", "--domain", @@ -61,7 +32,7 @@ def test_create_domain_fails_without_spcified_type(): "--no-headers", ], expected_code=ExitCodes.REQUEST_FAILED, - ).stderr.decode() + ) assert "Request failed: 400" in result assert "type is required" in result @@ -70,7 +41,7 @@ def test_create_domain_fails_without_spcified_type(): def test_create_master_domain_fails_without_soa_email(): timestamp = str(time.time_ns()) result = exec_failing_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "create", "--type", @@ -81,7 +52,7 @@ def test_create_master_domain_fails_without_soa_email(): "--no-headers", ], expected_code=ExitCodes.REQUEST_FAILED, - ).stderr.decode() + ) assert "Request failed: 400" in result assert "soa_email soa_email required when type=master" in result @@ -93,15 +64,15 @@ def test_create_master_domain(master_domain): assert re.search("[0-9]+", domain_id) -def test_update_master_domain_soa_email(master_test_domain): +def test_update_master_domain_soa_email(master_domain): # Remove --master_ips param when 872 is resolved timestamp = str(time.time_ns()) new_soa_email = "pthiel_new@linode.com" - domain_id = master_test_domain + domain_id = master_domain result = exec_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "update", domain_id, @@ -115,14 +86,14 @@ def test_update_master_domain_soa_email(master_test_domain): "--text", "--no-header", ] - ).stdout.decode() + ) assert new_soa_email in result -def test_list_master_domain(master_test_domain): +def test_list_master_domain(master_domain): result = exec_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "list", "--format=id,domain,type,status", @@ -131,14 +102,14 @@ def test_list_master_domain(master_test_domain): "--delimiter", ",", ] - ).stdout.decode() + ) - assert re.search("[0-9]+,BC[0-9]+-example.com,master,active", result) + assert re.search("master,active", result) -def test_show_domain_detail(master_test_domain): +def test_show_domain_detail(master_domain): result = exec_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "list", "--format=id,domain,type,status", @@ -147,6 +118,6 @@ def test_show_domain_detail(master_test_domain): "--delimiter", ",", ] - ).stdout.decode() + ) - assert re.search("[0-9]+,BC[0-9]+-example.com,master,active", result) + assert re.search("master,active", result) diff --git a/tests/integration/domains/test_domain_records.py b/tests/integration/domains/test_domain_records.py index 74eb1e9ff..59e307918 100644 --- a/tests/integration/domains/test_domain_records.py +++ b/tests/integration/domains/test_domain_records.py @@ -3,116 +3,63 @@ import pytest +from tests.integration.domains.fixtures import ( # noqa: F401 + domain_and_record, + master_domain, + slave_domain, +) from tests.integration.helpers import ( - SUCCESS_STATUS_CODE, + BASE_CMDS, contains_at_least_one_of, - delete_target_id, exec_test_command, ) -BASE_CMD = ["linode-cli", "domains"] - - -@pytest.fixture -def test_domain_and_record(): - timestamp = str(time.time_ns()) - # Create domain - domain_id = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--type", - "master", - "--domain", - timestamp + "example.com", - "--soa_email=pthiel@linode.com", - "--text", - "--no-header", - "--format=id", - ] - ) - .stdout.decode() - .rstrip() - ) - - # Create record - record_id = ( - exec_test_command( - BASE_CMD - + [ - "records-create", - "--protocol=tcp", - "--type=SRV", - "--port=23", - "--priority=4", - "--service=telnet", - "--target=record-setup", - "--weight=4", - "--text", - "--no-header", - "--delimiter=,", - "--format=id", - domain_id, - ] - ) - .stdout.decode() - .rstrip() - ) - - yield domain_id, record_id - - delete_target_id(target="domains", id=domain_id) - @pytest.mark.smoke def test_create_a_domain(master_domain): # Current domain list - process = exec_test_command( - BASE_CMD + ["list", '--format="id"', "--text", "--no-header"] + domain_list_before = exec_test_command( + BASE_CMDS["domains"] + + ["list", '--format="id"', "--text", "--no-header"] ) - output_current = process.stdout.decode() timestamp = str(time.time_ns()) # Create domain - another_domain = ( - exec_test_command( - [ - "linode-cli", - "domains", - "create", - "--type", - "master", - "--domain", - timestamp + "example.com", - "--soa_email", - "pthiel_test@linode.com", - "--text", - "--no-header", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() + another_domain = exec_test_command( + [ + "linode-cli", + "domains", + "create", + "--type", + "master", + "--domain", + timestamp + "example.com", + "--soa_email", + "pthiel_test@linode.com", + "--text", + "--no-header", + "--format", + "id", + ] ) - process = exec_test_command( - BASE_CMD + ["list", "--format=id", "--text", "--no-header"] + domain_list_after = exec_test_command( + BASE_CMDS["domains"] + ["list", "--format=id", "--text", "--no-header"] ) - output_after = process.stdout.decode() - # Check if list is bigger than previous list - assert len(output_after.splitlines()) > len(output_current.splitlines()) + assert len(domain_list_after.splitlines()) > len( + domain_list_before.splitlines() + ) + assert another_domain in domain_list_after @pytest.mark.smoke -def test_create_domain_srv_record(test_domain_and_record): - domain_id = test_domain_and_record[0] +def test_create_domain_srv_record(domain_and_record): + domain_id = domain_and_record[0] - process = exec_test_command( - BASE_CMD + output = exec_test_command( + BASE_CMDS["domains"] + [ "records-create", "--protocol=tcp", @@ -129,18 +76,16 @@ def test_create_domain_srv_record(test_domain_and_record): ] ) - output = process.stdout.decode() - assert re.search( - r"[0-9]+,SRV,_telnet\._tcp,target-test-record\.\d+example\.com,0,4,4\n", + r"[0-9]+,SRV,_telnet\._tcp,target-test-record\.\w+-example\.com,0,4,4", str(output), ) -def test_list_srv_record(test_domain_and_record): - domain_id = test_domain_and_record[0] - process = exec_test_command( - BASE_CMD +def test_list_srv_record(domain_and_record): + domain_id = domain_and_record[0] + output = exec_test_command( + BASE_CMDS["domains"] + [ "records-list", domain_id, @@ -149,21 +94,20 @@ def test_list_srv_record(test_domain_and_record): "--delimiter=,", ] ) - output = process.stdout.decode() assert re.search( - r"[0-9]+,SRV,_telnet\._tcp,record-setup\.\d+example\.com,0,4,4\n", + r"[0-9]+,SRV,_telnet\._tcp,record-setup\.\w+-example\.com,0,4,4", str(output), ) @pytest.mark.smoke -def test_view_domain_record(test_domain_and_record): - domain_id = test_domain_and_record[0] - record_id = test_domain_and_record[1] +def test_view_domain_record(domain_and_record): + domain_id = domain_and_record[0] + record_id = domain_and_record[1] - process = exec_test_command( - BASE_CMD + output = exec_test_command( + BASE_CMDS["domains"] + [ "records-view", domain_id, @@ -173,20 +117,19 @@ def test_view_domain_record(test_domain_and_record): "--delimiter=,", ] ) - output = process.stdout.decode() assert re.search( - r"[0-9]+,SRV,_telnet\._tcp,record-setup\.\d+example\.com,0,4,4\n", + r"[0-9]+,SRV,_telnet\._tcp,record-setup\.\w+-example\.com,0,4,4", output, ) -def test_update_domain_record(test_domain_and_record): - domain_id = test_domain_and_record[0] - record_id = test_domain_and_record[1] +def test_update_domain_record(domain_and_record): + domain_id = domain_and_record[0] + record_id = domain_and_record[1] - process = exec_test_command( - BASE_CMD + output = exec_test_command( + BASE_CMDS["domains"] + [ "records-update", domain_id, @@ -197,35 +140,30 @@ def test_update_domain_record(test_domain_and_record): "--delimiter=,", ] ) - output = process.stdout.decode() assert re.search( - r"[0-9]+,SRV,_telnet\._tcp,record-setup-update\.\d+example\.com,0,4,4\n", + r"[0-9]+,SRV,_telnet\._tcp,record-setup-update\.\w+-example\.com,0,4,4", str(output), ) -def test_delete_a_domain_record(test_domain_and_record): - domain_id = test_domain_and_record[0] - record_id = test_domain_and_record[1] +def test_delete_a_domain_record(domain_and_record): + domain_id = domain_and_record[0] + record_id = domain_and_record[1] - process = exec_test_command( - BASE_CMD + ["records-delete", domain_id, record_id] + exec_test_command( + BASE_CMDS["domains"] + ["records-delete", domain_id, record_id] ) - # Assert on status code returned from deleting domain - assert process.returncode == SUCCESS_STATUS_CODE - -def test_help_records_list(test_domain_and_record): - process = exec_test_command( - BASE_CMD +def test_help_records_list(domain_and_record): + output = exec_test_command( + BASE_CMDS["domains"] + [ "records-list", "--help", ] ) - output = process.stdout.decode() assert contains_at_least_one_of( output, ["List domain records", "Domain Records List"] diff --git a/tests/integration/domains/test_domain_slave.py b/tests/integration/domains/test_domain_slave.py new file mode 100644 index 000000000..05da7e16e --- /dev/null +++ b/tests/integration/domains/test_domain_slave.py @@ -0,0 +1,62 @@ +import os +import re + +import pytest + +from tests.integration.domains.fixtures import ( # noqa: F401 + slave_domain, +) +from tests.integration.helpers import ( + BASE_CMDS, + exec_test_command, + get_random_text, +) + + +def test_create_slave_domain_fails_without_master_dns_server(): + os.system( + 'linode-cli domains create --type slave --domain "' + + get_random_text(5) + + '-example.com" --text --no-header 2>&1 | tee /tmp/test.txt' + ) + result = exec_test_command(["cat", "/tmp/test.txt"]) + + assert "Request failed: 400" in result + assert ( + "master_ips You need at least one master DNS server IP address for this zone." + in result + ) + + +@pytest.mark.smoke +def test_create_slave_domain(slave_domain): + domain_id = slave_domain + assert re.search("[0-9]+", domain_id) + + +def test_list_slave_domain(slave_domain): + result = exec_test_command( + BASE_CMDS["domains"] + ["list", "--text", "--no-header"] + ) + assert "-example.com" in result + + +def test_update_slave_domain(slave_domain): + domain_id = slave_domain + output = exec_test_command( + BASE_CMDS["domains"] + + [ + "update", + "--type", + "slave", + "--master_ips", + "8.8.8.8", + domain_id, + "--text", + "--no-header", + "--delimiter=,", + ] + ) + + assert "slave,active" in output + assert domain_id in output diff --git a/tests/integration/domains/test_domains_tags.py b/tests/integration/domains/test_domain_tags.py similarity index 56% rename from tests/integration/domains/test_domains_tags.py rename to tests/integration/domains/test_domain_tags.py index 6cdfc8938..5f4474d3e 100644 --- a/tests/integration/domains/test_domains_tags.py +++ b/tests/integration/domains/test_domain_tags.py @@ -1,35 +1,31 @@ import re -import time import pytest from linodecli.exit_codes import ExitCodes from tests.integration.helpers import ( - delete_tag, + BASE_CMDS, delete_target_id, exec_failing_test_command, exec_test_command, + get_random_text, ) -BASE_CMD = ["linode-cli", "domains"] - -# @pytest.mark.skip(reason="BUG 943") def test_fail_to_create_master_domain_with_invalid_tags(): - timestamp = str(time.time_ns()) bad_tag = ( "a" * 300 ) # Tag validation rules changed — '*' is no longer rejected exec_failing_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "create", "--type", "master", "--domain", - timestamp + "example.com", - "--soa_email=" + timestamp + "pthiel@linode.com", + get_random_text(5) + "example.com", + "--soa_email=" + get_random_text(5) + "pthiel@linode.com", "--text", "--no-header", "--format=id", @@ -40,20 +36,18 @@ def test_fail_to_create_master_domain_with_invalid_tags(): ) -# @pytest.mark.skip(reason="BUG 943") def test_fail_to_create_slave_domain_with_invalid_tags(): - timestamp = str(time.time_ns()) bad_tag = "*" exec_failing_test_command( - BASE_CMD + BASE_CMDS["domains"] + [ "create", "--type", "slave", "--domain", - timestamp + "example.com", - "--soa_email=" + timestamp + "pthiel@linode.com", + get_random_text(5) + "example.com", + "--soa_email=" + get_random_text(5) + "pthiel@linode.com", "--text", "--no-header", "--format=id", @@ -66,18 +60,17 @@ def test_fail_to_create_slave_domain_with_invalid_tags(): @pytest.mark.smoke def test_create_master_domain_with_tags(): - timestamp = str(time.time_ns()) tag = "foo" - process = exec_test_command( - BASE_CMD + output = exec_test_command( + BASE_CMDS["domains"] + [ "create", "--type", "master", "--domain", - timestamp + "-example.com", - "--soa_email=" + timestamp + "pthiel@linode.com", + get_random_text(5) + "-example.com", + "--soa_email=" + get_random_text(5) + "pthiel@linode.com", "--text", "--no-header", "--delimiter=,", @@ -86,7 +79,7 @@ def test_create_master_domain_with_tags(): tag, ] ) - output = process.stdout.decode().rstrip() + assert re.search("[0-9]+,[0-9]+-example.com,master,active," + tag, output) res_arr = output.split(",") @@ -94,35 +87,29 @@ def test_create_master_domain_with_tags(): delete_target_id(target="domains", id=domain_id) -# @pytest.mark.skip(reason="BUG 943") def test_delete_domain_and_tag(): - timestamp = str(int(time.time())) tag = "zoo" - domain_id = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--type", - "master", - "--domain", - timestamp + "-example.com", - "--soa_email=" + timestamp + "pthiel@linode.com", - "--text", - "--no-header", - "--delimiter=,", - "--format=id", - "--tag", - tag, - ] - ) - .stdout.decode() - .rstrip() + domain_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "create", + "--type", + "master", + "--domain", + get_random_text(5) + "-example.com", + "--soa_email=" + get_random_text(5) + "pthiel@linode.com", + "--text", + "--no-header", + "--delimiter=,", + "--format=id", + "--tag", + tag, + ] ) # need to check if tag foo is still present while running this test - result = exec_test_command(["linode-cli", "tags", "list"]).stdout.decode() + result = exec_test_command(["linode-cli", "tags", "list"]) if "zoo" in result: - delete_tag("zoo") + delete_target_id(target="tags", id="zoo") delete_target_id(target="domains", id=domain_id) diff --git a/tests/integration/domains/test_slave_domains.py b/tests/integration/domains/test_slave_domains.py deleted file mode 100644 index 15fbeed31..000000000 --- a/tests/integration/domains/test_slave_domains.py +++ /dev/null @@ -1,105 +0,0 @@ -import os -import re -import time - -import pytest - -from tests.integration.helpers import ( - FAILED_STATUS_CODE, - SUCCESS_STATUS_CODE, - delete_target_id, - exec_test_command, -) - -BASE_CMD = ["linode-cli", "domains"] -timestamp = str(time.time_ns()) - - -@pytest.fixture -def slave_domain_setup(): - # Create domain - slave_domain_id = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--type", - "slave", - "--domain", - timestamp + "-example.com", - "--master_ips", - "1.1.1.1", - "--text", - "--no-header", - "--delimiter", - ",", - "--format=id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield slave_domain_id - - delete_target_id(target="domains", id=slave_domain_id) - - -def test_create_slave_domain_fails_without_master_dns_server(): - os.system( - 'linode-cli domains create --type slave --domain "' - + timestamp - + '-example.com" --text --no-header 2>&1 | tee /tmp/test.txt' - ) - result = exec_test_command(["cat", "/tmp/test.txt"]).stdout.decode() - - assert "Request failed: 400" in result - assert ( - "master_ips You need at least one master DNS server IP address for this zone." - in result - ) - - -@pytest.mark.smoke -def test_create_slave_domain(slave_domain): - domain_id = slave_domain - assert re.search("[0-9]+", domain_id) - - -def test_list_slave_domain(slave_domain): - result = exec_test_command( - BASE_CMD + ["list", "--text", "--no-header"] - ).stdout.decode() - assert "-example.com" in result - - -@pytest.mark.skip(reason="BUG 872") -def test_update_domain_fails_without_type(slave_domain_setup): - domain_id = slave_domain_setup - - result = os.system( - "linode-cli domains update " - + domain_id - + ' --master_ips 8.8.8.8 --text --no-header --deleteimiter "," --format "id,domain,type,status"' - ) - - assert result == FAILED_STATUS_CODE - - -def test_update_slave_domain(slave_domain_setup): - domain_id = slave_domain_setup - result = exec_test_command( - BASE_CMD - + [ - "update", - "--type", - "slave", - "--master_ips", - "8.8.8.8", - domain_id, - "--text", - "--no-header", - ] - ) - - assert result.returncode == SUCCESS_STATUS_CODE diff --git a/tests/integration/events/fixtures.py b/tests/integration/events/fixtures.py new file mode 100644 index 000000000..0c9b76478 --- /dev/null +++ b/tests/integration/events/fixtures.py @@ -0,0 +1,32 @@ +import pytest + +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, + get_random_text, +) + + +@pytest.fixture(scope="function") +def events_create_domain(): + # Create domain + domain_id = exec_test_command( + BASE_CMDS["domains"] + + [ + "create", + "--type", + "master", + "--domain", + "A" + get_random_text(5) + "example.com", + "--soa_email=developer-test@linode.com", + "--text", + "--no-header", + "--format", + "id", + ] + ) + + yield domain_id + + delete_target_id(target="domains", id=domain_id) diff --git a/tests/integration/events/test_events.py b/tests/integration/events/test_events.py index 066f272f4..67a1d3de7 100644 --- a/tests/integration/events/test_events.py +++ b/tests/integration/events/test_events.py @@ -1,46 +1,15 @@ import re -import time import pytest -from tests.integration.helpers import delete_target_id, exec_test_command - -BASE_CMD = ["linode-cli", "events"] - - -@pytest.fixture -def events_test_domain_id(): - timestamp = str(time.time_ns()) - # Create domain - domain_id = ( - exec_test_command( - [ - "linode-cli", - "domains", - "create", - "--type", - "master", - "--domain", - "A" + timestamp + "example.com", - "--soa_email=developer-test@linode.com", - "--text", - "--no-header", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield domain_id - - delete_target_id(target="domains", id=domain_id) +from tests.integration.events.fixtures import ( # noqa: F401 + events_create_domain, +) +from tests.integration.helpers import BASE_CMDS, exec_test_command def test_print_events_usage_information(): - process = exec_test_command(BASE_CMD) - output = process.stdout.decode() + output = exec_test_command(BASE_CMDS["events"]) assert "linode-cli events [ACTION]" in output @@ -56,124 +25,102 @@ def test_print_events_usage_information(): @pytest.mark.smoke def test_list_events(): - process = exec_test_command( - BASE_CMD + ["list", "--text", "--no-headers", "--delimiter", ","] + output = exec_test_command( + BASE_CMDS["events"] + + ["list", "--text", "--no-headers", "--delimiter", ","] ) - output = process.stdout.decode() assert re.search("[0-9]+,.*,.*,[0-9]+-[0-9][0-9]-.*,.*,[a-z]+.*", output) def test_view_events(): - event_id = ( - exec_test_command( - [ - "linode-cli", - "events", - "list", - "--format", - "id", - "--no-headers", - "--text", - ] - ) - .stdout.decode() - .rstrip() - .split()[0] - ) + event_id = exec_test_command( + BASE_CMDS["events"] + + [ + "list", + "--format", + "id", + "--no-headers", + "--text", + ] + ).split()[0] result = exec_test_command( - BASE_CMD + BASE_CMDS["events"] + ["view", event_id, "--text", "--no-headers", "--delimiter", ","] - ).stdout.decode() + ) assert re.search("[0-9]+,.*,.*,[0-9]+-[0-9][0-9]-.*,.*,[a-z]+.*", result) def test_mark_event_seen(): - event_id = ( - exec_test_command( - [ - "linode-cli", - "events", - "list", - "--format", - "id", - "--no-headers", - "--text", - ] - ) - .stdout.decode() - .rstrip() - .split()[0] - ) + event_id = exec_test_command( + BASE_CMDS["events"] + + [ + "list", + "--format", + "id", + "--no-headers", + "--text", + ] + ).split()[0] # mark event as seen exec_test_command( - BASE_CMD + BASE_CMDS["events"] + ["mark-seen", event_id, "--text", "--no-headers", "--delimiter", ","] - ).stdout.decode() + ) # view event result = exec_test_command( - BASE_CMD + BASE_CMDS["events"] + ["view", event_id, "--text", "--no-headers", "--delimiter", ","] - ).stdout.decode() + ) assert re.search("[0-9]+,.*,.*,[0-9]+-[0-9][0-9]-.*,.*,[a-z]+.*", result) @pytest.mark.smoke def test_mark_event_read(): - event_id = ( - exec_test_command( - [ - "linode-cli", - "events", - "list", - "--format", - "id", - "--no-headers", - "--text", - ] - ) - .stdout.decode() - .rstrip() - .split()[0] - ) + event_id = exec_test_command( + [ + "linode-cli", + "events", + "list", + "--format", + "id", + "--no-headers", + "--text", + ] + ).split()[0] # mark event as read exec_test_command( - BASE_CMD + BASE_CMDS["events"] + ["mark-read", event_id, "--text", "--no-headers", "--delimiter", ","] - ).stdout.decode() + ) # view event result = exec_test_command( - BASE_CMD + BASE_CMDS["events"] + ["view", event_id, "--text", "--no-headers", "--delimiter", ","] - ).stdout.decode() + ) assert re.search("[0-9]+,.*,.*,[0-9]+-[0-9][0-9]-.*,.*,[a-z]+.*", result) def test_filter_events_by_entity_id(): - event_id = ( - exec_test_command( - [ - "linode-cli", - "events", - "list", - "--format", - "id", - "--no-headers", - "--text", - ] - ) - .stdout.decode() - .rstrip() - .split()[0] - ) + event_id = exec_test_command( + [ + "linode-cli", + "events", + "list", + "--format", + "id", + "--no-headers", + "--text", + ] + ).split()[0] result = exec_test_command( - BASE_CMD + BASE_CMDS["events"] + [ "list", "--id", @@ -183,17 +130,16 @@ def test_filter_events_by_entity_id(): "--delimiter", ",", ] - ).stdout.decode() + ) assert re.search( event_id + ",.*,.*,[0-9]+-[0-9][0-9]-.*,.*,[a-z]+.*", result ) -@pytest.mark.skip(reason="https://github.com/linode/linode-cli/issues/500") -def test_create_domain_and_filter_domain_events(events_test_domain_id): - domain_id = events_test_domain_id - result = exec_test_command( - BASE_CMD +def test_create_domain_and_filter_domain_events(events_create_domain): + domain_id = events_create_domain + output = exec_test_command( + BASE_CMDS["events"] + [ "list", "--entity.id", @@ -205,8 +151,6 @@ def test_create_domain_and_filter_domain_events(events_test_domain_id): "--delimiter", ",", ] - ).stdout.decode() - - assert re.search( - "[0-9]+,.*,domain_create,A[0-9]+.*,[0-9]+-[0-9][0-9]-.*,.*", result ) + + assert "domain_create" in output diff --git a/tests/integration/firewalls/fixtures.py b/tests/integration/firewalls/fixtures.py new file mode 100644 index 000000000..72559555f --- /dev/null +++ b/tests/integration/firewalls/fixtures.py @@ -0,0 +1,34 @@ +import pytest + +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, + get_random_text, +) + +FIREWALL_LABEL = "label-fw-test-" + get_random_text(5) + + +@pytest.fixture(scope="session") +def firewall_id(): + firewall_id = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + FIREWALL_LABEL, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--text", + "--no-headers", + "--format", + "id", + ] + ) + + yield firewall_id + + delete_target_id(target="firewalls", id=firewall_id) diff --git a/tests/integration/firewalls/test_firewalls.py b/tests/integration/firewalls/test_firewalls.py index 495b43bd6..139979578 100644 --- a/tests/integration/firewalls/test_firewalls.py +++ b/tests/integration/firewalls/test_firewalls.py @@ -1,79 +1,46 @@ import re -import time import pytest from linodecli.exit_codes import ExitCodes +from tests.integration.firewalls.fixtures import ( # noqa: F401 + FIREWALL_LABEL, + firewall_id, +) from tests.integration.helpers import ( + BASE_CMDS, delete_target_id, exec_failing_test_command, exec_test_command, + get_random_text, ) -BASE_CMD = ["linode-cli", "firewalls"] -FIREWALL_LABEL = "label-fw-test" + str(int(time.time())) - - -@pytest.fixture -def test_firewall_id(): - firewall_id = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--label", - FIREWALL_LABEL, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield firewall_id - - delete_target_id(target="firewalls", id=firewall_id) - @pytest.mark.smoke -def test_view_firewall(test_firewall_id): - firewall_id = test_firewall_id - - result = ( - exec_test_command( - BASE_CMD - + [ - "view", - firewall_id, - "--no-headers", - "--text", - "--delimiter", - ",", - ] - ) - .stdout.decode() - .rstrip() +def test_view_firewall(firewall_id): + firewall_id = firewall_id + + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "view", + firewall_id, + "--no-headers", + "--text", + "--delimiter", + ",", + ] ) assert re.search(firewall_id + "," + FIREWALL_LABEL + ",enabled", result) -def test_list_firewall(test_firewall_id): - firewall_id = test_firewall_id +def test_list_firewall(firewall_id): + firewall_id = firewall_id - result = ( - exec_test_command( - BASE_CMD + ["list", "--no-headers", "--text", "--delimiter", ","] - ) - .stdout.decode() - .rstrip() + result = exec_test_command( + BASE_CMDS["firewalls"] + + ["list", "--no-headers", "--text", "--delimiter", ","] ) assert re.search(firewall_id + "," + FIREWALL_LABEL + ",enabled", result) @@ -81,27 +48,22 @@ def test_list_firewall(test_firewall_id): @pytest.mark.smoke def test_create_firewall_with_minimum_required_args(): - timestamp = str(time.time_ns()) - firewall_label = "label-fw-test" + timestamp - result = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--label", - firewall_label, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--text", - "--no-headers", - "--delimiter", - ",", - ] - ) - .stdout.decode() - .rstrip() + firewall_label = "label-fw-test" + get_random_text(5) + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--text", + "--no-headers", + "--delimiter", + ",", + ] ) assert re.search("[0-9]+," + firewall_label + ",enabled", result) @@ -112,109 +74,90 @@ def test_create_firewall_with_minimum_required_args(): def test_fails_to_create_firewall_without_inbound_policy(): - timestamp = str(time.time_ns()) - firewall_label = "fw_label" + timestamp - result = ( - exec_failing_test_command( - BASE_CMD - + [ - "create", - "--label", - firewall_label, - "--rules.outbound_policy", - "ACCEPT", - "--text", - "--no-headers", - "--format", - "id", - ], - ExitCodes.REQUEST_FAILED, - ) - .stderr.decode() - .rstrip() + firewall_label = "fw_label" + get_random_text(5) + result = exec_failing_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.outbound_policy", + "ACCEPT", + "--text", + "--no-headers", + "--format", + "id", + ], + ExitCodes.REQUEST_FAILED, ) assert "inbound_policy is required" in result def test_fails_to_create_firewall_without_outbound_policy(): - timestamp = str(time.time_ns()) - firewall_label = "fw_label" + timestamp - result = ( - exec_failing_test_command( - BASE_CMD - + [ - "create", - "--label", - firewall_label, - "--rules.inbound_policy", - "DROP", - "--text", - "--no-headers", - "--format", - "id", - ], - ExitCodes.REQUEST_FAILED, - ) - .stderr.decode() - .rstrip() + firewall_label = "fw_label" + get_random_text(5) + result = exec_failing_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.inbound_policy", + "DROP", + "--text", + "--no-headers", + "--format", + "id", + ], + ExitCodes.REQUEST_FAILED, ) assert "outbound_policy is required" in result -def test_firewall_label_must_be_unique_upon_creation(test_firewall_id): - result = ( - exec_failing_test_command( - BASE_CMD - + [ - "create", - "--label", - FIREWALL_LABEL, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--text", - "--no-headers", - "--format", - "id", - ], - ExitCodes.REQUEST_FAILED, - ) - .stderr.decode() - .rstrip() +def test_firewall_label_must_be_unique_upon_creation(firewall_id): + result = exec_failing_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + FIREWALL_LABEL, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--text", + "--no-headers", + "--format", + "id", + ], + ExitCodes.REQUEST_FAILED, ) assert "Label must be unique among your Cloud Firewalls" in result def test_create_firewall_with_inbound_and_outbound_args(): - timestamp = str(time.time_ns()) - firewall_label = "label-fw-test" + timestamp - result = ( - exec_test_command( - BASE_CMD - + [ - "create", - "--label", - firewall_label, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--rules.inbound", - '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "accept-inbound-SSH"}]', - "--rules.outbound", - '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "accept-outbound-SSH"}]', - "--text", - "--no-headers", - "--delimiter", - ",", - ] - ) - .stdout.decode() - .rstrip() + firewall_label = "label-fw-test" + get_random_text(5) + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--rules.inbound", + '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "accept-inbound-SSH"}]', + "--rules.outbound", + '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "accept-outbound-SSH"}]', + "--text", + "--no-headers", + "--delimiter", + ",", + ] ) assert re.search("[0-9]+," + firewall_label + ",enabled", result) @@ -224,32 +167,27 @@ def test_create_firewall_with_inbound_and_outbound_args(): delete_target_id(target="firewalls", id=firewall_id) -def test_update_firewall(test_firewall_id): - timestamp = str(time.time_ns()) - firewall_id = test_firewall_id - updated_tag = "updated-tag" + timestamp - updated_label = "updated-" + timestamp - - result = ( - exec_test_command( - BASE_CMD - + [ - "update", - firewall_id, - "--tags", - updated_tag, - "--label", - updated_label, - "--status", - "enabled", - "--no-headers", - "--text", - "--delimiter", - ",", - ] - ) - .stdout.decode() - .rstrip() +def test_update_firewall(firewall_id): + firewall_id = firewall_id + updated_tag = "updated-tag" + get_random_text(5) + updated_label = "updated-" + get_random_text(5) + + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "update", + firewall_id, + "--tags", + updated_tag, + "--label", + updated_label, + "--status", + "enabled", + "--no-headers", + "--text", + "--delimiter", + ",", + ] ) assert re.search(firewall_id + "," + updated_label + ",enabled", result) diff --git a/tests/integration/firewalls/test_firewalls_rules.py b/tests/integration/firewalls/test_firewalls_rules.py index f3bae428f..8767d9143 100644 --- a/tests/integration/firewalls/test_firewalls_rules.py +++ b/tests/integration/firewalls/test_firewalls_rules.py @@ -1,52 +1,26 @@ import json import re -import time -import pytest +from tests.integration.firewalls.fixtures import ( # noqa: F401 + firewall_id, +) +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, + get_random_text, +) -from tests.integration.helpers import delete_target_id, exec_test_command -BASE_CMD = ["linode-cli", "firewalls", "rules-update"] -FIREWALL_LABEL = "label-fw-test" + str(int(time.time())) - - -@pytest.fixture -def test_firewall_id(): - firewall_id = ( - exec_test_command( - [ - "linode-cli", - "firewalls", - "create", - "--label", - FIREWALL_LABEL, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - ) - - yield firewall_id - # teardown - delete all firewalls - delete_target_id(target="firewalls", id=firewall_id) - - -def test_add_rule_to_existing_firewall(test_firewall_id): - firewall_id = test_firewall_id +def test_add_rule_to_existing_firewall(firewall_id): + firewall_id = firewall_id inbound_rule = '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "accept-inbound-SSH"}]' outbound_rule = '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "accept-outbound-SSH"}]' result = json.loads( exec_test_command( - BASE_CMD + BASE_CMDS["firewalls"] + [ + "rules-update", firewall_id, "--inbound", inbound_rule, @@ -55,25 +29,28 @@ def test_add_rule_to_existing_firewall(test_firewall_id): "--json", ] ) - .stdout.decode() - .rstrip() ) assert result[0]["inbound"][0] == json.loads(inbound_rule)[0] assert result[0]["outbound"][0] == json.loads(outbound_rule)[0] -def test_add_multiple_rules(test_firewall_id): - firewall_id = test_firewall_id +def test_add_multiple_rules(firewall_id): + firewall_id = firewall_id inbound_rule_1 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "accept-inbound-SSH"}' inbound_rule_2 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "accept-inbound-SSH-2"}' inbound_rules = "[" + inbound_rule_1 + "," + inbound_rule_2 + "]" result = json.loads( exec_test_command( - BASE_CMD + [firewall_id, "--inbound", inbound_rules, "--json"] + BASE_CMDS["firewalls"] + + [ + "rules-update", + firewall_id, + "--inbound", + inbound_rules, + "--json", + ] ) - .stdout.decode() - .rstrip() ) assert result[0]["inbound"][0] == json.loads(inbound_rule_1) @@ -81,33 +58,27 @@ def test_add_multiple_rules(test_firewall_id): def test_swap_rules(): - timestamp = str(time.time_ns()) - firewall_label = "label-fw-test" + timestamp + firewall_label = "label-fw-test" + get_random_text(5) inbound_rule_1 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "swap_rule_1"}' inbound_rule_2 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "swap_rule_2"}' inbound_rules = "[" + inbound_rule_1 + "," + inbound_rule_2 + "]" - firewall_id = ( - exec_test_command( - [ - "linode-cli", - "firewalls", - "create", - "--label", - firewall_label, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--rules.inbound", - inbound_rules, - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() + firewall_id = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--rules.inbound", + inbound_rules, + "--text", + "--no-headers", + "--format", + "id", + ] ) swapped_rules = "[" + inbound_rule_2 + "," + inbound_rule_1 + "]" @@ -115,16 +86,15 @@ def test_swap_rules(): # swapping rules result = json.loads( exec_test_command( - BASE_CMD + BASE_CMDS["firewalls"] + [ + "rules-update", firewall_id, "--inbound", swapped_rules, "--json", ] ) - .stdout.decode() - .rstrip() ) assert result[0]["inbound"][0] == json.loads(inbound_rule_2) @@ -133,71 +103,67 @@ def test_swap_rules(): delete_target_id(target="firewalls", id=firewall_id) -def test_update_inbound_and_outbound_policy(test_firewall_id): - firewall_id = test_firewall_id +def test_update_inbound_and_outbound_policy(firewall_id): + firewall_id = firewall_id outbound_policy = "DROP" inbound_policy = "ACCEPT" - result = ( - exec_test_command( - BASE_CMD - + [ - firewall_id, - "--inbound_policy", - inbound_policy, - "--outbound_policy", - outbound_policy, - "--text", - "--no-headers", - "--delimiter", - ",", - ] - ) - .stdout.decode() - .rstrip() + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "rules-update", + firewall_id, + "--inbound_policy", + inbound_policy, + "--outbound_policy", + outbound_policy, + "--text", + "--no-headers", + "--delimiter", + ",", + ] ) assert re.search(inbound_policy + "," + outbound_policy, result) def test_remove_one_rule_via_rules_update(): - timestamp = str(time.time_ns()) - firewall_label = "label-fw-test" + timestamp + firewall_label = "label-fw-test" + get_random_text(5) inbound_rule_1 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": "test_rule_1"}' inbound_rule_2 = '{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.2/32"]}, "action": "ACCEPT", "label": "rule_to_delete"}' inbound_rules = "[" + inbound_rule_1 + "," + inbound_rule_2 + "]" - firewall_id = ( - exec_test_command( - [ - "linode-cli", - "firewalls", - "create", - "--label", - firewall_label, - "--rules.outbound_policy", - "ACCEPT", - "--rules.inbound_policy", - "DROP", - "--rules.inbound", - inbound_rules, - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() + firewall_id = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "create", + "--label", + firewall_label, + "--rules.outbound_policy", + "ACCEPT", + "--rules.inbound_policy", + "DROP", + "--rules.inbound", + inbound_rules, + "--text", + "--no-headers", + "--format", + "id", + ] ) new_rule = "[" + inbound_rule_1 + "]" # swapping rules result = json.loads( exec_test_command( - BASE_CMD - + [firewall_id, "--inbound", new_rule, "--json", "--no-headers"] + BASE_CMDS["firewalls"] + + [ + "rules-update", + firewall_id, + "--inbound", + new_rule, + "--json", + "--no-headers", + ] ) - .stdout.decode() - .rstrip() ) rule_labels = [v["label"] for v in result[0]["inbound"]] @@ -207,8 +173,8 @@ def test_remove_one_rule_via_rules_update(): delete_target_id(target="firewalls", id=firewall_id) -def test_list_rules(test_firewall_id): - firewall_id = test_firewall_id +def test_list_rules(firewall_id): + firewall_id = firewall_id new_label = '"rules-list-test"' inbound_rule = ( '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": ' @@ -217,8 +183,9 @@ def test_list_rules(test_firewall_id): ) # adding a rule exec_test_command( - BASE_CMD + BASE_CMDS["firewalls"] + [ + "rules-update", firewall_id, "--inbound", inbound_rule, @@ -227,27 +194,22 @@ def test_list_rules(test_firewall_id): "--delimiter", ",", ] - ).stdout.decode().rstrip() - result = ( - exec_test_command( - [ - "linode-cli", - "firewalls", - "rules-list", - firewall_id, - "--text", - "--no-headers", - ] - ) - .stdout.decode() - .rstrip() + ) + result = exec_test_command( + BASE_CMDS["firewalls"] + + [ + "rules-list", + firewall_id, + "--text", + "--no-headers", + ] ) assert new_label.replace('"', "") in result -def test_list_rules_json(test_firewall_id): - firewall_id = test_firewall_id +def test_list_rules_json(firewall_id): + firewall_id = firewall_id new_label = '"rules-list-test"' inbound_rule = ( '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": ' @@ -256,8 +218,9 @@ def test_list_rules_json(test_firewall_id): ) # adding a rule exec_test_command( - BASE_CMD + BASE_CMDS["firewalls"] + [ + "rules-update", firewall_id, "--inbound", inbound_rule, @@ -266,19 +229,16 @@ def test_list_rules_json(test_firewall_id): "--delimiter", ",", ] - ).stdout.decode().rstrip() + ) result = json.loads( exec_test_command( - [ - "linode-cli", - "firewalls", + BASE_CMDS["firewalls"] + + [ "rules-list", firewall_id, "--json", ] ) - .stdout.decode() - .rstrip() ) assert result[0]["inbound"][0]["action"] == "ACCEPT" @@ -286,8 +246,8 @@ def test_list_rules_json(test_firewall_id): assert result[0]["inbound"][0]["addresses"]["ipv4"] == ["198.0.0.1/32"] -def test_list_rules_json_format(test_firewall_id): - firewall_id = test_firewall_id +def test_list_rules_json_format(firewall_id): + firewall_id = firewall_id new_label = '"rules-list-test"' inbound_rule = ( '[{"ports": "22", "protocol": "TCP", "addresses": {"ipv4": ["198.0.0.1/32"]}, "action": "ACCEPT", "label": ' @@ -296,8 +256,9 @@ def test_list_rules_json_format(test_firewall_id): ) # adding a rule exec_test_command( - BASE_CMD + BASE_CMDS["firewalls"] + [ + "rules-update", firewall_id, "--inbound", inbound_rule, @@ -306,12 +267,11 @@ def test_list_rules_json_format(test_firewall_id): "--delimiter", ",", ] - ).stdout.decode().rstrip() + ) result = json.loads( exec_test_command( - [ - "linode-cli", - "firewalls", + BASE_CMDS["firewalls"] + + [ "rules-list", firewall_id, "--json", @@ -319,7 +279,5 @@ def test_list_rules_json_format(test_firewall_id): "label", ] ) - .stdout.decode() - .rstrip() ) assert result[0]["inbound"][0]["label"] == "rules-list-test" diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py index 41f0e5bec..b3b9c2e09 100644 --- a/tests/integration/helpers.py +++ b/tests/integration/helpers.py @@ -5,7 +5,6 @@ from string import ascii_lowercase from typing import Callable, Container, Iterable, List, TypeVar -from linodecli import ExitCodes from linodecli.exit_codes import ExitCodes BASE_URL = "https://api.linode.com/v4/" @@ -17,6 +16,27 @@ # TypeVars for generic type hints below T = TypeVar("T") +BASE_CMD = "linode-cli" +MODULES = [ + "account", + "domains", + "linodes", + "nodebalancers", + "betas", + "databases", + "domains", + "events", + "image", + "image-upload", + "firewalls", + "kernels", + "linodes", + "lke", + "longview", + "managed", +] +BASE_CMDS = {module: [BASE_CMD, module] for module in MODULES} + def get_random_text(length: int = 10): return "".join(random.choice(ascii_lowercase) for i in range(length)) @@ -36,108 +56,51 @@ def wait_for_condition(interval: int, timeout: int, condition: Callable): def exec_test_command(args: List[str]): - process = subprocess.run(args, stdout=subprocess.PIPE) - assert process.returncode == 0 - return process + process = subprocess.run( + args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) + if process.returncode != 0: + raise RuntimeError( + f"Command failed with exit code {process.returncode}\n" + f"Command: {' '.join(args)}\n" + f"Stdout:\n{process.stdout}\n" + f"Stderr:\n{process.stderr}" + ) + return process.stdout.rstrip() def exec_failing_test_command( args: List[str], expected_code: int = ExitCodes.REQUEST_FAILED ): - process = subprocess.run(args, stderr=subprocess.PIPE) - assert process.returncode == expected_code - return process - - -# Delete/Remove helper functions (mainly used in clean-ups after test -def delete_all_domains(): - domain_ids = exec_test_command( - [ - "linode-cli", - "--text", - "--no-headers", - "domains", - "list", - "--format=id", - ] - ).stdout.decode() - domain_id_arr = domain_ids.splitlines() - - for id in domain_id_arr: - exec_test_command(["linode-cli", "domains", "delete", id]) - + process = subprocess.run(args, stderr=subprocess.PIPE, text=True) + + if process.returncode != expected_code: + raise AssertionError( + f"Expected exit code {expected_code}, got {process.returncode}\n" + f"Command: {' '.join(args)}\n" + f"Stdout:\n{process.stdout}\n" + f"Stderr: {process.stderr}" + ) -def delete_tag(arg: str): - result = exec_test_command(["linode-cli", "tags", "delete", arg]) - assert result.returncode == SUCCESS_STATUS_CODE + return process.stderr.rstrip() +# Delete/Remove helper functions (mainly used in clean-ups after test def delete_target_id(target: str, id: str, delete_command: str = "delete"): command = ["linode-cli", target, delete_command, id] - result = exec_test_command(command) - assert result.returncode == SUCCESS_STATUS_CODE - - -def remove_lke_clusters(): - cluster_ids = ( - exec_test_command( - [ - "linode-cli", - "--text", - "--no-headers", - "lke", - "clusters-list", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - for id in cluster_ids: - exec_test_command(["linode-cli", "lke", "cluster-delete", id]) - - -def remove_all(target: str): - entity_ids = "" - if target == "stackscripts": - entity_ids = ( - exec_test_command( - [ - "linode-cli", - "--is_public=false", - "--text", - "--no-headers", - target, - "list", - "--format", - "id", - ] - ) - .stdout.decode() - .splitlines() + try: + subprocess.run( + command, + check=True, # Raises CalledProcessError on non-zero exit + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, ) - else: - entity_ids = ( - exec_test_command( - [ - "linode-cli", - "--text", - "--no-headers", - target, - "list", - "--format", - "id", - ] - ) - .stdout.decode() - .splitlines() + except Exception as e: + raise RuntimeError( + f"Error executing command '{' '.join(command)}': {e}" ) - for id in entity_ids: - exec_test_command(["linode-cli", target, "delete", id]) - def count_lines(text: str): return len(list(filter(len, text.split("\n")))) @@ -174,11 +137,9 @@ def retry_exec_test_command_with_delay( def get_random_region_with_caps( required_capabilities: List[str], site_type="core" ): - json_regions_data = ( - exec_test_command(["linode-cli", "regions", "ls", "--json"]) - .stdout.decode() - .strip() - ) + json_regions_data = exec_test_command( + ["linode-cli", "regions", "ls", "--json"] + ).stdout.strip() # Parse regions JSON data regions = json.loads(json_regions_data) @@ -197,21 +158,17 @@ def get_random_region_with_caps( def get_cluster_id(label: str): - cluster_id = ( - exec_test_command( - [ - "linode-cli", - "lke", - "clusters-list", - "--text", - "--format=id", - "--no-headers", - "--label", - label, - ] - ) - .stdout.decode() - .rstrip() + cluster_id = exec_test_command( + [ + "linode-cli", + "lke", + "clusters-list", + "--text", + "--format=id", + "--no-headers", + "--label", + label, + ] ) return cluster_id diff --git a/tests/integration/image/fixtures.py b/tests/integration/image/fixtures.py new file mode 100644 index 000000000..13960f74b --- /dev/null +++ b/tests/integration/image/fixtures.py @@ -0,0 +1,21 @@ +import os +import tempfile + +import pytest + +# A minimal gzipped image that will be accepted by the API +TEST_IMAGE_CONTENT = ( + b"\x1f\x8b\x08\x08\xbd\x5c\x91\x60\x00\x03\x74\x65\x73\x74\x2e\x69" + b"\x6d\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00" +) + + +@pytest.fixture(scope="session", autouse=True) +def fake_image_file(): + with tempfile.NamedTemporaryFile(delete=False) as fp: + fp.write(TEST_IMAGE_CONTENT) + file_path = fp.name + + yield file_path + + os.remove(file_path) diff --git a/tests/integration/image/test_plugin_image_upload.py b/tests/integration/image/test_plugin_image_upload.py index 5a05b0536..2c5aaa01f 100644 --- a/tests/integration/image/test_plugin_image_upload.py +++ b/tests/integration/image/test_plugin_image_upload.py @@ -1,53 +1,24 @@ import json -import os import re -import subprocess -import tempfile from sys import platform -from typing import List import pytest from tests.integration.helpers import ( + BASE_CMDS, assert_headers_in_lines, exec_failing_test_command, + exec_test_command, get_random_text, ) +from tests.integration.image.fixtures import fake_image_file # noqa: F401 -REGION = "us-iad" -BASE_CMD = ["linode-cli", "image-upload", "--region", REGION] - -# A minimal gzipped image that will be accepted by the API -TEST_IMAGE_CONTENT = ( - b"\x1f\x8b\x08\x08\xbd\x5c\x91\x60\x00\x03\x74\x65\x73\x74\x2e\x69" - b"\x6d\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00" -) - - -@pytest.fixture(scope="session", autouse=True) -def fake_image_file(): - with tempfile.NamedTemporaryFile(delete=False) as fp: - fp.write(TEST_IMAGE_CONTENT) - file_path = fp.name - - yield file_path - - os.remove(file_path) - - -def exec_test_command(args: List[str]): - process = subprocess.run( - args, - stdout=subprocess.PIPE, - ) - return process +BASE_CMD = BASE_CMDS["image-upload"] + ["--region", "us-iad"] def test_help(): - process = exec_test_command(BASE_CMD + ["--help"]) - output = process.stdout.decode() + output = exec_test_command(BASE_CMD + ["--help"]) - assert process.returncode == 0 assert "The image file to upload" in output assert "The region to upload the image to" in output @@ -56,13 +27,11 @@ def test_invalid_file( fake_image_file, ): file_path = fake_image_file + "_fake" - process = exec_failing_test_command( + output = exec_failing_test_command( BASE_CMD + ["--label", "notimportant", file_path], expected_code=8 ) - error_output = process.stderr.decode() - assert process.returncode == 8 - assert f"No file at {file_path}" in error_output + assert f"No file at {file_path}" in output @pytest.mark.smoke @@ -75,31 +44,24 @@ def test_file_upload( description = "test description" # Upload the test image - process = exec_test_command( + output = exec_test_command( BASE_CMD + ["--label", label, "--description", description, file_path] ) - output = process.stdout.decode() - - assert process.returncode == 0 - # Assertions now using keywords due to some chars getting cut off from lack of terminal space assert re.search("[0-9][0-9]+.[0-9]%", output) assert re.search("test", output) # Get the new image from the API - process = exec_test_command( + output = exec_test_command( ["linode-cli", "images", "ls", "--json", "--label", label] ) - assert process.returncode == 0 - - image = json.loads(process.stdout.decode()) + image = json.loads(output) assert image[0]["label"] == label # Delete the image - process = exec_test_command(["linode-cli", "images", "rm", image[0]["id"]]) - assert process.returncode == 0 + exec_test_command(["linode-cli", "images", "rm", image[0]["id"]]) @pytest.mark.skipif(platform == "win32", reason="Test N/A on Windows") @@ -111,7 +73,7 @@ def test_file_upload_cloud_init( description = "test description" # Upload the test image - process = exec_test_command( + exec_test_command( BASE_CMD + [ "--label", @@ -123,31 +85,23 @@ def test_file_upload_cloud_init( ] ) - assert process.returncode == 0 - # Get the new image from the API - process = exec_test_command( + output = exec_test_command( ["linode-cli", "images", "ls", "--json", "--label", label] ) - assert process.returncode == 0 - image = json.loads(process.stdout.decode()) + image = json.loads(output) assert image[0]["label"] == label assert "cloud-init" in image[0]["capabilities"] # Delete the image - process = exec_test_command(["linode-cli", "images", "rm", image[0]["id"]]) - assert process.returncode == 0 + exec_test_command(["linode-cli", "images", "rm", image[0]["id"]]) def test_image_list(): - res = ( - exec_test_command( - ["linode-cli", "images", "list", "--text", "--delimiter=,"] - ) - .stdout.decode() - .rstrip() + res = exec_test_command( + ["linode-cli", "images", "list", "--text", "--delimiter=,"] ) lines = res.splitlines() @@ -155,50 +109,36 @@ def test_image_list(): assert_headers_in_lines(headers, lines) -@pytest.fixture -def get_image_id(): - image_id = ( - exec_test_command( - [ - "linode-cli", - "images", - "list", - "--text", - "--no-headers", - "--delimiter", - ",", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) - first_id = image_id[0].split(",")[0] - yield first_id - - -def test_image_view(get_image_id): - image_id = get_image_id - res = ( - exec_test_command( - [ - "linode-cli", - "images", - "view", - image_id, - "--text", - "--delimiter=,", - ] - ) - .stdout.decode() - .rstrip() +def test_image_view(): + image_ids = exec_test_command( + [ + "linode-cli", + "images", + "list", + "--text", + "--no-headers", + "--delimiter", + ",", + "--format", + "id", + ] + ).splitlines() + image_id = image_ids[0].split(",")[0] + + res = exec_test_command( + [ + "linode-cli", + "images", + "view", + image_id, + "--text", + "--delimiter=,", + ] ) lines = res.splitlines() headers = ["label", "description"] assert_headers_in_lines(headers, lines) - # assert that regions in the output assert "regions" in lines + assert image_id in str(lines) diff --git a/tests/integration/kernels/test_kernels.py b/tests/integration/kernels/test_kernels.py index 9b6ec0326..214c30bb2 100644 --- a/tests/integration/kernels/test_kernels.py +++ b/tests/integration/kernels/test_kernels.py @@ -2,30 +2,32 @@ import pytest -from tests.integration.helpers import exec_test_command - -BASE_CMD = ["linode-cli", "kernels", "list", "--text", "--no-headers"] +from tests.integration.helpers import BASE_CMDS, exec_test_command def test_list_available_kernels(): - process = exec_test_command(BASE_CMD + ["--format", "id"]) - output = process.stdout.decode() + output = exec_test_command( + BASE_CMDS["kernels"] + + ["list", "--text", "--no-headers", "--format", "id"] + ) for line in output.splitlines(): assert "linode" in line def test_fields_from_kernels_list(): - process = exec_test_command( - BASE_CMD + output = exec_test_command( + BASE_CMDS["kernels"] + [ + "list", + "--text", + "--no-headers", "--delimiter", ",", "--format", "id,version,kvm,architecture,pvops,deprecated,built", ] ) - output = process.stdout.decode() for line in output.splitlines(): assert re.search( @@ -36,15 +38,16 @@ def test_fields_from_kernels_list(): @pytest.mark.smoke def test_view_kernel(): - process = exec_test_command(BASE_CMD + ["--format", "id"]) - output = process.stdout.decode() + output = exec_test_command( + BASE_CMDS["kernels"] + + ["list", "--text", "--no-headers", "--format", "id"] + ) lines = output.splitlines() - process = exec_test_command( - [ - "linode-cli", - "kernels", + output = exec_test_command( + BASE_CMDS["kernels"] + + [ "view", str(lines[0]), "--format", @@ -54,7 +57,6 @@ def test_view_kernel(): ",", ] ) - output = process.stdout.decode() assert "id,version,kvm,architecture,pvops,deprecated,built" in output diff --git a/tests/integration/linodes/fixtures.py b/tests/integration/linodes/fixtures.py new file mode 100644 index 000000000..62054e44f --- /dev/null +++ b/tests/integration/linodes/fixtures.py @@ -0,0 +1,156 @@ + +import pytest + +from tests.integration.helpers import ( + delete_target_id, + exec_test_command, + get_random_region_with_caps, + get_random_text, + retry_exec_test_command_with_delay, +) +from tests.integration.linodes.helpers_linodes import ( + BASE_CMD, + create_linode, + create_linode_backup_disabled, + get_disk_ids, + set_backups_enabled_in_account_settings, +) + + +# Backups +@pytest.fixture +def create_linode_setup(linode_cloud_firewall): + linode_id = create_linode(firewall_id=linode_cloud_firewall) + + yield linode_id + + delete_target_id("linodes", linode_id) + + +@pytest.fixture +def create_linode_backup_disabled_setup(linode_cloud_firewall): + res = set_backups_enabled_in_account_settings(toggle=False) + + if res == "True": + raise ValueError( + "Backups are unexpectedly enabled before setting up the test." + ) + + linode_id = create_linode_backup_disabled(firewall_id=linode_cloud_firewall) + + yield linode_id + + delete_target_id("linodes", linode_id) + + +# Configs + +TEST_REGION = get_random_region_with_caps( + required_capabilities=["Linodes", "VPCs"] +) + + +@pytest.fixture(scope="session", autouse=True) +def linode_instance_config_tests(linode_cloud_firewall): + linode_id = create_linode( + firewall_id=linode_cloud_firewall, + disk_encryption=False, + test_region=TEST_REGION, + ) + + yield linode_id + + delete_target_id(target="linodes", id=linode_id) + + +@pytest.fixture(scope="session", autouse=True) +def linode_disk_config(linode_instance_config_tests): + linode_id = linode_instance_config_tests + + label = get_random_text(5) + "_config" + disk_id = get_disk_ids(linode_id=linode_id)[1] + + config_id = ( + exec_test_command( + BASE_CMD + + [ + "config-create", + linode_id, + "--label", + label, + "--devices.sda.disk_id", + disk_id, + "--no-headers", + "--format=id", + "--text", + ] + ) + .stdout.decode() + .rstrip() + ) + + yield config_id + + +@pytest.fixture(scope="session", autouse=True) +def test_vpc_w_subnet(request): + vpc_json = create_vpc_w_subnet() + vpc_id = str(vpc_json["id"]) + + yield vpc_id + + +@pytest.fixture(scope="session", autouse=True) +def cleanup_vpc(request, test_vpc_w_subnet): + # Register finalizer to delete VPC after the entire session, with a delay + def delayed_cleanup(): + time.sleep(5) # Delay if necessary + delete_target_id(target="vpcs", id=test_vpc_w_subnet) + + request.addfinalizer(delayed_cleanup) + + +@pytest.fixture(scope="session", autouse=True) +def config_vpc_interface( + linode_instance_config_tests, linode_disk_config, test_vpc_w_subnet +): + linode_id = linode_instance_config_tests + config_id = linode_disk_config + subnet_id = get_subnet_id(vpc_id=test_vpc_w_subnet) + + interface_id = ( + exec_test_command( + BASE_CMD + + [ + "config-interface-add", + linode_id, + config_id, + "--purpose", + "vpc", + "--primary", + "false", + "--subnet_id", + subnet_id, + "--ipv4.vpc", + "10.0.0.3", + "--text", + "--no-headers", + "--format", + "id", + ] + ) + .stdout.decode() + .rstrip() + ) + + yield interface_id + + retry_exec_test_command_with_delay( + BASE_CMD + + [ + "config-interface-delete", + linode_id, + config_id, + interface_id, + ] + ) diff --git a/tests/integration/linodes/helpers_linodes.py b/tests/integration/linodes/helpers_linodes.py index e6b7323e0..8ba08c8ed 100644 --- a/tests/integration/linodes/helpers_linodes.py +++ b/tests/integration/linodes/helpers_linodes.py @@ -2,48 +2,34 @@ from tests.integration.helpers import exec_test_command -DEFAULT_RANDOM_PASS = ( - exec_test_command(["openssl", "rand", "-base64", "32"]) - .stdout.decode() - .rstrip() -) +DEFAULT_RANDOM_PASS = exec_test_command(["openssl", "rand", "-base64", "32"]) DEFAULT_REGION = "us-ord" -DEFAULT_TEST_IMAGE = ( - exec_test_command( - [ - "linode-cli", - "images", - "list", - "--text", - "--format", - "id", - "--no-headers", - "--is_public", - "True", - ] - ) - .stdout.decode() - .rstrip() - .splitlines()[0] -) +DEFAULT_TEST_IMAGE = exec_test_command( + [ + "linode-cli", + "images", + "list", + "--text", + "--format", + "id", + "--no-headers", + "--is_public", + "True", + ] +).splitlines()[0] -DEFAULT_LINODE_TYPE = ( - exec_test_command( - [ - "linode-cli", - "linodes", - "types", - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines()[0] -) +DEFAULT_LINODE_TYPE = exec_test_command( + [ + "linode-cli", + "linodes", + "types", + "--text", + "--no-headers", + "--format", + "id", + ] +).splitlines()[0] DEFAULT_LABEL = "cli-default" @@ -64,7 +50,7 @@ def wait_until(linode_id: "str", timeout, status: "str", period=5): "--text", "--no-headers", ] - ).stdout.decode() + ).stdout if status in result: return True time.sleep(period) @@ -75,31 +61,27 @@ def create_linode( firewall_id: "str", test_region=DEFAULT_REGION, disk_encryption=False ): # create linode - linode_id = ( - exec_test_command( - [ - "linode-cli", - "linodes", - "create", - "--type", - DEFAULT_LINODE_TYPE, - "--region", - test_region, - "--image", - DEFAULT_TEST_IMAGE, - "--root_pass", - DEFAULT_RANDOM_PASS, - "--firewall_id", - firewall_id, - "--disk_encryption", - "enabled" if disk_encryption else "disabled", - "--format=id", - "--text", - "--no-headers", - ] - ) - .stdout.decode() - .rstrip() + linode_id = exec_test_command( + [ + "linode-cli", + "linodes", + "create", + "--type", + DEFAULT_LINODE_TYPE, + "--region", + test_region, + "--image", + DEFAULT_TEST_IMAGE, + "--root_pass", + DEFAULT_RANDOM_PASS, + "--firewall_id", + firewall_id, + "--disk_encryption", + "enabled" if disk_encryption else "disabled", + "--format=id", + "--text", + "--no-headers", + ] ) return linode_id @@ -111,71 +93,57 @@ def create_linode_backup_disabled( result = set_backups_enabled_in_account_settings(toggle=False) # create linode - linode_id = ( - exec_test_command( - [ - "linode-cli", - "linodes", - "create", - "--type", - DEFAULT_LINODE_TYPE, - "--region", - test_region, - "--image", - DEFAULT_TEST_IMAGE, - "--root_pass", - DEFAULT_RANDOM_PASS, - "--firewall_id", - firewall_id, - "--format=id", - "--text", - "--no-headers", - "--backups_enabled", - "false", - ] - ) - .stdout.decode() - .rstrip() + linode_id = exec_test_command( + [ + "linode-cli", + "linodes", + "create", + "--type", + DEFAULT_LINODE_TYPE, + "--region", + test_region, + "--image", + DEFAULT_TEST_IMAGE, + "--root_pass", + DEFAULT_RANDOM_PASS, + "--firewall_id", + firewall_id, + "--format=id", + "--text", + "--no-headers", + "--backups_enabled", + "false", + ] ) return linode_id def shutdown_linodes(): - linode_ids = ( - exec_test_command( - [ - BASE_CMD, - "linodes", - "list", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + linode_ids = exec_test_command( + [ + BASE_CMD, + "linodes", + "list", + "--format", + "id", + ] + ).splitlines() for id in linode_ids: exec_test_command(["linode-cli", "linodes", "shutdown", id]) def remove_linodes(): - linode_ids = ( - exec_test_command( - [ - BASE_CMD, - "linodes", - "list", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + linode_ids = exec_test_command( + [ + BASE_CMD, + "linodes", + "list", + "--format", + "id", + ] + ).splitlines() for id in linode_ids: exec_test_command(["linode-cli", "linodes", "delete", id]) @@ -217,7 +185,7 @@ def create_linode_and_wait( if ssh_key: command.extend(["--authorized_keys", ssh_key]) - linode_id = exec_test_command(command).stdout.decode().strip() + linode_id = exec_test_command(command).stdout.strip() # wait until linode is running, wait_until returns True when it is in running state result = wait_until(linode_id=linode_id, timeout=240, status="running") @@ -243,27 +211,22 @@ def set_backups_enabled_in_account_settings(toggle: bool): else: command.extend(["--backups_enabled", "false"]) - result = exec_test_command(command).stdout.decode().rstrip() + result = exec_test_command(command) return result def get_disk_ids(linode_id): - disk_ids = ( - exec_test_command( - BASE_CMD - + [ - "disks-list", - linode_id, - "--text", - "--no-headers", - "--format", - "id", - ] - ) - .stdout.decode() - .rstrip() - .splitlines() - ) + disk_ids = exec_test_command( + BASE_CMD + + [ + "disks-list", + linode_id, + "--text", + "--no-headers", + "--format", + "id", + ] + ).splitlines() return disk_ids diff --git a/tests/integration/linodes/test_backups.py b/tests/integration/linodes/test_backups.py index dd5c904e7..b78be3dca 100755 --- a/tests/integration/linodes/test_backups.py +++ b/tests/integration/linodes/test_backups.py @@ -12,10 +12,6 @@ set_backups_enabled_in_account_settings, ) -# ################################################################## -# # WARNING: THIS TEST WILL DELETE ALL OF YOUR LINODES # -# # WARNING: USE A SEPARATE TEST ACCOUNT WHEN RUNNING THESE TESTS # -# ################################################################## snapshot_label = "test_snapshot1" @@ -55,7 +51,7 @@ def check_account_settings(): "managed", "--no-headers", ] - ).stdout.decode() + ).stdout return result @@ -79,7 +75,7 @@ def test_create_linode_with_backup_disabled( "--text", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search(linode_id + ",False", result) @@ -108,7 +104,7 @@ def test_enable_backups(create_linode_setup): "--text", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search(linode_id + ",True", result) @@ -125,7 +121,7 @@ def test_create_backup_with_backup_enabled(linode_backup_enabled): "--text", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search(linode_id + ",True", result) @@ -150,7 +146,7 @@ def test_take_snapshot_of_linode(): ",", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search( "[0-9]+,pending,snapshot,[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+," + snapshot_label, @@ -177,7 +173,7 @@ def test_view_the_snapshot(snapshot_of_linode): "--text", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search( "[0-9]+,pending,snapshot,[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+," @@ -207,7 +203,7 @@ def test_cancel_backups(snapshot_of_linode): ",", "--no-headers", ] - ).stdout.decode() + ).stdout assert re.search( "[0-9]+,pending,snapshot,[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+," + new_snapshot_label, diff --git a/tests/integration/networking/test_networking.py b/tests/integration/networking/test_networking.py index 2d29dd166..a7645eb5c 100644 --- a/tests/integration/networking/test_networking.py +++ b/tests/integration/networking/test_networking.py @@ -83,7 +83,7 @@ def test_view_an_ip_address(test_linode_id): "--text", "--no-headers", ] - ).stdout.rstrip() + ) result = exec_test_command( BASE_CMD