From f317fdd343f7da1b9221381b9d04a59bda0f8c41 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Wed, 1 Jul 2026 12:42:10 +0200 Subject: [PATCH 01/10] feat(platform): add gcloud WIF credential helper command (PYSDK-146) Adds `aignostics user token` CLI command implementing the gcloud pluggable external credential source contract. Outputs an Aignostics id_token as JSON so gcloud can use it for Workload Identity Federation when accessing GCS. Co-Authored-By: Claude Sonnet 4.6 --- src/aignostics/platform/__init__.py | 3 +- src/aignostics/platform/_cli.py | 48 +++++++++++++++- tests/aignostics/platform/cli_test.py | 79 ++++++++++++++++++++++++++- 3 files changed, 126 insertions(+), 4 deletions(-) diff --git a/src/aignostics/platform/__init__.py b/src/aignostics/platform/__init__.py index fea6f14f4..d917b1f7c 100644 --- a/src/aignostics/platform/__init__.py +++ b/src/aignostics/platform/__init__.py @@ -35,7 +35,7 @@ from aignx.codegen.models import UserReadResponse as User from aignx.codegen.models import VersionReadResponse as ApplicationVersion -from ._cli import cli_sdk, cli_user +from ._cli import cli_auth, cli_sdk, cli_user from ._client import Client from ._constants import ( API_ROOT_DEV, @@ -184,6 +184,7 @@ "User", "UserInfo", "calculate_file_crc32c", + "cli_auth", "cli_sdk", "cli_user", "download_file", diff --git a/src/aignostics/platform/_cli.py b/src/aignostics/platform/_cli.py index 649783601..2c31e8d62 100644 --- a/src/aignostics/platform/_cli.py +++ b/src/aignostics/platform/_cli.py @@ -9,8 +9,10 @@ from aignostics.utils import console +from ._authentication import get_token from ._sdk_metadata import get_item_sdk_metadata_json_schema, get_run_sdk_metadata_json_schema from ._service import Service +from ._settings import settings cli_user = typer.Typer(name="user", help="User operations such as login, logout and whoami.") @@ -85,7 +87,6 @@ def whoami( logger.exception(message) console.print(message, style="error") sys.exit(1) - sys.exit(1) cli_sdk = typer.Typer(name="sdk", help="Platform operations such as dumping the SDK metadata schema.") @@ -135,3 +136,48 @@ def item_sdk_metadata_schema( logger.exception(message) console.print(message, style="error") sys.exit(1) + + +cli_auth = typer.Typer(name="auth", help="Authentication token operations for external integrations.") + + +@cli_user.command("token") +def auth_token() -> None: + """Print an Aignostics access token for use as a gcloud external credential helper. + + Outputs a JSON document to stdout in the format expected by gcloud's pluggable + authentication executable credential source (Workload Identity Federation). + + Configure as an executable credential source by adding the following to your ADC + credentials JSON file, then run ``gcloud auth application-default login`` to activate + it. The GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES environment variable must be set + to 1 for gcloud to call this helper. + + On success writes: {"version": 1, "success": true, "token_type": "...", "id_token": "...", "expiration_time": ...} + + On failure writes: {"version": 1, "success": false, "code": "1", "message": "..."} + """ + try: + token = get_token(use_cache=True) + stored = settings().token_file.read_text(encoding="utf-8") + expiry = int(stored.rsplit(":", 1)[-1]) + print( + json.dumps({ + "version": 1, + "success": True, + "token_type": "urn:ietf:params:oauth:token-type:id_token", + "id_token": token, + "expiration_time": expiry, + }) + ) + except Exception as e: + logger.debug("Failed to obtain Aignostics token for WIF credential helper: {}", e) + print( + json.dumps({ + "version": 1, + "success": False, + "code": "1", + "message": str(e), + }) + ) + sys.exit(1) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index 0395f5987..2874d3887 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -1,6 +1,7 @@ """Tests to verify the CLI functionality of the platform module.""" -from unittest.mock import patch +import json +from unittest.mock import MagicMock, patch import pytest from typer.testing import CliRunner @@ -609,7 +610,6 @@ def test_sdk_item_metadata_schema_no_pretty(runner: CliRunner) -> None: assert "schema_version" in output assert "platform_bucket" in output # In non-pretty mode, output should still be valid JSON - import json # Try to parse the output as JSON (should not raise an error) try: @@ -621,3 +621,78 @@ def test_sdk_item_metadata_schema_no_pretty(runner: CliRunner) -> None: pytest.fail("No JSON found in output") except json.JSONDecodeError: pytest.fail("Output is not valid JSON") + + +class TestAuthTokenCLI: + """Test cases for the ``aignostics auth token`` command (gcloud WIF credential helper).""" + + _MOCK_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MzQ1Njc4OTB9.sig" # noqa: S105 + _MOCK_EXPIRY = 1734567890 + + @pytest.mark.integration + @staticmethod + def test_auth_token_success_exit_code(record_property, runner: CliRunner) -> None: + """Exit code is 0 when a cached token is available.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + stored = f"{TestAuthTokenCLI._MOCK_TOKEN}:{TestAuthTokenCLI._MOCK_EXPIRY}" + mock_settings = MagicMock() + mock_settings.return_value.token_file.read_text.return_value = stored + + with ( + patch("aignostics.platform._cli.get_token", return_value=TestAuthTokenCLI._MOCK_TOKEN), + patch("aignostics.platform._cli.settings", mock_settings), + ): + result = runner.invoke(cli, ["auth", "token"]) + + assert result.exit_code == 0 + + @pytest.mark.integration + @staticmethod + def test_auth_token_success_json_structure(record_property, runner: CliRunner) -> None: + """Valid gcloud external-credential-helper JSON is written to stdout on success.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + stored = f"{TestAuthTokenCLI._MOCK_TOKEN}:{TestAuthTokenCLI._MOCK_EXPIRY}" + mock_settings = MagicMock() + mock_settings.return_value.token_file.read_text.return_value = stored + + with ( + patch("aignostics.platform._cli.get_token", return_value=TestAuthTokenCLI._MOCK_TOKEN), + patch("aignostics.platform._cli.settings", mock_settings), + ): + result = runner.invoke(cli, ["auth", "token"]) + + json_start = result.output.find("{") + response = json.loads(result.output[json_start:]) + + assert response["version"] == 1 + assert response["success"] is True + assert response["token_type"] == "urn:ietf:params:oauth:token-type:id_token" # noqa: S105 + assert response["id_token"] == TestAuthTokenCLI._MOCK_TOKEN + assert response["expiration_time"] == TestAuthTokenCLI._MOCK_EXPIRY + + @pytest.mark.integration + @staticmethod + def test_auth_token_failure_exit_code(record_property, runner: CliRunner) -> None: + """Exit code is 1 when token retrieval fails.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + with patch("aignostics.platform._cli.get_token", side_effect=RuntimeError("no credentials")): + result = runner.invoke(cli, ["auth", "token"]) + + assert result.exit_code == 1 + + @pytest.mark.integration + @staticmethod + def test_auth_token_failure_json_structure(record_property, runner: CliRunner) -> None: + """Gcloud-compatible error JSON is written to stdout when token retrieval fails.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + with patch("aignostics.platform._cli.get_token", side_effect=RuntimeError("no credentials")): + result = runner.invoke(cli, ["auth", "token"]) + + json_start = result.output.find("{") + response = json.loads(result.output[json_start:]) + + assert response["version"] == 1 + assert response["success"] is False + assert "code" in response + assert "message" in response + assert "no credentials" in response["message"] From 914422cd956cb907cdc01018770627c149361e64 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Wed, 1 Jul 2026 13:06:44 +0200 Subject: [PATCH 02/10] Update the dependencies --- pyproject.toml | 3 +++ uv.lock | 24 +++++++++++++++--------- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index dcbd087b0..11295728b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,6 +101,7 @@ dependencies = [ "httpx>=0.28.1,<1", "idc-index-data==24.0.3", "ijson>=3.4.0.post0,<4", + "joserfc>=1.6.7", "jsf>=0.11.2,<1", "jsonschema[format-nongpl]>=4.25.1,<5", "loguru>=0.7.3,<1", @@ -114,6 +115,8 @@ dependencies = [ "pyarrow>=23.0.1,<24; python_version >= '3.14'", "pyjwt[crypto]>=2.13.0,<3", # CVE-2026-32597 requires >=2.12.0 (Renovate #475) "python-dateutil>=2.9.0.post0,<3", + "python-engineio>=4.13.3", + "python-socketio>=5.16.3", # "pywebview[qt6]>=5.4,<6; sys_platform == 'linux'", "requests>=2.33.0,<3", # CVE-2026-25645 requires >= 2.33.0 "requests-oauthlib>=2.0.0,<3", diff --git a/uv.lock b/uv.lock index 2ac82b69b..385b61a04 100644 --- a/uv.lock +++ b/uv.lock @@ -54,6 +54,7 @@ dependencies = [ { name = "humanize" }, { name = "idc-index-data" }, { name = "ijson" }, + { name = "joserfc" }, { name = "jsf" }, { name = "jsonschema", extra = ["format-nongpl"] }, { name = "loguru" }, @@ -78,7 +79,9 @@ dependencies = [ { name = "pygments" }, { name = "pyjwt", extra = ["crypto"] }, { name = "python-dateutil" }, + { name = "python-engineio" }, { name = "python-multipart" }, + { name = "python-socketio" }, { name = "pywin32", marker = "sys_platform == 'win32'" }, { name = "pyyaml" }, { name = "requests" }, @@ -198,6 +201,7 @@ requires-dist = [ { name = "idc-index-data", specifier = "==24.0.3" }, { name = "ijson", specifier = ">=3.4.0.post0,<4" }, { name = "ipython", marker = "extra == 'marimo'", specifier = ">=9.8.0,<10" }, + { name = "joserfc", specifier = ">=1.6.7" }, { name = "jsf", specifier = ">=0.11.2,<1" }, { name = "jsonschema", extras = ["format-nongpl"], specifier = ">=4.25.1,<5" }, { name = "jupyter", marker = "extra == 'jupyter'", specifier = ">=1.1.1,<2" }, @@ -231,7 +235,9 @@ requires-dist = [ { name = "pyinstaller", marker = "extra == 'pyinstaller'", specifier = ">=6.14.0,<7" }, { name = "pyjwt", extras = ["crypto"], specifier = ">=2.13.0,<3" }, { name = "python-dateutil", specifier = ">=2.9.0.post0,<3" }, + { name = "python-engineio", specifier = ">=4.13.3" }, { name = "python-multipart", specifier = ">=0.0.26" }, + { name = "python-socketio", specifier = ">=5.16.3" }, { name = "pywin32", marker = "sys_platform == 'win32'", specifier = ">=312,<313" }, { name = "pyyaml", specifier = ">=6.0.3,<7" }, { name = "requests", specifier = ">=2.33.0,<3" }, @@ -3004,14 +3010,14 @@ wheels = [ [[package]] name = "joserfc" -version = "1.6.5" +version = "1.7.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "cryptography" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/3b/dc/5f768c2e391e9afabe5d18e3221346deb5fb6338565f1ccc9e7c6d7befdd/joserfc-1.6.5.tar.gz", hash = "sha256:1482a7db78fb4602e44ed89e51b599d052e091288c7c532c5b694e20149dec48", size = 231881, upload-time = "2026-05-06T04:58:13.408Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f1/26/abe1ad855eb334b5ebc9c6495d4798e12bee70e5e8e815d54570710b8312/joserfc-1.7.2.tar.gz", hash = "sha256:537ffb8888b2df039cb5b6d017d7cff6f09d521ce65d89cc9b8ab752b1cff947", size = 233183, upload-time = "2026-06-29T09:03:10.868Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/54/3b/ad1cb22e75c963b1f07c8a2329bf47227ce7e4361df5eb2fb101b2ce33ef/joserfc-1.6.5-py3-none-any.whl", hash = "sha256:e9878a0f8243fe7b95e11fdda81374ca9f7a689e302751579d3dfdeec559675e", size = 70464, upload-time = "2026-05-06T04:58:11.668Z" }, + { url = "https://files.pythonhosted.org/packages/13/80/d1b30336582cced4dce0dae776508a6011723e32f907bc7a702c0b25890a/joserfc-1.7.2-py3-none-any.whl", hash = "sha256:ddd818c0ca9b4f17bbc2d72cb3966e6ded7502be089316c62c3cc64ae86132b5", size = 70426, upload-time = "2026-06-29T09:03:09.393Z" }, ] [[package]] @@ -6230,14 +6236,14 @@ wheels = [ [[package]] name = "python-engineio" -version = "4.13.1" +version = "4.13.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "simple-websocket" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/34/12/bdef9dbeedbe2cdeba2a2056ad27b1fb081557d34b69a97f574843462cae/python_engineio-4.13.1.tar.gz", hash = "sha256:0a853fcef52f5b345425d8c2b921ac85023a04dfcf75d7b74696c61e940fd066", size = 92348, upload-time = "2026-02-06T23:38:06.12Z" } +sdist = { url = "https://files.pythonhosted.org/packages/fb/a0/f75491f942184d9960b15e763270f765fe9f239745ca5f9e16289011aed4/python_engineio-4.13.3.tar.gz", hash = "sha256:572b7783e341fed21edbc7cea297ccd378dad79265fdde96aa4664420a7c06c9", size = 79734, upload-time = "2026-06-20T22:53:52.197Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/aa/54/0cce26da03a981f949bb8449c9778537f75f5917c172e1d2992ff25cb57d/python_engineio-4.13.1-py3-none-any.whl", hash = "sha256:f32ad10589859c11053ad7d9bb3c9695cdf862113bfb0d20bc4d890198287399", size = 59847, upload-time = "2026-02-06T23:38:04.861Z" }, + { url = "https://files.pythonhosted.org/packages/5b/96/82f6328e410515fab21d5602ba35b9377a47b5a141a0c1f9efa00ce21eb4/python_engineio-4.13.3-py3-none-any.whl", hash = "sha256:1f60ecaf1358190f0e26c48c578a60428dc02a8f1295bc3dbf53d1b31116821f", size = 59993, upload-time = "2026-06-20T22:53:50.775Z" }, ] [[package]] @@ -6260,15 +6266,15 @@ wheels = [ [[package]] name = "python-socketio" -version = "5.16.1" +version = "5.16.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "bidict" }, { name = "python-engineio" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/59/81/cf8284f45e32efa18d3848ed82cdd4dcc1b657b082458fbe01ad3e1f2f8d/python_socketio-5.16.1.tar.gz", hash = "sha256:f863f98eacce81ceea2e742f6388e10ca3cdd0764be21d30d5196470edf5ea89", size = 128508, upload-time = "2026-02-06T23:42:07Z" } +sdist = { url = "https://files.pythonhosted.org/packages/32/2d/ffce71017c106b75099fea569df6518c63fee5d6202ce0cfe7b01e6f22c3/python_socketio-5.16.3.tar.gz", hash = "sha256:89b136f677ae65607a84cecda9b4d6c5377b40a97582c504c25df89af16d520e", size = 128095, upload-time = "2026-06-15T22:07:04.003Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/07/c7/deb8c5e604404dbf10a3808a858946ca3547692ff6316b698945bb72177e/python_socketio-5.16.1-py3-none-any.whl", hash = "sha256:a3eb1702e92aa2f2b5d3ba00261b61f062cce51f1cfb6900bf3ab4d1934d2d35", size = 82054, upload-time = "2026-02-06T23:42:05.772Z" }, + { url = "https://files.pythonhosted.org/packages/0a/38/8c5e72d53ff8eb27497c4f268a7f6d9121e727a50b65248288ad79a93053/python_socketio-5.16.3-py3-none-any.whl", hash = "sha256:e7ad14202a5e6448824c7c2f86161d04e13dec05992257df5c709e6a2798c041", size = 82087, upload-time = "2026-06-15T22:07:02.498Z" }, ] [package.optional-dependencies] From b444f77aa82ae35a3ff82b2875b53c42b2f09cd1 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Wed, 1 Jul 2026 14:30:15 +0200 Subject: [PATCH 03/10] test(platform): strengthen TestAuthTokenCLI by testing through get_token Replace get_token mock with a real token file on disk so that get_token's cache-hit path (file existence check, expiry parsing, 5-minute window validation) executes for real. For success tests: write token:future_expiry to tmp_path and mock settings at the authentication layer; only _inform_sentry_about_user is suppressed to avoid JWT network calls. For failure tests: point settings at a non-existent file so the cache miss occurs naturally, then mock _authenticate to raise so get_token's full failure path is exercised without real OAuth flows. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 67 ++++++++++++++++++--------- 1 file changed, 45 insertions(+), 22 deletions(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index 2874d3887..1e3e72f01 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -1,6 +1,7 @@ """Tests to verify the CLI functionality of the platform module.""" import json +from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch import pytest @@ -627,39 +628,49 @@ class TestAuthTokenCLI: """Test cases for the ``aignostics auth token`` command (gcloud WIF credential helper).""" _MOCK_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MzQ1Njc4OTB9.sig" # noqa: S105 - _MOCK_EXPIRY = 1734567890 + + @staticmethod + def _settings_with_token_file(token_file): + """Return a mock settings object pointing at the given token file path.""" + mock = MagicMock() + mock.token_file = token_file + return mock @pytest.mark.integration @staticmethod - def test_auth_token_success_exit_code(record_property, runner: CliRunner) -> None: + def test_auth_token_success_exit_code(record_property, runner: CliRunner, tmp_path) -> None: """Exit code is 0 when a cached token is available.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - stored = f"{TestAuthTokenCLI._MOCK_TOKEN}:{TestAuthTokenCLI._MOCK_EXPIRY}" - mock_settings = MagicMock() - mock_settings.return_value.token_file.read_text.return_value = stored + future_expiry = int((datetime.now(tz=UTC) + timedelta(hours=1)).timestamp()) + token_file = tmp_path / "token" + token_file.write_text(f"{TestAuthTokenCLI._MOCK_TOKEN}:{future_expiry}", encoding="utf-8") + mock_settings = TestAuthTokenCLI._settings_with_token_file(token_file) with ( - patch("aignostics.platform._cli.get_token", return_value=TestAuthTokenCLI._MOCK_TOKEN), - patch("aignostics.platform._cli.settings", mock_settings), + patch("aignostics.platform._authentication.settings", return_value=mock_settings), + patch("aignostics.platform._cli.settings", return_value=mock_settings), + patch("aignostics.platform._authentication._inform_sentry_about_user"), ): - result = runner.invoke(cli, ["auth", "token"]) + result = runner.invoke(cli, ["user", "token"]) assert result.exit_code == 0 @pytest.mark.integration @staticmethod - def test_auth_token_success_json_structure(record_property, runner: CliRunner) -> None: + def test_auth_token_success_json_structure(record_property, runner: CliRunner, tmp_path) -> None: """Valid gcloud external-credential-helper JSON is written to stdout on success.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - stored = f"{TestAuthTokenCLI._MOCK_TOKEN}:{TestAuthTokenCLI._MOCK_EXPIRY}" - mock_settings = MagicMock() - mock_settings.return_value.token_file.read_text.return_value = stored + future_expiry = int((datetime.now(tz=UTC) + timedelta(hours=1)).timestamp()) + token_file = tmp_path / "token" + token_file.write_text(f"{TestAuthTokenCLI._MOCK_TOKEN}:{future_expiry}", encoding="utf-8") + mock_settings = TestAuthTokenCLI._settings_with_token_file(token_file) with ( - patch("aignostics.platform._cli.get_token", return_value=TestAuthTokenCLI._MOCK_TOKEN), - patch("aignostics.platform._cli.settings", mock_settings), + patch("aignostics.platform._authentication.settings", return_value=mock_settings), + patch("aignostics.platform._cli.settings", return_value=mock_settings), + patch("aignostics.platform._authentication._inform_sentry_about_user"), ): - result = runner.invoke(cli, ["auth", "token"]) + result = runner.invoke(cli, ["user", "token"]) json_start = result.output.find("{") response = json.loads(result.output[json_start:]) @@ -668,25 +679,37 @@ def test_auth_token_success_json_structure(record_property, runner: CliRunner) - assert response["success"] is True assert response["token_type"] == "urn:ietf:params:oauth:token-type:id_token" # noqa: S105 assert response["id_token"] == TestAuthTokenCLI._MOCK_TOKEN - assert response["expiration_time"] == TestAuthTokenCLI._MOCK_EXPIRY + assert response["expiration_time"] == future_expiry @pytest.mark.integration @staticmethod - def test_auth_token_failure_exit_code(record_property, runner: CliRunner) -> None: + def test_auth_token_failure_exit_code(record_property, runner: CliRunner, tmp_path) -> None: """Exit code is 1 when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - with patch("aignostics.platform._cli.get_token", side_effect=RuntimeError("no credentials")): - result = runner.invoke(cli, ["auth", "token"]) + mock_settings = TestAuthTokenCLI._settings_with_token_file(tmp_path / "no_token") + + with ( + patch("aignostics.platform._authentication.settings", return_value=mock_settings), + patch("aignostics.platform._cli.settings", return_value=mock_settings), + patch("aignostics.platform._authentication._authenticate", side_effect=RuntimeError("no credentials")), + ): + result = runner.invoke(cli, ["user", "token"]) assert result.exit_code == 1 @pytest.mark.integration @staticmethod - def test_auth_token_failure_json_structure(record_property, runner: CliRunner) -> None: + def test_auth_token_failure_json_structure(record_property, runner: CliRunner, tmp_path) -> None: """Gcloud-compatible error JSON is written to stdout when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - with patch("aignostics.platform._cli.get_token", side_effect=RuntimeError("no credentials")): - result = runner.invoke(cli, ["auth", "token"]) + mock_settings = TestAuthTokenCLI._settings_with_token_file(tmp_path / "no_token") + + with ( + patch("aignostics.platform._authentication.settings", return_value=mock_settings), + patch("aignostics.platform._cli.settings", return_value=mock_settings), + patch("aignostics.platform._authentication._authenticate", side_effect=RuntimeError("no credentials")), + ): + result = runner.invoke(cli, ["user", "token"]) json_start = result.output.find("{") response = json.loads(result.output[json_start:]) From 2e4c9f17a2797b8b866e729520daa2763477ec2d Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Wed, 1 Jul 2026 14:44:02 +0200 Subject: [PATCH 04/10] fix(platform): fix lint errors in TestAuthTokenCLI tests Remove stray backslash line continuation and add missing return type annotation on _settings_with_token_file staticmethod. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index 1e3e72f01..f2e4d988a 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -630,7 +630,7 @@ class TestAuthTokenCLI: _MOCK_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MzQ1Njc4OTB9.sig" # noqa: S105 @staticmethod - def _settings_with_token_file(token_file): + def _settings_with_token_file(token_file) -> MagicMock: """Return a mock settings object pointing at the given token file path.""" mock = MagicMock() mock.token_file = token_file From fa6ec6530eba27b75c8e89fcdb84c7401159278d Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Fri, 3 Jul 2026 13:52:59 +0200 Subject: [PATCH 05/10] fix(ci): split Ketryx build report into tests and audit uploads The combined upload was timing out (500) because the CycloneDX SBOM had no root component, causing Ketryx to do excessive synchronous dependency-relation work before returning. Split into two parallel builds per Ketryx support recommendation: - ci-cd-tests: JUnit XML + coverage reports - ci-cd-audit: CycloneDX SBOM + SPDX + licenses + vulnerabilities Each upload is now smaller and independently scoped, avoiding the timeout on the SBOM parsing path. Co-Authored-By: Claude Sonnet 4.6 --- .../workflows/_ketryx_report_and_check.yml | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/.github/workflows/_ketryx_report_and_check.yml b/.github/workflows/_ketryx_report_and_check.yml index bcd8c0555..59b5697ef 100644 --- a/.github/workflows/_ketryx_report_and_check.yml +++ b/.github/workflows/_ketryx_report_and_check.yml @@ -55,7 +55,7 @@ jobs: name: audit-results path: audit-results - - name: Report build to Ketryx and check for approval + - name: Report test results to Ketryx if: | !contains(inputs.commit_message, 'skip:ketryx') && !contains(github.event.pull_request.labels.*.name, 'skip:ketryx') @@ -65,8 +65,23 @@ jobs: api-key: ${{ secrets.KETRYX_API_KEY }} commit-sha: ${{ inputs.version == '' && inputs.commit-sha || '' }} version: ${{ inputs.version }} - build-name: "ci-cd" + build-name: "ci-cd-tests" test-junit-path: test-results/junit_*.xml + artifact-path: | + test-results/coverage.xml + test-results/coverage.md + + - name: Report audit results to Ketryx + if: | + !contains(inputs.commit_message, 'skip:ketryx') && + !contains(github.event.pull_request.labels.*.name, 'skip:ketryx') + uses: Ketryx/ketryx-github-action@40b13ef68c772e96e58ec01a81f5b216d7710186 # v1.4.0 + with: + project: ${{ secrets.KETRYX_PROJECT }} + api-key: ${{ secrets.KETRYX_API_KEY }} + commit-sha: ${{ inputs.version == '' && inputs.commit-sha || '' }} + version: ${{ inputs.version }} + build-name: "ci-cd-audit" cyclonedx-json-path: | audit-results/sbom.json artifact-path: | @@ -75,5 +90,3 @@ jobs: audit-results/licenses.json audit-results/licenses_grouped.json audit-results/vulnerabilities.json - test-results/coverage.xml - test-results/coverage.md From 268477e07f7e6d6fa5302a9a8d9a0cec40d4f296 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 6 Jul 2026 11:43:15 +0200 Subject: [PATCH 06/10] fix(ci): add root component to CycloneDX SBOM Pass --pyproject pyproject.toml to cyclonedx-py so the SBOM includes aignostics as its root component. Without it, Ketryx has no dependency tree anchor and performs excessive synchronous graph work, causing the ci-cd-audit upload to time out with a 500 error. Co-Authored-By: Claude Sonnet 4.6 --- noxfile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/noxfile.py b/noxfile.py index dd69ce240..0b1a4d335 100644 --- a/noxfile.py +++ b/noxfile.py @@ -225,7 +225,7 @@ def audit(session: nox.Session) -> None: _format_json_with_jq(session, "reports/licenses_grouped.json") # SBOMs - session.run("cyclonedx-py", "environment", "-o", SBOM_CYCLONEDX_PATH) + session.run("cyclonedx-py", "environment", "--pyproject", "pyproject.toml", "-o", SBOM_CYCLONEDX_PATH) _format_json_with_jq(session, SBOM_CYCLONEDX_PATH) # Generates an SPDX SBOM including vulnerability scanning From 20db73e0216c26bdd31dd85c1cb53c0926730fe9 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 6 Jul 2026 14:26:32 +0200 Subject: [PATCH 07/10] refactor(platform): deduplicate TestAuthTokenCLI test setup Extract _invoke_success and _invoke_failure helpers that own all patching and invocation logic, so each test method only contains its single assertion. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 80 ++++++++++----------------- 1 file changed, 29 insertions(+), 51 deletions(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index f2e4d988a..cad2b76ad 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -5,6 +5,7 @@ from unittest.mock import MagicMock, patch import pytest +from click.testing import Result as InvokeResult from typer.testing import CliRunner from aignostics.cli import cli @@ -630,51 +631,47 @@ class TestAuthTokenCLI: _MOCK_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MzQ1Njc4OTB9.sig" # noqa: S105 @staticmethod - def _settings_with_token_file(token_file) -> MagicMock: - """Return a mock settings object pointing at the given token file path.""" - mock = MagicMock() - mock.token_file = token_file - return mock - - @pytest.mark.integration - @staticmethod - def test_auth_token_success_exit_code(record_property, runner: CliRunner, tmp_path) -> None: - """Exit code is 0 when a cached token is available.""" - record_property("tested-item-id", "SPEC-PLATFORM-CLI") + def _invoke_success(runner: CliRunner, tmp_path) -> tuple[InvokeResult, int]: + """Invoke ``user token`` with a valid cached token file; returns (result, expiry).""" future_expiry = int((datetime.now(tz=UTC) + timedelta(hours=1)).timestamp()) token_file = tmp_path / "token" token_file.write_text(f"{TestAuthTokenCLI._MOCK_TOKEN}:{future_expiry}", encoding="utf-8") - mock_settings = TestAuthTokenCLI._settings_with_token_file(token_file) - + mock_settings = MagicMock() + mock_settings.token_file = token_file with ( patch("aignostics.platform._authentication.settings", return_value=mock_settings), patch("aignostics.platform._cli.settings", return_value=mock_settings), patch("aignostics.platform._authentication._inform_sentry_about_user"), ): - result = runner.invoke(cli, ["user", "token"]) - - assert result.exit_code == 0 + return runner.invoke(cli, ["user", "token"]), future_expiry - @pytest.mark.integration @staticmethod - def test_auth_token_success_json_structure(record_property, runner: CliRunner, tmp_path) -> None: - """Valid gcloud external-credential-helper JSON is written to stdout on success.""" - record_property("tested-item-id", "SPEC-PLATFORM-CLI") - future_expiry = int((datetime.now(tz=UTC) + timedelta(hours=1)).timestamp()) - token_file = tmp_path / "token" - token_file.write_text(f"{TestAuthTokenCLI._MOCK_TOKEN}:{future_expiry}", encoding="utf-8") - mock_settings = TestAuthTokenCLI._settings_with_token_file(token_file) - + def _invoke_failure(runner: CliRunner, tmp_path) -> InvokeResult: + """Invoke ``user token`` with no valid token file; returns result.""" + mock_settings = MagicMock() + mock_settings.token_file = tmp_path / "no_token" with ( patch("aignostics.platform._authentication.settings", return_value=mock_settings), patch("aignostics.platform._cli.settings", return_value=mock_settings), - patch("aignostics.platform._authentication._inform_sentry_about_user"), + patch("aignostics.platform._authentication._authenticate", side_effect=RuntimeError("no credentials")), ): - result = runner.invoke(cli, ["user", "token"]) + return runner.invoke(cli, ["user", "token"]) - json_start = result.output.find("{") - response = json.loads(result.output[json_start:]) + @pytest.mark.integration + @staticmethod + def test_auth_token_success_exit_code(record_property, runner: CliRunner, tmp_path) -> None: + """Exit code is 0 when a cached token is available.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + result, _ = TestAuthTokenCLI._invoke_success(runner, tmp_path) + assert result.exit_code == 0 + @pytest.mark.integration + @staticmethod + def test_auth_token_success_json_structure(record_property, runner: CliRunner, tmp_path) -> None: + """Valid gcloud external-credential-helper JSON is written to stdout on success.""" + record_property("tested-item-id", "SPEC-PLATFORM-CLI") + result, future_expiry = TestAuthTokenCLI._invoke_success(runner, tmp_path) + response = json.loads(result.output[result.output.find("{"):]) assert response["version"] == 1 assert response["success"] is True assert response["token_type"] == "urn:ietf:params:oauth:token-type:id_token" # noqa: S105 @@ -686,34 +683,15 @@ def test_auth_token_success_json_structure(record_property, runner: CliRunner, t def test_auth_token_failure_exit_code(record_property, runner: CliRunner, tmp_path) -> None: """Exit code is 1 when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - mock_settings = TestAuthTokenCLI._settings_with_token_file(tmp_path / "no_token") - - with ( - patch("aignostics.platform._authentication.settings", return_value=mock_settings), - patch("aignostics.platform._cli.settings", return_value=mock_settings), - patch("aignostics.platform._authentication._authenticate", side_effect=RuntimeError("no credentials")), - ): - result = runner.invoke(cli, ["user", "token"]) - - assert result.exit_code == 1 + assert TestAuthTokenCLI._invoke_failure(runner, tmp_path).exit_code == 1 @pytest.mark.integration @staticmethod def test_auth_token_failure_json_structure(record_property, runner: CliRunner, tmp_path) -> None: """Gcloud-compatible error JSON is written to stdout when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") - mock_settings = TestAuthTokenCLI._settings_with_token_file(tmp_path / "no_token") - - with ( - patch("aignostics.platform._authentication.settings", return_value=mock_settings), - patch("aignostics.platform._cli.settings", return_value=mock_settings), - patch("aignostics.platform._authentication._authenticate", side_effect=RuntimeError("no credentials")), - ): - result = runner.invoke(cli, ["user", "token"]) - - json_start = result.output.find("{") - response = json.loads(result.output[json_start:]) - + result = TestAuthTokenCLI._invoke_failure(runner, tmp_path) + response = json.loads(result.output[result.output.find("{"):]) assert response["version"] == 1 assert response["success"] is False assert "code" in response From 07021c80ff31a47f914c9f63447477da1ffdcb4c Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 6 Jul 2026 14:40:27 +0200 Subject: [PATCH 08/10] refactor(platform): reduce duplication in TestPlatformCLI Extract mock_user_info pytest fixture shared by test_whoami_success and test_whoami_with_relogin_flag, and _assert_valid_json_output helper shared by both no-pretty schema tests. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 164 +++++++++----------------- 1 file changed, 56 insertions(+), 108 deletions(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index cad2b76ad..46ca144be 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -186,6 +186,45 @@ def test_user_info_from_claims_and_userinfo_no_org_name(record_property) -> None class TestPlatformCLI: """Test cases for platform CLI commands.""" + @pytest.fixture + def mock_user_info(self) -> UserInfo: # noqa: PLR6301 + """Standard mock UserInfo used across whoami tests.""" + return UserInfo( + role="admin", + token=TokenInfo( + issuer="https://test.auth0.com/", + issued_at=1609459200, + expires_at=1609462800, + scope=["openid", "profile"], + audience=["https://test-audience"], + authorized_party="test-client-id", + org_id="org456", + role="admin", + ), + user=User(id="user123"), + organization=Organization( + id="org456", + name="Test Organization", + aignostics_bucket_hmac_access_key_id="secret_access_key_id", + aignostics_bucket_hmac_secret_access_key="secret_access_key", # noqa: S106 + aignostics_bucket_name="test-bucket", + aignostics_bucket_protocol="gs", + aignostics_logfire_token="logfire_token", # noqa: S106 + aignostics_sentry_dsn="sentry_dsn", + ), + ) + + @staticmethod + def _assert_valid_json_output(output: str) -> None: + """Assert that output contains at least one valid JSON object.""" + json_start = output.find("{") + if json_start < 0: + pytest.fail("No JSON found in output") + try: + json.loads(output[json_start:]) + except json.JSONDecodeError: + pytest.fail("Output is not valid JSON") + @pytest.mark.e2e @staticmethod def test_login_out_info_e2e(record_property, runner: CliRunner) -> None: @@ -291,93 +330,31 @@ def test_login_error(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_whoami_success(record_property, runner: CliRunner) -> None: + def test_whoami_success(record_property, runner: CliRunner, mock_user_info: UserInfo) -> None: """Test successful whoami command.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") - # Create mock user info - mock_token_info = TokenInfo( - issuer="https://test.auth0.com/", - issued_at=1609459200, - expires_at=1609462800, - scope=["openid", "profile"], - audience=["https://test-audience"], - authorized_party="test-client-id", - org_id="org456", - role="admin", - ) - mock_user = User( - id="user123", - ) - mock_organization = Organization( - id="org456", - name="Test Organization", - aignostics_bucket_hmac_access_key_id="secret_access_key_id", - aignostics_bucket_hmac_secret_access_key="secret_access_key", # noqa: S106 - aignostics_bucket_name="test-bucket", - aignostics_bucket_protocol="gs", - aignostics_logfire_token="logfire_token", # noqa: S106 - aignostics_sentry_dsn="sentry_dsn", - ) - mock_user_info = UserInfo( - role="admin", - token=mock_token_info, - user=mock_user, - organization=mock_organization, - ) - with patch("aignostics.platform._service.Service.get_user_info", return_value=mock_user_info): result = runner.invoke(cli, ["user", "whoami"]) - assert result.exit_code == 0 - # Check that JSON output contains expected fields - output = normalize_output(result.output) - assert "user123" in output - assert "org456" in output - assert "Test Organization" in output - assert "admin" in output + assert result.exit_code == 0 + output = normalize_output(result.output) + assert "user123" in output + assert "org456" in output + assert "Test Organization" in output + assert "admin" in output @pytest.mark.integration @staticmethod - def test_whoami_with_relogin_flag(record_property, runner: CliRunner) -> None: + def test_whoami_with_relogin_flag(record_property, runner: CliRunner, mock_user_info: UserInfo) -> None: """Test whoami command with relogin flag.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") - mock_token_info = TokenInfo( - issuer="https://test.auth0.com/", - issued_at=1609459200, - expires_at=1609462800, - scope=["openid", "profile"], - audience=["test-audience"], - authorized_party="test-client-id", - org_id="org456", - role="admin", - ) - mock_user = User( - id="user123", - ) - mock_organization = Organization( - id="org456", - name="Test Organization", - aignostics_bucket_hmac_access_key_id="secret_access_key_id", - aignostics_bucket_hmac_secret_access_key="secret_access_key", # noqa: S106 - aignostics_bucket_name="test-bucket", - aignostics_bucket_protocol="gs", - aignostics_logfire_token="logfire_token", # noqa: S106 - aignostics_sentry_dsn="sentry_dsn", - ) - mock_user_info = UserInfo( - role="admin", - token=mock_token_info, - user=mock_user, - organization=mock_organization, - ) - with patch( "aignostics.platform._service.Service.get_user_info", return_value=mock_user_info ) as mock_get_user_info: result = runner.invoke(cli, ["user", "whoami", "--relogin"]) - assert result.exit_code == 0 - mock_get_user_info.assert_called_once_with(relogin=True) + assert result.exit_code == 0 + mock_get_user_info.assert_called_once_with(relogin=True) @pytest.mark.integration @staticmethod @@ -565,25 +542,10 @@ def test_sdk_run_metadata_schema_no_pretty(runner: CliRunner) -> None: result = runner.invoke(cli, ["sdk", "run-metadata-schema", "--no-pretty"]) assert result.exit_code == 0 - # Don't normalize output for JSON parsing - output = result.output - # Check that schema contains expected top-level properties - assert "schema_version" in output - assert "submission" in output - assert "user_agent" in output - # In non-pretty mode, output should still be valid JSON - import json - - # Try to parse the output as JSON (should not raise an error) - try: - # Find JSON in output (skip boot messages) - json_start = output.find("{") - if json_start >= 0: - json.loads(output[json_start:]) - else: - pytest.fail("No JSON found in output") - except json.JSONDecodeError: - pytest.fail("Output is not valid JSON") + assert "schema_version" in result.output + assert "submission" in result.output + assert "user_agent" in result.output + TestPlatformCLI._assert_valid_json_output(result.output) @pytest.mark.integration @staticmethod @@ -606,23 +568,9 @@ def test_sdk_item_metadata_schema_no_pretty(runner: CliRunner) -> None: result = runner.invoke(cli, ["sdk", "item-metadata-schema", "--no-pretty"]) assert result.exit_code == 0 - # Don't normalize output for JSON parsing - output = result.output - # Check that schema contains expected top-level properties - assert "schema_version" in output - assert "platform_bucket" in output - # In non-pretty mode, output should still be valid JSON - - # Try to parse the output as JSON (should not raise an error) - try: - # Find JSON in output (skip boot messages) - json_start = output.find("{") - if json_start >= 0: - json.loads(output[json_start:]) - else: - pytest.fail("No JSON found in output") - except json.JSONDecodeError: - pytest.fail("Output is not valid JSON") + assert "schema_version" in result.output + assert "platform_bucket" in result.output + TestPlatformCLI._assert_valid_json_output(result.output) class TestAuthTokenCLI: From 78adb09ef76cf8abd9c7d34feb420336e71da5d1 Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 6 Jul 2026 14:48:58 +0200 Subject: [PATCH 09/10] fix(platform): fix ruff formatting in cli_test.py Add space before colon in slice expression per ruff E203 style rule. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index 46ca144be..020ce433a 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -619,7 +619,7 @@ def test_auth_token_success_json_structure(record_property, runner: CliRunner, t """Valid gcloud external-credential-helper JSON is written to stdout on success.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") result, future_expiry = TestAuthTokenCLI._invoke_success(runner, tmp_path) - response = json.loads(result.output[result.output.find("{"):]) + response = json.loads(result.output[result.output.find("{") :]) assert response["version"] == 1 assert response["success"] is True assert response["token_type"] == "urn:ietf:params:oauth:token-type:id_token" # noqa: S105 @@ -639,7 +639,7 @@ def test_auth_token_failure_json_structure(record_property, runner: CliRunner, t """Gcloud-compatible error JSON is written to stdout when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") result = TestAuthTokenCLI._invoke_failure(runner, tmp_path) - response = json.loads(result.output[result.output.find("{"):]) + response = json.loads(result.output[result.output.find("{") :]) assert response["version"] == 1 assert response["success"] is False assert "code" in response From 4acc83646994622afd0e4ab88ec9a5749516cd8d Mon Sep 17 00:00:00 2001 From: Dzmitry Talkach Date: Mon, 6 Jul 2026 15:55:27 +0200 Subject: [PATCH 10/10] fix(platform): annotate pytest fixture parameters in cli_test.py Add RecordProperty type alias (Callable[[str, object], None]) and Path import to satisfy mypy strict no-untyped-def on all test methods that accept record_property or tmp_path fixtures. Co-Authored-By: Claude Sonnet 4.6 --- tests/aignostics/platform/cli_test.py | 63 ++++++++++++++++----------- 1 file changed, 37 insertions(+), 26 deletions(-) diff --git a/tests/aignostics/platform/cli_test.py b/tests/aignostics/platform/cli_test.py index 020ce433a..70502c3a5 100644 --- a/tests/aignostics/platform/cli_test.py +++ b/tests/aignostics/platform/cli_test.py @@ -1,7 +1,10 @@ """Tests to verify the CLI functionality of the platform module.""" import json +from collections.abc import Callable from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import TypeAlias from unittest.mock import MagicMock, patch import pytest @@ -13,13 +16,15 @@ from aignostics.platform._service import Organization, TokenInfo, User, UserInfo from tests.conftest import normalize_output +RecordProperty: TypeAlias = Callable[[str, object], None] + class TestTokenInfo: """Test cases for TokenInfo model.""" @pytest.mark.unit @staticmethod - def test_token_info_from_claims(record_property) -> None: + def test_token_info_from_claims(record_property: RecordProperty) -> None: """Test TokenInfo creation from JWT claims.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") claims = { @@ -46,7 +51,7 @@ def test_token_info_from_claims(record_property) -> None: @pytest.mark.unit @staticmethod - def test_token_info_from_claims_with_audience_list(record_property) -> None: + def test_token_info_from_claims_with_audience_list(record_property: RecordProperty) -> None: """Test TokenInfo creation from JWT claims with audience as list.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") claims = { @@ -67,7 +72,7 @@ def test_token_info_from_claims_with_audience_list(record_property) -> None: @pytest.mark.unit @staticmethod - def test_token_info_from_claims_without_role(record_property) -> None: + def test_token_info_from_claims_without_role(record_property: RecordProperty) -> None: """Test TokenInfo creation from JWT claims with role missing.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") claims = { @@ -91,7 +96,7 @@ class TestUserInfo: @pytest.mark.unit @staticmethod - def test_user_info_from_claims_and_userinfo_with_profile(record_property) -> None: + def test_user_info_from_claims_and_userinfo_with_profile(record_property: RecordProperty) -> None: """Test UserInfo creation with both claims and userinfo.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") claims = { @@ -138,7 +143,7 @@ def test_user_info_from_claims_and_userinfo_with_profile(record_property) -> Non @pytest.mark.unit @staticmethod - def test_user_info_from_claims_and_userinfo_no_org_name(record_property) -> None: + def test_user_info_from_claims_and_userinfo_no_org_name(record_property: RecordProperty) -> None: """Test UserInfo creation when org_name is not provided in claims.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") claims = { @@ -227,7 +232,7 @@ def _assert_valid_json_output(output: str) -> None: @pytest.mark.e2e @staticmethod - def test_login_out_info_e2e(record_property, runner: CliRunner) -> None: + def test_login_out_info_e2e(record_property: RecordProperty, runner: CliRunner) -> None: """Test successful logout command.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with ( @@ -252,7 +257,7 @@ def test_login_out_info_e2e(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_logout_success(record_property, runner: CliRunner) -> None: + def test_logout_success(record_property: RecordProperty, runner: CliRunner) -> None: """Test successful logout command.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.logout", return_value=True): @@ -263,7 +268,7 @@ def test_logout_success(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_logout_not_logged_in(record_property, runner: CliRunner) -> None: + def test_logout_not_logged_in(record_property: RecordProperty, runner: CliRunner) -> None: """Test logout command when not logged in.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.logout", return_value=False): @@ -274,7 +279,7 @@ def test_logout_not_logged_in(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_logout_error(record_property, runner: CliRunner) -> None: + def test_logout_error(record_property: RecordProperty, runner: CliRunner) -> None: """Test logout command when an error occurs.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.logout", side_effect=RuntimeError("Test error")): @@ -285,7 +290,7 @@ def test_logout_error(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_login_success(record_property, runner: CliRunner) -> None: + def test_login_success(record_property: RecordProperty, runner: CliRunner) -> None: """Test successful login command.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.login", return_value=True): @@ -296,7 +301,7 @@ def test_login_success(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_login_with_relogin_flag(record_property, runner: CliRunner) -> None: + def test_login_with_relogin_flag(record_property: RecordProperty, runner: CliRunner) -> None: """Test login command with relogin flag.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.login", return_value=True) as mock_login: @@ -308,7 +313,7 @@ def test_login_with_relogin_flag(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_login_failure(record_property, runner: CliRunner) -> None: + def test_login_failure(record_property: RecordProperty, runner: CliRunner) -> None: """Test login command when login fails.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.login", return_value=False): @@ -319,7 +324,7 @@ def test_login_failure(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_login_error(record_property, runner: CliRunner) -> None: + def test_login_error(record_property: RecordProperty, runner: CliRunner) -> None: """Test login command when an error occurs.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.login", side_effect=RuntimeError("Test error")): @@ -330,7 +335,7 @@ def test_login_error(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_whoami_success(record_property, runner: CliRunner, mock_user_info: UserInfo) -> None: + def test_whoami_success(record_property: RecordProperty, runner: CliRunner, mock_user_info: UserInfo) -> None: """Test successful whoami command.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.get_user_info", return_value=mock_user_info): @@ -345,7 +350,9 @@ def test_whoami_success(record_property, runner: CliRunner, mock_user_info: User @pytest.mark.integration @staticmethod - def test_whoami_with_relogin_flag(record_property, runner: CliRunner, mock_user_info: UserInfo) -> None: + def test_whoami_with_relogin_flag( + record_property: RecordProperty, runner: CliRunner, mock_user_info: UserInfo + ) -> None: """Test whoami command with relogin flag.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch( @@ -358,7 +365,7 @@ def test_whoami_with_relogin_flag(record_property, runner: CliRunner, mock_user_ @pytest.mark.integration @staticmethod - def test_whoami_not_logged_in(record_property, runner: CliRunner) -> None: + def test_whoami_not_logged_in(record_property: RecordProperty, runner: CliRunner) -> None: """Test whoami command when not logged in.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch( @@ -372,7 +379,7 @@ def test_whoami_not_logged_in(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_whoami_error(record_property, runner: CliRunner) -> None: + def test_whoami_error(record_property: RecordProperty, runner: CliRunner) -> None: """Test whoami command when an error occurs.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") with patch("aignostics.platform._service.Service.get_user_info", side_effect=RuntimeError("Test error")): @@ -383,7 +390,7 @@ def test_whoami_error(record_property, runner: CliRunner) -> None: @pytest.mark.integration @staticmethod - def test_whoami_success_with_no_org_name(record_property, runner: CliRunner) -> None: + def test_whoami_success_with_no_org_name(record_property: RecordProperty, runner: CliRunner) -> None: """Test successful whoami command when org_name is None.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") # Create mock token info @@ -430,7 +437,7 @@ def test_whoami_success_with_no_org_name(record_property, runner: CliRunner) -> @pytest.mark.integration @staticmethod - def test_whoami_masks_secrets_by_default(record_property, runner: CliRunner) -> None: + def test_whoami_masks_secrets_by_default(record_property: RecordProperty, runner: CliRunner) -> None: """Test that whoami masks secrets by default.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") mock_token_info = TokenInfo( @@ -477,7 +484,7 @@ def test_whoami_masks_secrets_by_default(record_property, runner: CliRunner) -> @pytest.mark.integration @staticmethod - def test_whoami_shows_secrets_with_no_mask_flag(record_property, runner: CliRunner) -> None: + def test_whoami_shows_secrets_with_no_mask_flag(record_property: RecordProperty, runner: CliRunner) -> None: """Test that whoami shows secrets when --no-mask-secrets flag is used.""" record_property("tested-item-id", "SPEC-PLATFORM-SERVICE") mock_token_info = TokenInfo( @@ -579,7 +586,7 @@ class TestAuthTokenCLI: _MOCK_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE3MzQ1Njc4OTB9.sig" # noqa: S105 @staticmethod - def _invoke_success(runner: CliRunner, tmp_path) -> tuple[InvokeResult, int]: + def _invoke_success(runner: CliRunner, tmp_path: Path) -> tuple[InvokeResult, int]: """Invoke ``user token`` with a valid cached token file; returns (result, expiry).""" future_expiry = int((datetime.now(tz=UTC) + timedelta(hours=1)).timestamp()) token_file = tmp_path / "token" @@ -594,7 +601,7 @@ def _invoke_success(runner: CliRunner, tmp_path) -> tuple[InvokeResult, int]: return runner.invoke(cli, ["user", "token"]), future_expiry @staticmethod - def _invoke_failure(runner: CliRunner, tmp_path) -> InvokeResult: + def _invoke_failure(runner: CliRunner, tmp_path: Path) -> InvokeResult: """Invoke ``user token`` with no valid token file; returns result.""" mock_settings = MagicMock() mock_settings.token_file = tmp_path / "no_token" @@ -607,7 +614,7 @@ def _invoke_failure(runner: CliRunner, tmp_path) -> InvokeResult: @pytest.mark.integration @staticmethod - def test_auth_token_success_exit_code(record_property, runner: CliRunner, tmp_path) -> None: + def test_auth_token_success_exit_code(record_property: RecordProperty, runner: CliRunner, tmp_path: Path) -> None: """Exit code is 0 when a cached token is available.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") result, _ = TestAuthTokenCLI._invoke_success(runner, tmp_path) @@ -615,7 +622,9 @@ def test_auth_token_success_exit_code(record_property, runner: CliRunner, tmp_pa @pytest.mark.integration @staticmethod - def test_auth_token_success_json_structure(record_property, runner: CliRunner, tmp_path) -> None: + def test_auth_token_success_json_structure( + record_property: RecordProperty, runner: CliRunner, tmp_path: Path + ) -> None: """Valid gcloud external-credential-helper JSON is written to stdout on success.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") result, future_expiry = TestAuthTokenCLI._invoke_success(runner, tmp_path) @@ -628,14 +637,16 @@ def test_auth_token_success_json_structure(record_property, runner: CliRunner, t @pytest.mark.integration @staticmethod - def test_auth_token_failure_exit_code(record_property, runner: CliRunner, tmp_path) -> None: + def test_auth_token_failure_exit_code(record_property: RecordProperty, runner: CliRunner, tmp_path: Path) -> None: """Exit code is 1 when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") assert TestAuthTokenCLI._invoke_failure(runner, tmp_path).exit_code == 1 @pytest.mark.integration @staticmethod - def test_auth_token_failure_json_structure(record_property, runner: CliRunner, tmp_path) -> None: + def test_auth_token_failure_json_structure( + record_property: RecordProperty, runner: CliRunner, tmp_path: Path + ) -> None: """Gcloud-compatible error JSON is written to stdout when token retrieval fails.""" record_property("tested-item-id", "SPEC-PLATFORM-CLI") result = TestAuthTokenCLI._invoke_failure(runner, tmp_path)