From 251dbbc6a3258884cad74bfdaac1a6062337e8d2 Mon Sep 17 00:00:00 2001 From: Michal Wojcik <32574975+mgwoj@users.noreply.github.com> Date: Fri, 18 Sep 2026 18:57:16 +0200 Subject: [PATCH 1/6] TPT-4175: cli: Update interactive config token access check (#881) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Zhiwei Liang <121905282+zliang-akamai@users.noreply.github.com> Co-authored-by: Zhiwei Liang --- linodecli/configuration/__init__.py | 1 - linodecli/configuration/auth.py | 29 ------- linodecli/configuration/config.py | 27 ++++-- tests/unit/test_configuration.py | 129 +++++++++++++++++++++++++++- 4 files changed, 147 insertions(+), 39 deletions(-) diff --git a/linodecli/configuration/__init__.py b/linodecli/configuration/__init__.py index 93062629a..a89c8affa 100644 --- a/linodecli/configuration/__init__.py +++ b/linodecli/configuration/__init__.py @@ -5,7 +5,6 @@ # Private methods need to be imported explicitly from .auth import * from .auth import ( - _check_full_access, _do_get_request, _get_token_terminal, _get_token_web, diff --git a/linodecli/configuration/auth.py b/linodecli/configuration/auth.py index a23db3f86..5f68b2175 100644 --- a/linodecli/configuration/auth.py +++ b/linodecli/configuration/auth.py @@ -152,35 +152,6 @@ def _do_request( return result.json() -def _check_full_access(base_url: str, token: str) -> bool: - """ - Checks whether the given token has full-access permissions. - - :param base_url: The base URL for the API. - :type base_url: str - :param token: The access token to use. - :type token :str - - :returns: Whether the user has full access. - :rtype: bool - """ - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - } - - result = requests.get( - base_url + "/profile/grants", - headers=headers, - timeout=120, - verify=API_CA_PATH, - ) - - _handle_response_status(result, exit_on_error=True) - - return result.status_code == 204 - - def _username_for_token(base_url: str, token: str) -> str: """ A helper function that returns the username associated with a token by diff --git a/linodecli/configuration/config.py b/linodecli/configuration/config.py index 9dc85d9dc..1a98b033c 100644 --- a/linodecli/configuration/config.py +++ b/linodecli/configuration/config.py @@ -10,7 +10,6 @@ from linodecli.exit_codes import ExitCodes from .auth import ( - _check_full_access, _do_get_request, _get_token_terminal, _get_token_web, @@ -430,18 +429,32 @@ def configure( [i["id"] for i in _do_get_request(self.base_url, "/images")["data"]] ) - is_full_access = _check_full_access(self.base_url, token) - auth_users = [] - if is_full_access: + # Use /profile/sshkeys as a lightweight capability check. If the token + # lacks profile access (401) or the user is IAM-enrolled (403), skip + # the authorized_users prompt entirely. Capture the status code via a + # closure passed as status_validator so no new API surface is needed. + sshkeys_status_code = None + + def _capture_sshkeys_status(status: int) -> bool: + nonlocal sshkeys_status_code + sshkeys_status_code = status + return status in (401, 403) + + _do_get_request( + self.base_url, + "/profile/sshkeys", + token=token, + status_validator=_capture_sshkeys_status, + ) + + if sshkeys_status_code == 200: users = _do_get_request( self.base_url, "/account/users", token=token, - # Allow 401 responses so tokens without - # account perms can be configured - status_validator=lambda status: status == 401, + status_validator=lambda status: status in (401, 403), ) if "data" in users: diff --git a/tests/unit/test_configuration.py b/tests/unit/test_configuration.py index 409250330..c68a2416f 100644 --- a/tests/unit/test_configuration.py +++ b/tests/unit/test_configuration.py @@ -298,7 +298,7 @@ def mock_input(prompt): requests_mock.Mocker() as m, ): m.get(f"{self.base_url}/profile", json={"username": "cli-dev"}) - m.get(f"{self.base_url}/profile/grants", status_code=204) + m.get(f"{self.base_url}/profile/sshkeys", json={"data": []}) m.get( f"{self.base_url}/regions", json={"data": [{"id": "test-region"}]}, @@ -349,7 +349,7 @@ def mock_input(prompt): requests_mock.Mocker() as m, ): m.get(f"{self.base_url}/profile", json={"username": "cli-dev"}) - m.get(f"{self.base_url}/profile/grants", status_code=204) + m.get(f"{self.base_url}/profile/sshkeys", json={"data": []}) m.get( f"{self.base_url}/regions", json={"data": [{"id": "test-region"}]}, @@ -676,3 +676,128 @@ def test_custom_config_path(self, monkeypatch, tmp_path): for i, _ in enumerate(expected_configs): assert expected_configs[i] == configs[i] + + +class TestConfigureAuthorizedUsers: + """ + Unit tests for the authorized_users gate in CLIConfig.configure(), + which uses GET /profile/sshkeys as a capability check instead of + GET /profile/grants (which is inaccessible to IAM-enrolled users). + """ + + base_url = "https://linode-test.com" + test_token = "cli-dev-token" + + def _base_mocks(self, m, *, sshkeys_status=200, users_json=None): + """Register the common mocks needed for a configure() call.""" + m.get(f"{self.base_url}/profile", json={"username": "cli-dev"}) + m.get( + f"{self.base_url}/profile/sshkeys", + status_code=sshkeys_status, + json={"data": []}, + ) + m.get(f"{self.base_url}/regions", json={"data": [{"id": "us-east"}]}) + m.get( + f"{self.base_url}/linode/types", + json={"data": [{"id": "g6-nanode-1"}]}, + ) + m.get( + f"{self.base_url}/images", + json={"data": [{"id": "linode/debian12"}]}, + ) + if users_json is not None: + m.get(f"{self.base_url}/account/users", json=users_json) + + def test_sshkeys_success_populates_authorized_users(self): + """ + When /profile/sshkeys succeeds, /account/users is fetched and + authorized_users is offered. + """ + conf = configuration.CLIConfig(self.base_url, skip_config=True) + answers = iter(["1", "1", "1", "1", "n", "n"]) + + with ( + patch("linodecli.configuration.open", mock_open()), + patch("os.chmod", lambda a, b: None), + patch("builtins.input", lambda _: next(answers)), + contextlib.redirect_stdout(io.StringIO()), + patch("linodecli.configuration._check_browsers", lambda: False), + patch.dict(os.environ, {"LINODE_CLI_TOKEN": self.test_token}), + requests_mock.Mocker() as m, + ): + self._base_mocks( + m, + users_json={ + "data": [{"username": "cli-dev", "ssh_keys": ["key1"]}] + }, + ) + conf.configure() + + assert conf.get_value("authorized_users") == "cli-dev" + + def test_sshkeys_401_skips_authorized_users(self): + """ + When /profile/sshkeys returns 401 (restricted token), the + authorized_users prompt is skipped. + """ + conf = configuration.CLIConfig(self.base_url, skip_config=True) + answers = iter(["1", "1", "1", "n", "n"]) + + with ( + patch("linodecli.configuration.open", mock_open()), + patch("os.chmod", lambda a, b: None), + patch("builtins.input", lambda _: next(answers)), + contextlib.redirect_stdout(io.StringIO()), + patch("linodecli.configuration._check_browsers", lambda: False), + patch.dict(os.environ, {"LINODE_CLI_TOKEN": self.test_token}), + requests_mock.Mocker() as m, + ): + self._base_mocks(m, sshkeys_status=401) + conf.configure() + + assert conf.get_value("authorized_users") is None + + def test_sshkeys_403_iam_skips_authorized_users(self): + """ + When /profile/sshkeys returns 403 (IAM-enrolled user), the + authorized_users prompt is skipped. + """ + conf = configuration.CLIConfig(self.base_url, skip_config=True) + answers = iter(["1", "1", "1", "n", "n"]) + + with ( + patch("linodecli.configuration.open", mock_open()), + patch("os.chmod", lambda a, b: None), + patch("builtins.input", lambda _: next(answers)), + contextlib.redirect_stdout(io.StringIO()), + patch("linodecli.configuration._check_browsers", lambda: False), + patch.dict(os.environ, {"LINODE_CLI_TOKEN": self.test_token}), + requests_mock.Mocker() as m, + ): + self._base_mocks(m, sshkeys_status=403) + conf.configure() + + assert conf.get_value("authorized_users") is None + + def test_sshkeys_unexpected_status_exits(self): + """ + When /profile/sshkeys returns an unexpected error (e.g. 500), + configure() should exit rather than skipping authorized_users. + """ + conf = configuration.CLIConfig(self.base_url, skip_config=True) + + with ( + patch("linodecli.configuration.open", mock_open()), + patch("os.chmod", lambda a, b: None), + patch("builtins.input", lambda _: None), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + patch("linodecli.configuration._check_browsers", lambda: False), + patch.dict(os.environ, {"LINODE_CLI_TOKEN": self.test_token}), + requests_mock.Mocker() as m, + pytest.raises(SystemExit) as err, + ): + self._base_mocks(m, sshkeys_status=500) + conf.configure() + + assert err.value.code == 2 From ca78c9b7dc9676d01c54ebeb4b05f279cfd6ddee Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 14:42:41 -0400 Subject: [PATCH 2/6] build(deps): bump WyriHaximus/github-action-get-previous-tag from 2.0.0 to 2.1.0 (#929) Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --- .github/workflows/remote-release-trigger.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/remote-release-trigger.yml b/.github/workflows/remote-release-trigger.yml index f3c042f8a..584688c0d 100644 --- a/.github/workflows/remote-release-trigger.yml +++ b/.github/workflows/remote-release-trigger.yml @@ -23,7 +23,7 @@ jobs: - name: Get previous tag id: previoustag - uses: WyriHaximus/github-action-get-previous-tag@61819f33034117e6c686e6a31dba995a85afc9de # pin@v2.0.0 + uses: WyriHaximus/github-action-get-previous-tag@83f26fea93bc7efcbca2eb5591f5eaaf66b8f206 # pin@v2.1.0 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} From 3b75a8aec6c62c7a029bd489e8cd1315dba397a6 Mon Sep 17 00:00:00 2001 From: Maciej Wilk Date: Thu, 15 Oct 2026 12:17:04 +0200 Subject: [PATCH 3/6] TPT-4723: Retest and refactor tests skipped due to ARB-8019 (#930) --- tests/integration/vpc/test_vpc.py | 39 +++++++++++++++++++------------ 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/tests/integration/vpc/test_vpc.py b/tests/integration/vpc/test_vpc.py index 23a51f4d1..4db622c92 100644 --- a/tests/integration/vpc/test_vpc.py +++ b/tests/integration/vpc/test_vpc.py @@ -171,19 +171,21 @@ def test_subnet_with_rdma_type_vpc(get_test_subnet_w_rdma_type): assert output["vpc_type"] == "rdma" -@pytest.mark.skip(reason="Defect: ARB-8019") def test_fails_to_create_vpc_invalid_label(): - invalid_label = "invalid_label" + invalid_label = "invalid_label!?#" region = get_random_region_with_caps(required_capabilities=["VPCs"]) res = exec_failing_test_command( BASE_CMDS["vpcs"] - + ["create", "--label", invalid_label, "--region", region], + + ["create", "--label", invalid_label, "--region", region, "--text"], ExitCodes.REQUEST_FAILED, ) assert "Request failed: 400" in res - assert "Must only use ASCII letters, numbers, and dashes" in res + assert ( + "Must only use ASCII letters, numbers, underscores, dashes and periods" + in res + ) def test_fails_to_create_vpc_duplicate_label(get_test_vpc_wo_subnet): @@ -202,24 +204,26 @@ def test_fails_to_create_vpc_duplicate_label(get_test_vpc_wo_subnet): assert "Label must be unique among your VPCs" in res -@pytest.mark.skip(reason="Defect: ARB-8019") def test_fails_to_update_vpc_invalid_label(get_test_vpc_wo_subnet): vpc_id = get_test_vpc_wo_subnet - invalid_label = "invalid_label" + invalid_label = "invalid_label!?#" res = exec_failing_test_command( - BASE_CMDS["vpcs"] + ["update", vpc_id, "--label", invalid_label], + BASE_CMDS["vpcs"] + + ["update", vpc_id, "--label", invalid_label, "--text"], ExitCodes.REQUEST_FAILED, ) assert "Request failed: 400" in res - assert "Must only use ASCII letters, numbers, and dashes" in res + assert ( + "Must only use ASCII letters, numbers, underscores, dashes and periods" + in res + ) -@pytest.mark.skip(reason="Defect: ARB-8019") def test_fails_to_create_vpc_subnet_w_invalid_label(get_test_vpc_wo_subnet): vpc_id = get_test_vpc_wo_subnet - invalid_label = "invalid_label" + invalid_label = "invalid_label!?#" res = exec_failing_test_command( BASE_CMDS["vpcs"] @@ -230,19 +234,21 @@ def test_fails_to_create_vpc_subnet_w_invalid_label(get_test_vpc_wo_subnet): "--ipv4", "10.1.0.0/24", vpc_id, + "--text", ], ExitCodes.REQUEST_FAILED, ) assert "Request failed: 400" in res - assert "Must only use ASCII letters, numbers, and dashes" in res + assert ( + "Must only use ASCII letters, numbers, underscores, dashes and periods" + in res + ) -@pytest.mark.skip(reason="Defect: ARB-8019") def test_fails_to_update_vpc_subnet_w_invalid_label(get_test_vpc_w_subnet): vpc_id = get_test_vpc_w_subnet - - invalid_label = "invalid_label" + invalid_label = "invalid_label!?#" subnet_id = exec_test_command( BASE_CMDS["vpcs"] @@ -265,7 +271,10 @@ def test_fails_to_update_vpc_subnet_w_invalid_label(get_test_vpc_w_subnet): ) assert "Request failed: 400" in res - assert "Must only use ASCII letters, numbers, and dashes" in res + assert ( + "Must only use ASCII letters, numbers, underscores, dashes and periods" + in res + ) def test_create_vpc_with_ipv6_auto(create_vpc_with_ipv6): From d86a2cb2f619aee7400275c8e0ef12bab16eaa8c Mon Sep 17 00:00:00 2001 From: Maciej Wilk Date: Tue, 20 Oct 2026 13:32:14 +0200 Subject: [PATCH 4/6] TPT-4741: Fix failing backup tests (#934) --- tests/integration/linodes/fixtures.py | 4 ++-- tests/integration/linodes/test_backups.py | 19 ++++++++++++------- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/tests/integration/linodes/fixtures.py b/tests/integration/linodes/fixtures.py index 8e57df9a3..7542ae47d 100644 --- a/tests/integration/linodes/fixtures.py +++ b/tests/integration/linodes/fixtures.py @@ -86,9 +86,9 @@ def linode_backup_disabled(linode_cloud_firewall): @pytest.fixture(scope="module") -def snapshot_of_linode(): +def snapshot_of_linode(linode_cloud_firewall): # get linode id after creation and wait for "running" status - linode_id = create_linode_and_wait() + linode_id = create_linode_and_wait(firewall_id=linode_cloud_firewall) new_snapshot_label = "test_snapshot_" + get_random_text(5) result = exec_test_command( diff --git a/tests/integration/linodes/test_backups.py b/tests/integration/linodes/test_backups.py index 0949fbd5b..a4c0642c6 100755 --- a/tests/integration/linodes/test_backups.py +++ b/tests/integration/linodes/test_backups.py @@ -3,11 +3,16 @@ import pytest -from tests.integration.helpers import BASE_CMDS, exec_test_command +from tests.integration.helpers import ( + BASE_CMDS, + delete_target_id, + exec_test_command, +) from tests.integration.linodes.fixtures import ( # noqa: F401 linode_backup_disabled, linode_backup_enabled, linode_basic_with_firewall, + snapshot_of_linode, ) from tests.integration.linodes.helpers import ( check_account_settings, @@ -91,9 +96,9 @@ def test_create_backup_with_backup_enabled(linode_backup_enabled): os.environ.get("RUN_LONG_TESTS", None) != "True", reason="Skipping long-running Test, to run set RUN_LONG_TESTS=True", ) -def test_take_snapshot_of_linode(firewall_id): +def test_take_snapshot_of_linode(linode_cloud_firewall): # get linode id after creation and wait for "running" status - linode_id = create_linode_and_wait(firewall_id) + linode_id = create_linode_and_wait(firewall_id=linode_cloud_firewall) snapshot_label = "test_snapshot1" @@ -116,6 +121,8 @@ def test_take_snapshot_of_linode(firewall_id): result, ) + delete_target_id("linodes", linode_id) + @pytest.mark.skipif( os.environ.get("RUN_LONG_TESTS", None) != "True", @@ -157,10 +164,8 @@ def test_cancel_backups(snapshot_of_linode): result = exec_test_command( BASE_CMDS["linodes"] + [ - "snapshot", + "backups-list", linode_id, - "--label", - new_snapshot_label, "--text", "--delimiter", ",", @@ -168,7 +173,7 @@ def test_cancel_backups(snapshot_of_linode): ] ) assert re.search( - "[0-9]+,pending,snapshot,[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+," + "[0-9]+,(pending|running),snapshot,[0-9]+-[0-9]+-[0-9]+T[0-9]+:[0-9]+:[0-9]+," + new_snapshot_label, result, ) From 7d9343e5d683f5332ac3019dc013e7a3a499093b Mon Sep 17 00:00:00 2001 From: Zhiwei Liang <121905282+zliang-akamai@users.noreply.github.com> Date: Wed, 21 Oct 2026 02:05:03 -0400 Subject: [PATCH 5/6] TPT-4731: Fix nested CLI arguments for schemas without an explicit type (#931) --- linodecli/baked/AGENTS.md | 2 + linodecli/baked/request.py | 9 +++- linodecli/baked/response.py | 4 +- tests/fixtures/operation_with_one_ofs.yaml | 1 - tests/unit/test_request.py | 51 ++++++++++++++++++++++ tests/unit/test_response.py | 2 + 6 files changed, 65 insertions(+), 4 deletions(-) diff --git a/linodecli/baked/AGENTS.md b/linodecli/baked/AGENTS.md index f92d13c6b..cdc386e42 100644 --- a/linodecli/baked/AGENTS.md +++ b/linodecli/baked/AGENTS.md @@ -30,6 +30,7 @@ ## Request Parsing Gotchas - `_aggregate_schema_properties()` merges `oneOf`, `anyOf`, and `allOf`; a field is marked required only if every leaf schema that defines properties requires it. Schemas that only nest composition without their own `properties` do not increment the required-count denominator. +- Nested request objects are expanded when `_aggregate_schema_properties()` finds properties and the schema has `type: object` or no explicit type, unless the effective format is `json`. Explicit scalar and array types retain their dedicated parsing paths. A nested `oneOf`/`anyOf`/`allOf` without `type` normally becomes dotted CLI args (for example `--details.host`); `x-linode-cli-format: json` or `format: json` instead preserves a raw-JSON parent argument, with the CLI extension taking precedence. - `OpenAPIRequest.attr_routes` stores per-`oneOf` option args keyed by schema `title`; oneOf entries without titles raise `ValueError`. - `OpenAPIOperation.arg_routes` returns `self.request.attr_routes` (a dict) when a request exists, but `[]` when there is no request. Treat the empty case as a type inconsistency (list vs dict); callers that always call `.items()` will break on the no-request path. - Request and response attributes marked `linode-cli-skip` are omitted from generated args/output. @@ -48,6 +49,7 @@ - `--order-by` is restricted to filterable attrs and becomes `+order_by` in `X-Filter`; list filters serialize as `+and` entries. ## Response And Output Gotchas +- Nested response schemas with aggregated properties are expanded even without `type: object`; arbitrary-key dictionaries retain their existing leaf handling, and scalar compositions without properties are not expanded. - Bake-time pagination detection is structural: top-level response schema must have exactly `pages`, `page`, `results`, and `data`; only `data.items` becomes display attrs (`is_paginated`). - `OpenAPIResponse.fix_json()` handles `linode-cli-rows`, then `linode-cli-nested-list`, then unwraps via `"pages" in json` (not `self.is_paginated`), then wraps non-lists. Runtime unwrap is looser than bake-time pagination; do not tighten it to `is_paginated` without checking odd payloads that only include `pages`. - `linode-cli-rows` paths are extracted from the raw API JSON and concatenated; list values extend the result, scalar/object values are appended. diff --git a/linodecli/baked/request.py b/linodecli/baked/request.py index bbd255288..8648be043 100644 --- a/linodecli/baked/request.py +++ b/linodecli/baked/request.py @@ -164,11 +164,16 @@ def _parse_request_model( if v.extensions.get("linode-cli-skip"): continue + # Nested oneOf/anyOf/allOf schemas may omit type: object. + nested_props, _ = _aggregate_schema_properties(v) + effective_format = v.extensions.get("linode-cli-format") or v.format + # Handle nested objects which aren't read-only and have properties if ( - v.type == "object" + v.type in (None, "object") and not v.readOnly - and len(_aggregate_schema_properties(v)[0]) > 0 + and nested_props + and effective_format != "json" ): # nested objects receive a prefix and are otherwise parsed normally pref = prefix + "." + k if prefix else k diff --git a/linodecli/baked/response.py b/linodecli/baked/response.py index 901ece294..2b5cae2dd 100644 --- a/linodecli/baked/response.py +++ b/linodecli/baked/response.py @@ -208,7 +208,9 @@ def _parse_response_model(schema, prefix=None, nested_list_depth=0): k, v, prefix=prefix, nested_list_depth=nested_list_depth ) ) - elif v.type == "object": + elif v.type == "object" or ( + v.type is None and bool(_aggregate_schema_properties(v)[0]) + ): attrs += _parse_response_model(v, prefix=pref) elif v.type == "array" and ( v.items.type == "object" diff --git a/tests/fixtures/operation_with_one_ofs.yaml b/tests/fixtures/operation_with_one_ofs.yaml index 679e04ccb..79faccfdc 100644 --- a/tests/fixtures/operation_with_one_ofs.yaml +++ b/tests/fixtures/operation_with_one_ofs.yaml @@ -57,7 +57,6 @@ components: description: Some foofoo. barbar: description: Some barbar. - type: object anyOf: - type: object properties: diff --git a/tests/unit/test_request.py b/tests/unit/test_request.py index 81f55d07a..6b5260bb3 100644 --- a/tests/unit/test_request.py +++ b/tests/unit/test_request.py @@ -1,3 +1,11 @@ +import pytest +from openapi3 import OpenAPI +from yaml import safe_load + +from linodecli.baked.request import _parse_request_model +from tests.unit.conftest import open_fixture + + class TestRequest: """ Unit tests for baked requests. @@ -17,11 +25,54 @@ def test_handle_one_ofs(self, post_operation_with_one_ofs): "barbar.baz": ("boolean", "Some baz.", False), } + assert len(arg_map) == len(expected) + for k, v in expected.items(): assert arg_map[k].datatype == v[0] assert arg_map[k].description == v[1] assert arg_map[k].required == v[2] + @pytest.mark.parametrize("format_marker", ["x-linode-cli-format", "format"]) + @pytest.mark.parametrize("schema_type", [None, "object"]) + def test_json_format_composed_objects(self, format_marker, schema_type): + with open_fixture("operation_with_one_ofs.yaml") as fixture: + document = safe_load(fixture) + + variant = document["components"]["schemas"]["Foo"]["oneOf"][1] + nested_schema = variant["properties"]["barbar"] + nested_schema[format_marker] = "json" + if schema_type is not None: + nested_schema["type"] = schema_type + + schema = OpenAPI(document).components.schemas["Foo"] + arg_map = {arg.path: arg for arg in _parse_request_model(schema)} + + assert set(arg_map) == {"foobar", "barfoo", "foofoo", "barbar"} + assert arg_map["barbar"].format == "json" + assert arg_map["barbar"].datatype == "object" + + @pytest.mark.parametrize("schema_type", ["string", "integer", "array"]) + def test_explicit_non_object_composed_schema(self, schema_type): + with open_fixture("operation_with_one_ofs.yaml") as fixture: + document = safe_load(fixture) + + variant = document["components"]["schemas"]["Foo"]["oneOf"][1] + nested_schema = variant["properties"]["barbar"] + nested_schema["type"] = schema_type + for branch in nested_schema["anyOf"]: + branch.pop("type") + if schema_type == "array": + nested_schema["items"] = {"type": "string"} + + schema = OpenAPI(document).components.schemas["Foo"] + arg_map = {arg.path: arg for arg in _parse_request_model(schema)} + + assert set(arg_map) == {"foobar", "barfoo", "foofoo", "barbar"} + assert arg_map["barbar"].datatype == schema_type + assert arg_map["barbar"].format is None + if schema_type == "array": + assert arg_map["barbar"].item_type == "string" + def test_skip_request_attributes(self, post_skip_test_operation): """ Test that request attributes with x-linode-cli-skip extension are excluded. diff --git a/tests/unit/test_response.py b/tests/unit/test_response.py index f5e3632a3..4e740fee0 100644 --- a/tests/unit/test_response.py +++ b/tests/unit/test_response.py @@ -114,6 +114,8 @@ def test_handle_one_ofs(self, post_operation_with_one_ofs): "barbar.baz": ("boolean", "Some baz"), } + assert len(attr_map) == len(expected) + for k, v in expected.items(): assert attr_map[k].datatype == v[0] assert attr_map[k].description == v[1] From cc40f8ddcc390136756a5fdecae0b01272be89cc Mon Sep 17 00:00:00 2001 From: Pawel <100145168+psnoch-akamai@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:28:48 +0100 Subject: [PATCH 6/6] TPT-4651: Add integration tests for Premium NodeBalancer changes (#933) --- tests/integration/helpers.py | 9 ++ tests/integration/nodebalancers/fixtures.py | 79 ++++++++++++++++ .../nodebalancers/test_node_balancers.py | 91 +++++++++++++++++++ 3 files changed, 179 insertions(+) diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py index a1ba7d6ef..84dbc1720 100644 --- a/tests/integration/helpers.py +++ b/tests/integration/helpers.py @@ -224,6 +224,15 @@ def assert_help_actions_list(expected_actions, help_output): assert expected_action in output_actions +def assert_help_parameters_list_for_command(expected_parameters, help_output): + for expected_parameter in expected_parameters: + assert re.search( + rf"^\s*--{re.escape(expected_parameter)}(?=\s|:)", + help_output, + re.MULTILINE, + ), f"Missing help parameter: --{expected_parameter}" + + def view_command_attribute( command: str, action: str, item_id: str, attribute: str ) -> str: diff --git a/tests/integration/nodebalancers/fixtures.py b/tests/integration/nodebalancers/fixtures.py index 958b802de..3f56de53b 100644 --- a/tests/integration/nodebalancers/fixtures.py +++ b/tests/integration/nodebalancers/fixtures.py @@ -5,6 +5,8 @@ check_attribute_value, delete_target_id, exec_test_command, + get_random_region_with_caps, + get_random_text, wait_for_condition, ) from tests.integration.linodes.helpers import DEFAULT_TEST_IMAGE @@ -362,3 +364,80 @@ def nodebalancer_with_reserved_ipv4(create_reserved_ip): yield res_ip, nodebalancer_id delete_target_id(target="nodebalancers", id=nodebalancer_id) + + +@pytest.fixture +def nodebalancer_premium_with_backend_vpc( + linode_cloud_firewall, + vpc_for_nb_premium_backend_connectivity, +): + test_subnet_id = exec_test_command( + BASE_CMDS["vpcs"] + + [ + "subnets-list", + vpc_for_nb_premium_backend_connectivity[0], + "--text", + "--no-headers", + "--delimiter=,", + ] + ).split(",")[0] + + result = exec_test_command( + BASE_CMDS["nodebalancers"] + + [ + "create", + "--region", + vpc_for_nb_premium_backend_connectivity[1], + "--firewall_id", + linode_cloud_firewall, + "--type", + "premium", + "--backend_connectivity", + "vpc", + "--vpcs.ipv4_range", + "192.168.0.64/30", + "--vpcs.subnet_id", + test_subnet_id, + "--text", + "--delimiter", + ",", + "--suppress-warnings", + "--no-headers", + ] + ) + nodebalancer_id = result.split(",")[0] + + yield nodebalancer_id, result + delete_target_id(target="nodebalancers", id=nodebalancer_id) + + +@pytest.fixture +def vpc_for_nb_premium_backend_connectivity(): + vpc_label = "test_vpc_" + get_random_text(5) + subnet_label = "test_vpc_" + get_random_text(5) + test_region = get_random_region_with_caps( + ["NodeBalancers", "Premium NodeBalancers"] + ) + + test_vpc_id = exec_test_command( + BASE_CMDS["vpcs"] + + [ + "create", + "--label", + vpc_label, + "--region", + test_region, + "--subnets.ipv4", + "192.168.0.0/25", + "--subnets.label", + subnet_label, + "--text", + "--suppress-warnings", + "--format", + "id", + "--no-headers", + ] + ) + + yield test_vpc_id, test_region + delete_target_id(target="vpcs", id=test_vpc_id) diff --git a/tests/integration/nodebalancers/test_node_balancers.py b/tests/integration/nodebalancers/test_node_balancers.py index 69564852a..95c8023e3 100644 --- a/tests/integration/nodebalancers/test_node_balancers.py +++ b/tests/integration/nodebalancers/test_node_balancers.py @@ -5,16 +5,21 @@ from linodecli.exit_codes import ExitCodes from tests.integration.helpers import ( BASE_CMDS, + assert_headers_in_lines, + assert_help_parameters_list_for_command, exec_failing_test_command, exec_test_command, + get_random_text, ) from tests.integration.nodebalancers.fixtures import ( # noqa: F401 linode_to_add, + nodebalancer_premium_with_backend_vpc, nodebalancer_w_config_and_node, nodebalancer_with_default_conf, nodebalancer_with_reserved_ipv4, nodebalancer_with_udp_config_and_node, simple_nodebalancer_with_config, + vpc_for_nb_premium_backend_connectivity, ) @@ -52,6 +57,28 @@ def test_list_nodebalancers_and_status(nodebalancer_w_config_and_node): assert re.search(nodebalancer_created(), result) +def test_verify_columns_for_list_nodebalancers(): + result = exec_test_command( + BASE_CMDS["nodebalancers"] + + [ + "list", + "--text", + "--delimiter", + ",", + ] + ) + lines = result.splitlines() + headers = [ + "id", + "label", + "region", + "type", + "backend_connectivity", + "backend_ipv6_prefix", + ] + assert_headers_in_lines(headers, lines) + + def test_display_public_ipv4_for_nodebalancer(nodebalancer_w_config_and_node): nodebalancer_id = nodebalancer_w_config_and_node[0] @@ -564,3 +591,67 @@ def test_view_node_for_node_balancer_udp_configuration( def nodebalancer_created(): return "[0-9]+,balancer[0-9]+,us-ord,[0-9]+-[0-9]+-[0-9]+-[0-9]+.ip.linodeusercontent.com,0" + + +def test_premium_nodebalancer_with_vpc_backend_connectivity( + nodebalancer_premium_with_backend_vpc, +): + result_create = nodebalancer_premium_with_backend_vpc[1] + assert "premium" in result_create + assert "vpc" in result_create + assert "/96" in result_create + + result_view = exec_test_command( + BASE_CMDS["nodebalancers"] + + [ + "view", + nodebalancer_premium_with_backend_vpc[0], + "--text", + "--no-headers", + "--format", + "type,backend_connectivity,backend_ipv6_prefix", + "--delimiter", + ",", + ] + ).split(",") + assert result_view[0] == "premium" + assert result_view[1] == "vpc" + assert result_view[2].endswith("/96") + + result_update = exec_test_command( + BASE_CMDS["nodebalancers"] + + [ + "update", + nodebalancer_premium_with_backend_vpc[0], + "--label", + "update_label_" + get_random_text(5), + "--client_conn_throttle", + "0", + "--text", + "--no-headers", + "--delimiter", + ",", + ] + ) + assert "premium" in result_update + assert "vpc" in result_update + assert "/96" in result_update + + +def test_create_nodebalancer_help_command(): + output = exec_test_command( + BASE_CMDS["nodebalancers"] + ["create", "--help", "--text"] + ) + available_arguments = [ + "backend_connectivity", + "client_conn_throttle", + "configs", + "firewall_id", + "ipv4", + "region", + "type", + "vpcs.ipv4_range", + "vpcs.ipv4_range_auto_assign", + "vpcs.subnet_id", + ] + assert_help_parameters_list_for_command(available_arguments, output)