From ac3121097a68fbda1d63ccd008fc7e138a4e06b6 Mon Sep 17 00:00:00 2001 From: Chris Lo <46541035+topher-lo@users.noreply.github.com> Date: Wed, 5 Aug 2026 14:25:34 -0700 Subject: [PATCH 1/3] test(registry): drop third-party contract tests --- packages/tracecat-registry/AGENTS.md | 62 +++ packages/tracecat-registry/CLAUDE.md | 1 + tests/registry/test_ansible.py | 56 -- tests/registry/test_cloudflare_sdk.py | 215 ------- .../test_elastic_security_template.py | 107 ---- tests/registry/test_elastic_templates.py | 526 ------------------ tests/registry/test_exa_templates.py | 191 ------- tests/registry/test_freshservice.py | 433 -------------- tests/registry/test_google_api.py | 505 ----------------- tests/registry/test_google_scc_templates.py | 292 ---------- tests/registry/test_kubernetes_sdk.py | 264 --------- tests/registry/test_misp_templates.py | 306 ---------- tests/registry/test_okta_sdk.py | 314 ----------- tests/registry/test_opensearch_templates.py | 289 ---------- tests/registry/test_scanner_templates.py | 220 -------- tests/registry/test_sentinel_one_templates.py | 226 -------- tests/registry/test_slack_sdk.py | 172 ------ .../unit/test_microsoft_sentinel_templates.py | 113 ---- 18 files changed, 63 insertions(+), 4229 deletions(-) create mode 120000 packages/tracecat-registry/CLAUDE.md delete mode 100644 tests/registry/test_ansible.py delete mode 100644 tests/registry/test_cloudflare_sdk.py delete mode 100644 tests/registry/test_elastic_security_template.py delete mode 100644 tests/registry/test_elastic_templates.py delete mode 100644 tests/registry/test_exa_templates.py delete mode 100644 tests/registry/test_freshservice.py delete mode 100644 tests/registry/test_google_api.py delete mode 100644 tests/registry/test_google_scc_templates.py delete mode 100644 tests/registry/test_kubernetes_sdk.py delete mode 100644 tests/registry/test_misp_templates.py delete mode 100644 tests/registry/test_okta_sdk.py delete mode 100644 tests/registry/test_opensearch_templates.py delete mode 100644 tests/registry/test_scanner_templates.py delete mode 100644 tests/registry/test_sentinel_one_templates.py delete mode 100644 tests/registry/test_slack_sdk.py delete mode 100644 tests/unit/test_microsoft_sentinel_templates.py diff --git a/packages/tracecat-registry/AGENTS.md b/packages/tracecat-registry/AGENTS.md index d008073da7..ac64fbe7a0 100644 --- a/packages/tracecat-registry/AGENTS.md +++ b/packages/tracecat-registry/AGENTS.md @@ -2,6 +2,68 @@ Guidance for work under `packages/tracecat-registry/`, especially templates and integration wrappers. +## Third-party integrations + +Use these rules for new or materially expanded third-party integrations. Existing +integrations are compatibility references, not sources of truth for provider APIs; +do not break their public inputs or outputs without an explicitly planned migration. + +### Research and implementation choice + +- Deeply research the provider before authoring actions. Triangulate endpoint-specific + official documentation, the official OpenAPI specification when one exists, and + relevant official SDK or MCP schemas. Check authentication, scopes, API versions, + request and response shapes, pagination, errors, and asynchronous states. +- Deep-link each action to its official endpoint documentation. If primary sources + are incomplete or conflict, surface the gap during planning instead of guessing. +- Strongly prefer YAML templates that call Tracecat's core HTTP actions for REST + APIs. If research finds a maintained official Python SDK, ask the user during + planning whether to use it. If approved, add generic direct and paginated SDK UDFs + plus YAML endpoint templates over those wrappers. + +### Thin-wrapper contract + +- Keep one action close to one provider endpoint or SDK method. Use provider-native + argument names and API-native `params` and `payload` shapes rather than recreating + the provider's model, validation, or business logic. +- Do not add provider enums, duplicated argument validation, defensive state + machines, or exception translation. Let HTTP status codes or native SDK exceptions + remain authoritative. +- Narrow boundary handling is allowed for credential isolation and security, + blocking private SDK dispatch, URL and path encoding, JSON serialization, + binary or streaming values, protocol-required checks, and bounded pagination. + Small input parsing or normalization is allowed only when it makes the outbound + API call more robust; it must not become a semantic data transform. +- Declare credentials through `RegistrySecret` or OAuth rather than ordinary action + inputs. Document required scopes, API versions, and product-tier constraints. +- Encode every dynamic URL path segment with `FN.url_encode`. +- Declare REST `base_url` as `str | None` with a `null` default. Resolve it in this + order: `inputs.base_url || VARS..base_url || `. + Omit only the final fallback when the provider has no universal public endpoint. +- Return the untouched full `core.http_request` result, including `status_code`, + `headers`, and `data`. Do not select, rename, filter, or reshape provider output. + SDK wrappers may only adapt values enough to make the native result serializable. +- Pagination is the only general output-shaping exception. Follow the Slack and + boto3 wrapper patterns: preserve provider order, document whether the result is a + page list or flattened item list, and enforce the documented bound without + otherwise transforming items. +- For polling, stop when the documented transient HTTP code or body state is no + longer present. Do not test exact success equality or membership in a set of + success values. Return the raw terminal response, including provider-declared + failure states. + +### Tests + +- Do not add provider-specific white-box tests that mock an API or SDK and assert + URLs, arguments, schemas, enums, payloads, or outputs. These restate the + implementation without validating the real provider contract. +- Add live or sandbox provider tests only when a reliable environment exists and the + user explicitly chooses that coverage during planning. +- Generic registry and template validation remains required. Narrow unit tests are + allowed for Tracecat-owned security or protocol mechanics such as credential + isolation, private-method blocking, serialization limits, and reusable pagination + machinery. + ## Template design - Treat templates as thin API wrappers. Prefer passing through API-native shapes over reimplementing API validation or business logic in YAML/Python steps. diff --git a/packages/tracecat-registry/CLAUDE.md b/packages/tracecat-registry/CLAUDE.md new file mode 120000 index 0000000000..47dc3e3d86 --- /dev/null +++ b/packages/tracecat-registry/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/tests/registry/test_ansible.py b/tests/registry/test_ansible.py deleted file mode 100644 index cf68126601..0000000000 --- a/tests/registry/test_ansible.py +++ /dev/null @@ -1,56 +0,0 @@ -from __future__ import annotations - -from typing import Any - -import tracecat_registry.integrations.ansible as ansible_integration - - -class _FakeRunner: - def __init__(self) -> None: - self.stdout = "" - self.events = [] - - -def test_run_playbook_sets_quiet_by_default(monkeypatch) -> None: - captured: dict[str, Any] = {} - - monkeypatch.setattr(ansible_integration, "Runner", _FakeRunner) - monkeypatch.setattr(ansible_integration.secrets, "get", lambda _key: None) - - def fake_run(**kwargs): - captured.update(kwargs) - return _FakeRunner() - - monkeypatch.setattr(ansible_integration, "run", fake_run) - - ansible_integration.run_playbook( - playbook=[{"name": "Smoke test", "hosts": "all", "tasks": []}], - host="192.168.1.10", - host_name="test-host", - user="ubuntu", - ) - - assert captured["quiet"] is True - - -def test_run_playbook_allows_quiet_override(monkeypatch) -> None: - captured: dict[str, Any] = {} - - monkeypatch.setattr(ansible_integration, "Runner", _FakeRunner) - monkeypatch.setattr(ansible_integration.secrets, "get", lambda _key: None) - - def fake_run(**kwargs): - captured.update(kwargs) - return _FakeRunner() - - monkeypatch.setattr(ansible_integration, "run", fake_run) - - ansible_integration.run_playbook( - playbook=[{"name": "Smoke test", "hosts": "all", "tasks": []}], - host="192.168.1.10", - host_name="test-host", - user="ubuntu", - runner_kwargs={"quiet": False}, - ) - - assert captured["quiet"] is False diff --git a/tests/registry/test_cloudflare_sdk.py b/tests/registry/test_cloudflare_sdk.py deleted file mode 100644 index f55caddf53..0000000000 --- a/tests/registry/test_cloudflare_sdk.py +++ /dev/null @@ -1,215 +0,0 @@ -from types import SimpleNamespace -from typing import Any, cast - -import pydantic -import pytest -from tracecat_registry.integrations import cloudflare_sdk - - -class FakeCloudflareModel(pydantic.BaseModel): - """Cloudflare SDK responses are pydantic models; to_jsonable_python serializes them.""" - - id: str - - -class FakeCloudflarePage: - """Mimics a Cloudflare page: iterating it auto-paginates across all pages.""" - - def __init__(self, pages: list[list[Any]]) -> None: - self.pages = pages - self.pages_fetched = 0 - - def __iter__(self) -> Any: - for page in self.pages: - self.pages_fetched += 1 - yield from page - - -def test_call_method_resolves_nested_resource_and_passes_params(monkeypatch) -> None: - calls: dict[str, Any] = {} - - def list_records(**params: Any) -> dict[str, Any]: - calls["params"] = params - return {"ok": True, "result": [{"id": "record-id"}]} - - client = SimpleNamespace( - dns=SimpleNamespace(records=SimpleNamespace(list=list_records)) - ) - - def fake_cloudflare(*, api_token: str) -> SimpleNamespace: - calls["api_token"] = api_token - return client - - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr(cloudflare_sdk, "Cloudflare", fake_cloudflare) - - result = cloudflare_sdk.call_method( - resource="dns.records", - method_name="list", - params={"zone_id": "zone-id"}, - ) - - assert result == {"ok": True, "result": [{"id": "record-id"}]} - assert calls == { - "api_token": "cf-token", - "params": {"zone_id": "zone-id"}, - } - - -def test_call_method_rejects_empty_resource_segment() -> None: - with pytest.raises(ValueError, match="empty"): - cloudflare_sdk._resolve_resource( - cast(Any, SimpleNamespace(zones=SimpleNamespace())), "zones..records" - ) - - -@pytest.mark.parametrize( - ("resource", "method_name", "message"), - [ - ("zones._raw", "list", "Resource path segment"), - ("zones", "_post", "Method name"), - ("zones", "", "Method name"), - ], -) -def test_call_method_rejects_private_or_empty_names( - monkeypatch, resource: str, method_name: str, message: str -) -> None: - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr( - cloudflare_sdk, - "Cloudflare", - lambda *, api_token: SimpleNamespace( - zones=SimpleNamespace(_raw=SimpleNamespace(list=lambda: {})) - ), - ) - - with pytest.raises(ValueError, match=message): - cloudflare_sdk.call_method( - resource=resource, - method_name=method_name, - ) - - -def test_call_method_rejects_paginated_result(monkeypatch) -> None: - monkeypatch.setattr(cloudflare_sdk, "BaseSyncPage", FakeCloudflarePage) - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr( - cloudflare_sdk, - "Cloudflare", - lambda *, api_token: SimpleNamespace( - zones=SimpleNamespace( - list=lambda **_params: FakeCloudflarePage([[{"id": "1"}]]) - ) - ), - ) - - with pytest.raises(ValueError, match="call_paginated_method"): - cloudflare_sdk.call_method(resource="zones", method_name="list") - - -def test_call_paginated_method_flattens_all_page_items(monkeypatch) -> None: - calls: dict[str, Any] = {} - page = FakeCloudflarePage( - [ - [FakeCloudflareModel(id="one")], - [FakeCloudflareModel(id="two"), {"id": "three"}], - ] - ) - - def list_zones(**params: Any) -> FakeCloudflarePage: - calls["params"] = params - return page - - monkeypatch.setattr(cloudflare_sdk, "BaseSyncPage", FakeCloudflarePage) - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr( - cloudflare_sdk, - "Cloudflare", - lambda *, api_token: SimpleNamespace(zones=SimpleNamespace(list=list_zones)), - ) - - result = cloudflare_sdk.call_paginated_method( - resource="zones", - method_name="list", - params={"account_id": "account-id"}, - ) - - assert result == [{"id": "one"}, {"id": "two"}, {"id": "three"}] - assert calls == {"params": {"account_id": "account-id"}} - - -def test_call_paginated_method_stops_at_limit(monkeypatch) -> None: - page = FakeCloudflarePage( - [ - [FakeCloudflareModel(id="one"), FakeCloudflareModel(id="two")], - [FakeCloudflareModel(id="three"), FakeCloudflareModel(id="four")], - [FakeCloudflareModel(id="five")], - ] - ) - - monkeypatch.setattr(cloudflare_sdk, "BaseSyncPage", FakeCloudflarePage) - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr( - cloudflare_sdk, - "Cloudflare", - lambda *, api_token: SimpleNamespace( - zones=SimpleNamespace(list=lambda **_params: page) - ), - ) - - result = cloudflare_sdk.call_paginated_method( - resource="zones", - method_name="list", - limit=3, - ) - - assert result == [{"id": "one"}, {"id": "two"}, {"id": "three"}] - # Iteration must stop at the limit instead of fetching remaining pages. - assert page.pages_fetched == 2 - - -def test_call_paginated_method_limit_above_total_returns_all(monkeypatch) -> None: - page = FakeCloudflarePage([[FakeCloudflareModel(id="one")], [{"id": "two"}]]) - - monkeypatch.setattr(cloudflare_sdk, "BaseSyncPage", FakeCloudflarePage) - monkeypatch.setattr( - cloudflare_sdk.secrets, - "get", - lambda key: "cf-token" if key == "CLOUDFLARE_API_TOKEN" else None, - ) - monkeypatch.setattr( - cloudflare_sdk, - "Cloudflare", - lambda *, api_token: SimpleNamespace( - zones=SimpleNamespace(list=lambda **_params: page) - ), - ) - - result = cloudflare_sdk.call_paginated_method( - resource="zones", - method_name="list", - limit=10, - ) - - assert result == [{"id": "one"}, {"id": "two"}] - assert page.pages_fetched == 2 diff --git a/tests/registry/test_elastic_security_template.py b/tests/registry/test_elastic_security_template.py deleted file mode 100644 index deabe08cd9..0000000000 --- a/tests/registry/test_elastic_security_template.py +++ /dev/null @@ -1,107 +0,0 @@ -"""Regression tests for the Elastic Security `list_detection_signals` template. - -The template assembles the request body via an inline Python step so optional -inputs (notably `_source`) are omitted from the payload when not provided. -These tests pin that contract down without spinning up the executor. -""" - -from __future__ import annotations - -from pathlib import Path -from typing import Any - -import pytest - -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATE_PATH = ( - Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/elastic_security" - ) - / "list_detection_signals.yml" -) - -START = "2026-05-01T00:00:00Z" -END = "2026-05-02T00:00:00Z" -QUERY: dict[str, Any] = {"bool": {"must": [{"match_all": {}}]}} - - -@pytest.fixture(scope="module") -def template() -> TemplateAction: - return TemplateAction.from_yaml(TEMPLATE_PATH) - - -@pytest.fixture(scope="module") -def build_payload(template: TemplateAction): - step = next(s for s in template.definition.steps if s.ref == "build_search_payload") - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"] - - -def test_expects_declares_optional_source_fields(template: TemplateAction) -> None: - expects = template.definition.expects - assert "source_fields" in expects, "source_fields input must be declared" - field = expects["source_fields"] - assert field.default is None - # The type string is what surfaces in the UI/schema — keep it strict. - assert field.type == "list[str] | dict[str, Any] | None" - - -def test_legacy_action_metadata_points_to_replacement(template: TemplateAction) -> None: - definition = template.definition - replacement = "tools.elastic_security.search_detection_alerts" - assert definition.name == "list_detection_signals" - assert definition.action == "tools.elastic_security.list_detection_signals" - assert definition.title == "(Deprecated) List detection alerts" - assert replacement in definition.description - assert definition.deprecated is not None - assert replacement in definition.deprecated - - -def test_legacy_runtime_contract_is_unchanged(template: TemplateAction) -> None: - assert set(template.definition.expects) == { - "start_time", - "end_time", - "query", - "limit", - "source_fields", - "base_url", - "verify_ssl", - } - assert [step.ref for step in template.definition.steps] == [ - "search_query", - "build_search_payload", - "query_detection_alerts", - ] - request = template.definition.steps[-1] - assert request.args["url"].endswith("/api/detection_engine/signals/search") - assert ( - template.definition.returns == "${{ steps.query_detection_alerts.result.data }}" - ) - - -def test_payload_omits_source_when_unset(build_payload) -> None: - payload = build_payload(START, END, QUERY, 100, None) - assert "_source" not in payload - assert payload == {"start": START, "end": END, "query": QUERY, "size": 100} - - -def test_payload_includes_source_list(build_payload) -> None: - fields = ["@timestamp", "kibana.alert.uuid", "kibana.alert.severity"] - payload = build_payload(START, END, QUERY, 50, fields) - assert payload["_source"] == fields - assert payload["size"] == 50 - - -def test_payload_includes_source_dict_with_includes_excludes(build_payload) -> None: - spec = {"includes": ["kibana.alert.*"], "excludes": ["kibana.alert.rule.note"]} - payload = build_payload(START, END, QUERY, 25, spec) - assert payload["_source"] == spec - - -def test_payload_preserves_empty_source_list(build_payload) -> None: - # An explicit empty list is a valid Elastic instruction to return no source - # fields. It must not be silently dropped. - payload = build_payload(START, END, QUERY, 10, []) - assert payload["_source"] == [] diff --git a/tests/registry/test_elastic_templates.py b/tests/registry/test_elastic_templates.py deleted file mode 100644 index 439d305e3b..0000000000 --- a/tests/registry/test_elastic_templates.py +++ /dev/null @@ -1,526 +0,0 @@ -"""Catalog and contract tests for security-focused Elastic actions.""" - -from __future__ import annotations - -import re -from dataclasses import dataclass -from pathlib import Path -from typing import Any - -import pytest - -from tracecat.expressions.functions import url_encode -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATE_ROOT = Path("packages/tracecat-registry/tracecat_registry/templates/tools") - - -@dataclass(frozen=True) -class CatalogEntry: - namespace: str - name: str - method: str - endpoint_path: str - display_group: str - doc_url: str - - @property - def action(self) -> str: - return f"{self.namespace}.{self.name}" - - -def elasticsearch_entry( - name: str, method: str, path: str, operation: str -) -> CatalogEntry: - suffix = f"operation/operation-{operation}" - return CatalogEntry( - namespace="tools.elasticsearch", - name=name, - method=method, - endpoint_path=path, - display_group="Elasticsearch", - doc_url=f"https://www.elastic.co/docs/api/doc/elasticsearch/{suffix}", - ) - - -def security_entry( - name: str, - method: str, - path: str, - operation: str, - *, - version: str | None = None, -) -> CatalogEntry: - version_path = f"{version}/" if version else "" - return CatalogEntry( - namespace="tools.elastic_security", - name=name, - method=method, - endpoint_path=path, - display_group="Elastic Security", - doc_url=( - "https://www.elastic.co/docs/api/doc/kibana/" - f"{version_path}operation/operation-{operation}" - ), - ) - - -CATALOG = ( - # Threat hunting and event retrieval (7) - elasticsearch_entry("list_indexes", "GET", "/_cat/indices/{index}", "cat-indices"), - elasticsearch_entry("search_events", "POST", "/{index}/_search", "search"), - elasticsearch_entry("get_document", "GET", "/{index}/_doc/{id}", "get"), - elasticsearch_entry( - "get_mapping", "GET", "/{index}/_mapping", "indices-get-mapping" - ), - elasticsearch_entry("esql", "POST", "/_query", "esql-query"), - elasticsearch_entry("eql", "POST", "/{index}/_eql/search", "eql-search"), - # Alerts and detection engineering (12) - security_entry( - "list_detection_signals", - "POST", - "/api/detection_engine/signals/search", - "searchalerts", - version="v8", - ), - security_entry( - "search_detection_alerts", - "POST", - "/api/detection_engine/signals/search", - "searchalerts", - ), - security_entry( - "set_detection_alert_status", - "POST", - "/api/detection_engine/signals/status", - "setalertsstatus", - ), - security_entry( - "update_detection_alert_tags", - "POST", - "/api/detection_engine/signals/tags", - "setalerttags", - ), - security_entry( - "assign_detection_alert_users", - "POST", - "/api/detection_engine/signals/assignees", - "setalertassignees", - ), - security_entry( - "list_detection_rules", "GET", "/api/detection_engine/rules/_find", "findrules" - ), - security_entry( - "create_detection_rule", "POST", "/api/detection_engine/rules", "createrule" - ), - security_entry( - "patch_detection_rule", "PATCH", "/api/detection_engine/rules", "patchrule" - ), - security_entry( - "bulk_action_detection_rules", - "POST", - "/api/detection_engine/rules/_bulk_action", - "performrulesbulkaction", - ), - security_entry( - "import_detection_rules", - "POST", - "/api/detection_engine/rules/_import", - "importrules", - ), - security_entry( - "export_detection_rules", - "POST", - "/api/detection_engine/rules/_export", - "exportrules", - ), - security_entry( - "preview_detection_rule", - "POST", - "/api/detection_engine/rules/preview", - "rulepreview", - ), - # AI-assisted and live investigation (3) - security_entry( - "search_attack_discoveries", - "GET", - "/api/attack_discovery/_find", - "attackdiscoveryfind", - ), - security_entry( - "search_entities", - "GET", - "/api/security/entity_store/entities", - "get-security-entity-store-entities", - ), - security_entry( - "run_osquery_live_query", - "POST", - "/api/osquery/live_queries", - "osquerycreatelivequery", - ), - # Exception-list containers and their conditions (5) - security_entry( - "create_exception_list", "POST", "/api/exception_lists", "createexceptionlist" - ), - security_entry( - "list_exception_lists", - "GET", - "/api/exception_lists/_find", - "findexceptionlists", - ), - security_entry( - "create_exception_list_item", - "POST", - "/api/exception_lists/items", - "createexceptionlistitem", - ), - security_entry( - "list_exception_list_items", - "GET", - "/api/exception_lists/items/_find", - "findexceptionlistitems", - ), - security_entry( - "delete_exception_list_item", - "DELETE", - "/api/exception_lists/items", - "deleteexceptionlistitem", - ), - # Endpoint response (20) - security_entry( - "list_endpoints", "GET", "/api/endpoint/metadata", "getendpointmetadatalist" - ), - security_entry( - "get_endpoint", "GET", "/api/endpoint/metadata/{id}", "getendpointmetadata" - ), - security_entry( - "list_response_actions", "GET", "/api/endpoint/action", "endpointgetactionslist" - ), - security_entry( - "get_response_action", - "GET", - "/api/endpoint/action/{action_id}", - "endpointgetactionsdetails", - ), - security_entry( - "cancel_response_action", "POST", "/api/endpoint/action/cancel", "cancelaction" - ), - security_entry( - "isolate_endpoint", - "POST", - "/api/endpoint/action/isolate", - "endpointisolateaction", - ), - security_entry( - "release_endpoint", - "POST", - "/api/endpoint/action/unisolate", - "endpointunisolateaction", - ), - security_entry( - "run_endpoint_command", - "POST", - "/api/endpoint/action/execute", - "endpointexecuteaction", - ), - security_entry( - "get_endpoint_processes", - "POST", - "/api/endpoint/action/running_procs", - "endpointgetprocessesaction", - ), - security_entry( - "terminate_endpoint_process", - "POST", - "/api/endpoint/action/kill_process", - "endpointkillprocessaction", - ), - security_entry( - "suspend_endpoint_process", - "POST", - "/api/endpoint/action/suspend_process", - "endpointsuspendprocessaction", - ), - security_entry( - "scan_endpoint_path", "POST", "/api/endpoint/action/scan", "endpointscanaction" - ), - security_entry( - "get_endpoint_file", - "POST", - "/api/endpoint/action/get_file", - "endpointgetfileaction", - ), - security_entry( - "get_response_file_info", - "GET", - "/api/endpoint/action/{action_id}/file/{file_id}", - "endpointfileinfo", - ), - security_entry( - "download_response_file", - "GET", - "/api/endpoint/action/{action_id}/file/{file_id}/download", - "endpointfiledownload", - ), - security_entry( - "upload_endpoint_file", - "POST", - "/api/endpoint/action/upload", - "endpointuploadaction", - ), - security_entry( - "generate_endpoint_memory_dump", - "POST", - "/api/endpoint/action/memory_dump", - "endpointgeneratememorydump", - ), - security_entry( - "run_endpoint_script", - "POST", - "/api/endpoint/action/run_script", - "runscriptaction", - ), - security_entry( - "list_endpoint_scripts", - "GET", - "/api/endpoint/scripts_library", - "endpointscriptlibrarylistscripts", - ), - security_entry( - "create_endpoint_script", - "POST", - "/api/endpoint/scripts_library", - "endpointscriptlibrarycreatescript", - ), -) - - -def load_catalog() -> dict[str, tuple[TemplateAction, Path]]: - loaded: dict[str, tuple[TemplateAction, Path]] = {} - for integration in ("elasticsearch", "elastic_security"): - for path in sorted((TEMPLATE_ROOT / integration).rglob("*.yml")): - template = TemplateAction.from_yaml(path) - loaded[template.definition.action] = (template, path) - return loaded - - -@pytest.fixture(scope="module") -def loaded_catalog() -> dict[str, tuple[TemplateAction, Path]]: - return load_catalog() - - -def http_step(template: TemplateAction): - return next( - step for step in template.definition.steps if step.action == "core.http_request" - ) - - -def execute_script(template: TemplateAction, ref: str): - step = next(step for step in template.definition.steps if step.ref == ref) - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"] - - -def test_catalog_is_exact_and_loads_by_fully_qualified_name( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - expected = {entry.action for entry in CATALOG} - assert len(CATALOG) == 46 - assert len(expected) == 46 - assert sum(entry.namespace == "tools.elasticsearch" for entry in CATALOG) == 6 - assert sum(entry.namespace == "tools.elastic_security" for entry in CATALOG) == 40 - assert set(loaded_catalog) == expected - for action_name, (template, _) in loaded_catalog.items(): - assert template.definition.action == action_name - - -@pytest.mark.parametrize("entry", CATALOG, ids=lambda entry: entry.action) -def test_catalog_metadata_method_and_path( - entry: CatalogEntry, - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, path = loaded_catalog[entry.action] - definition = template.definition - assert definition.namespace == entry.namespace - assert definition.name == entry.name - assert definition.display_group == entry.display_group - assert str(definition.doc_url) == entry.doc_url - - step = http_step(template) - assert step.args["method"] == entry.method - - source = path.read_text() - for fragment in re.split(r"\{[^}]+\}", entry.endpoint_path): - normalized_fragment = fragment.strip("/") - if len(normalized_fragment) >= 2: - assert normalized_fragment in source - - -def test_common_input_contracts( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - for action_name, (template, _) in loaded_catalog.items(): - expects = template.definition.expects - assert expects["base_url"].type == "str | None", action_name - assert expects["verify_ssl"].type == "bool", action_name - assert expects["verify_ssl"].default is True, action_name - if ( - not action_name.startswith("tools.elasticsearch.") - and action_name != "tools.elastic_security.list_detection_signals" - ): - assert expects["space_id"].type == "str | None", action_name - assert expects["space_id"].default is None, action_name - - -def test_path_identifiers_are_url_encoded( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - assert url_encode("document/with space") == "document%2Fwith%20space" - for action in ( - "tools.elasticsearch.get_document", - "tools.elastic_security.get_response_action", - ): - _, path = loaded_catalog[action] - assert "FN.url_encode" in path.read_text() - - template, _ = loaded_catalog["tools.elasticsearch.eql"] - build_path = execute_script(template, "build_path") - assert build_path("logs-*/events") == "/logs-*%2Fevents/_eql/search" - - -def test_default_and_named_space_paths( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog["tools.elastic_security.search_detection_alerts"] - build_path = execute_script(template, "build_path") - endpoint = "/api/detection_engine/signals/search" - assert build_path(endpoint, None) == endpoint - assert build_path(endpoint, "blue team") == f"/s/blue%20team{endpoint}" - - -@pytest.mark.parametrize( - ("action", "endpoint"), - ( - ( - "tools.elastic_security.search_attack_discoveries", - "/api/attack_discovery/_find", - ), - ( - "tools.elastic_security.search_entities", - "/api/security/entity_store/entities", - ), - ( - "tools.elastic_security.run_osquery_live_query", - "/api/osquery/live_queries", - ), - ), -) -def test_investigation_actions_support_default_and_named_spaces( - action: str, - endpoint: str, - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog[action] - build_path = execute_script(template, "build_path") - assert build_path(None) == endpoint - assert build_path("blue team") == f"/s/blue%20team{endpoint}" - - -@pytest.mark.parametrize( - "action", - ( - "tools.elasticsearch.eql", - "tools.elasticsearch.esql", - "tools.elastic_security.run_osquery_live_query", - "tools.elastic_security.search_detection_alerts", - ), -) -def test_native_payload_and_optional_params_pass_through( - action: str, - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog[action] - expects = template.definition.expects - assert expects["payload"].type == "dict[str, Any]" - assert expects["params"].default is None - request = http_step(template) - assert request.args["payload"] == "${{ inputs.payload }}" - assert request.args["params"] == "${{ inputs.params }}" - - -@pytest.mark.parametrize( - "action", - ( - "tools.elastic_security.search_attack_discoveries", - "tools.elastic_security.search_entities", - ), -) -def test_investigation_search_params_pass_through( - action: str, - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog[action] - assert template.definition.expects["params"].default is None - assert http_step(template).args["params"] == "${{ inputs.params }}" - - -def test_exception_container_and_item_list_actions_are_distinct( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - container, _ = loaded_catalog["tools.elastic_security.list_exception_lists"] - item, _ = loaded_catalog["tools.elastic_security.list_exception_list_items"] - assert "containers and metadata" in container.definition.description - assert "within" in item.definition.description - assert http_step(container).args["url"] == http_step(item).args["url"] - container_path = execute_script(container, "build_path")(None) - item_path = execute_script(item, "build_path")(None) - assert container_path != item_path - - -def test_import_export_and_binary_contracts( - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog["tools.elastic_security.import_detection_rules"] - request = http_step(template) - assert ( - request.args["files"]["file"]["content_base64"] - == "${{ inputs.base64_content }}" - ) - - template, _ = loaded_catalog["tools.elastic_security.export_detection_rules"] - assert template.definition.returns == "${{ steps.request.result.data }}" - - template, _ = loaded_catalog["tools.elastic_security.download_response_file"] - assert http_step(template).args["base64_encode_data"] is True - - -@pytest.mark.parametrize( - ("action", "array_field"), - ( - ("tools.elastic_security.upload_endpoint_file", "endpoint_ids"), - ("tools.elastic_security.create_endpoint_script", "platform"), - ), -) -def test_multipart_payload_assembly_is_mechanical( - action: str, - array_field: str, - loaded_catalog: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = loaded_catalog[action] - build_form_data = execute_script(template, "build_form_data") - result = build_form_data( - { - array_field: ["first", "second"], - "parameters": {"overwrite": False}, - "requiresInput": False, - "name": "Collect host data", - } - ) - assert result == { - array_field: ["first", "second"], - "parameters": '{"overwrite":false}', - "requiresInput": "false", - "name": "Collect host data", - } diff --git a/tests/registry/test_exa_templates.py b/tests/registry/test_exa_templates.py deleted file mode 100644 index 85af9edee4..0000000000 --- a/tests/registry/test_exa_templates.py +++ /dev/null @@ -1,191 +0,0 @@ -import pathlib - -import yaml - -ROOT = pathlib.Path(__file__).resolve().parents[2] -EXA_ROOT = ( - ROOT - / "packages" - / "tracecat-registry" - / "tracecat_registry" - / "templates" - / "tools" - / "exa" -) -EXPECTED_SECRET = [{"name": "exa", "keys": ["EXA_API_KEY"]}] -EXA_SEARCH_TYPE = ( - 'enum["instant", "fast", "auto", "deep-lite", "deep", "deep-reasoning"] | None' -) -EXA_COMPLIANCE = 'enum["hipaa"] | None' - - -def load_template(filename: str) -> dict: - with (EXA_ROOT / filename).open() as handle: - return yaml.safe_load(handle) - - -def request_step(template: dict) -> dict: - return next( - step - for step in template["definition"]["steps"] - if step["action"] == "core.http_request" - ) - - -def test_exa_template_surface_matches_current_core_tools(): - templates = {path.name: load_template(path.name) for path in EXA_ROOT.glob("*.yml")} - - assert set(templates) == { - "search.yml", - "search_news.yml", - "search_people.yml", - "search_companies.yml", - "get_contents.yml", - "answer.yml", - "deep_research.yml", - } - - for template in templates.values(): - definition = template["definition"] - assert definition["namespace"] == "tools.exa" - assert definition["display_group"] == "Exa" - assert definition["secrets"] == EXPECTED_SECRET - assert definition["returns"].endswith(".result.data }}") - assert [step["action"] for step in definition["steps"]] == ["core.http_request"] - - -def test_exa_templates_use_current_endpoints_and_auth(): - expected = { - "search.yml": ("search", "https://api.exa.ai/search"), - "search_news.yml": ("search_news", "https://api.exa.ai/search"), - "search_people.yml": ("search_people", "https://api.exa.ai/search"), - "search_companies.yml": ("search_companies", "https://api.exa.ai/search"), - "get_contents.yml": ("get_contents", "https://api.exa.ai/contents"), - "answer.yml": ("answer", "https://api.exa.ai/answer"), - "deep_research.yml": ("deep_research", "https://api.exa.ai/search"), - } - - for filename, (name, url) in expected.items(): - template = load_template(filename) - definition = template["definition"] - request = request_step(template) - - assert definition["name"] == name - assert request["args"]["url"] == url - assert request["args"]["method"] == "POST" - assert ( - request["args"]["headers"]["x-api-key"] == "${{ SECRETS.exa.EXA_API_KEY }}" - ) - - -def test_exa_search_exposes_advanced_current_parameters(): - search = load_template("search.yml") - expects = search["definition"]["expects"] - payload = request_step(search)["args"]["payload"] - - for field in [ - "type", - "numResults", - "category", - "includeDomains", - "excludeDomains", - "startPublishedDate", - "endPublishedDate", - "moderation", - "contents", - "additionalQueries", - "systemPrompt", - "outputSchema", - "userLocation", - "compliance", - ]: - assert field in expects - assert payload[field] == f"${{{{ inputs.{field} }}}}" - assert expects["type"]["type"] == EXA_SEARCH_TYPE - assert expects["category"]["type"] == "str | None" - assert expects["compliance"]["type"] == EXA_COMPLIANCE - - -def test_exa_vertical_search_wrappers_set_supported_categories(): - expected = { - "search_people.yml": "people", - "search_companies.yml": "company", - } - - for filename, category in expected.items(): - template = load_template(filename) - expects = template["definition"]["expects"] - payload = request_step(template)["args"]["payload"] - - assert payload["category"] == category - for supported in ["query", "type", "numResults", "contents", "outputSchema"]: - assert supported in expects - for unsupported in [ - "includeDomains", - "excludeDomains", - "startPublishedDate", - "endPublishedDate", - ]: - assert unsupported not in expects - assert unsupported not in payload - assert expects["type"]["type"] == EXA_SEARCH_TYPE - - -def test_exa_news_search_wrapper_sets_news_category_and_filters(): - template = load_template("search_news.yml") - expects = template["definition"]["expects"] - payload = request_step(template)["args"]["payload"] - - assert payload["category"] == "news" - for supported in [ - "query", - "type", - "numResults", - "includeDomains", - "excludeDomains", - "startPublishedDate", - "endPublishedDate", - "contents", - "outputSchema", - "compliance", - ]: - assert supported in expects - assert "category" not in expects - assert expects["type"]["type"] == EXA_SEARCH_TYPE - assert expects["compliance"]["type"] == EXA_COMPLIANCE - - -def test_exa_contents_uses_top_level_content_options(): - get_contents = load_template("get_contents.yml") - expects = get_contents["definition"]["expects"] - payload = request_step(get_contents)["args"]["payload"] - - assert expects["text"]["default"] is True - assert payload["urls"] == "${{ inputs.urls }}" - for field in [ - "text", - "highlights", - "summary", - "maxAgeHours", - "livecrawlTimeout", - "subpages", - "subpageTarget", - "extras", - "compliance", - ]: - assert payload[field] == f"${{{{ inputs.{field} }}}}" - assert "contents" not in payload - assert expects["compliance"]["type"] == EXA_COMPLIANCE - - -def test_exa_deep_research_replaces_deprecated_research_api(): - combined = "\n".join(path.read_text() for path in EXA_ROOT.glob("*.yml")) - deep_research = load_template("deep_research.yml") - payload = request_step(deep_research)["args"]["payload"] - - assert "/research/v1" not in combined - assert payload["type"] == "deep-reasoning" - assert ( - deep_research["definition"]["expects"]["compliance"]["type"] == EXA_COMPLIANCE - ) - assert request_step(deep_research)["args"]["url"] == "https://api.exa.ai/search" diff --git a/tests/registry/test_freshservice.py b/tests/registry/test_freshservice.py deleted file mode 100644 index 5670badd87..0000000000 --- a/tests/registry/test_freshservice.py +++ /dev/null @@ -1,433 +0,0 @@ -from base64 import b64encode -from collections.abc import Callable -from typing import Any - -import httpx -import pytest -import respx -from tracecat_registry import SecretNotFoundError -from tracecat_registry.integrations import freshservice - - -def _secret_getter(values: dict[str, str]) -> Any: - return lambda key, default=None: values.get(key, default) - - -def test_freshservice_secret_form() -> None: - assert freshservice.freshservice_secret.name == "freshservice" - assert freshservice.freshservice_secret.keys == ["FRESHSERVICE_API_KEY"] - assert freshservice.freshservice_secret.optional_keys == ["FRESHSERVICE_BASE_URL"] - - -def test_resolve_base_url_adds_scheme_and_api_path() -> None: - assert ( - freshservice._resolve_base_url("example.freshservice.com") - == "https://example.freshservice.com/api/v2" - ) - assert ( - freshservice._resolve_base_url("https://example.freshservice.com/") - == "https://example.freshservice.com/api/v2" - ) - assert ( - freshservice._resolve_base_url("https://example.freshservice.com/api/v2/") - == "https://example.freshservice.com/api/v2" - ) - - -def test_resolve_base_url_uses_secret( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - freshservice.secrets, - "get_or_default", - _secret_getter({"FRESHSERVICE_BASE_URL": "example.freshservice.com"}), - ) - - assert freshservice._resolve_base_url(None) == ( - "https://example.freshservice.com/api/v2" - ) - - -def test_resolve_base_url_requires_url( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr(freshservice.secrets, "get_or_default", _secret_getter({})) - - with pytest.raises(SecretNotFoundError, match="Freshservice calls require"): - freshservice._resolve_base_url(None) - - -def test_normalize_path_rejects_absolute_urls() -> None: - with pytest.raises(ValueError, match="must be relative"): - freshservice._normalize_path("https://example.freshservice.com/api/v2/tickets") - - -@pytest.mark.anyio -@respx.mock -async def test_call_endpoint_uses_basic_auth( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr(freshservice.secrets, "get", lambda key: "api-key") - monkeypatch.setattr(freshservice.secrets, "get_or_default", _secret_getter({})) - route = respx.post( - "https://example.freshservice.com/api/v2/tickets", - params={"source": "tracecat"}, - ).mock(return_value=httpx.Response(201, json={"ticket": {"id": 123}})) - - result = await freshservice.call_endpoint( - method="POST", - path="/api/v2/tickets", - query_params={"source": "tracecat"}, - json_body={"subject": "Test"}, - base_url="example.freshservice.com", - ) - - assert result == {"ticket": {"id": 123}} - assert route.called - request = route.calls[0].request - assert request.headers["authorization"] == ( - "Basic " + b64encode(b"api-key:X").decode() - ) - assert request.headers["accept"] == "application/json" - assert request.headers["content-type"] == "application/json" - - -@pytest.mark.anyio -async def test_list_tickets_paginates( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - - async def fake_request( - **kwargs: Any, - ) -> tuple[dict[str, Any], httpx.Headers]: - calls.append(kwargs) - page = kwargs["query_params"]["page"] - if page == 2: - return {"tickets": [{"id": 3}]}, httpx.Headers({}) - return {"tickets": [{"id": 1}, {"id": 2}]}, httpx.Headers( - { - "link": ( - "; rel="next"' - ) - } - ) - - monkeypatch.setattr(freshservice, "_request_freshservice_http", fake_request) - - result = await freshservice.list_tickets( - query_params={"filter": "new_and_my_open"}, - page=1, - per_page=2, - base_url="https://example.freshservice.com", - ) - - assert result == { - "items": [{"id": 1}, {"id": 2}, {"id": 3}], - "pages": 2, - "next_page": None, - } - assert calls[0]["query_params"] == { - "filter": "new_and_my_open", - "page": 1, - "per_page": 2, - } - assert calls[1]["query_params"] == { - "filter": "new_and_my_open", - "page": 2, - "per_page": 2, - } - - -def test_extract_next_page_follows_link_header() -> None: - headers = httpx.Headers( - { - "link": ( - "; rel="next"' - ) - } - ) - - assert freshservice._extract_next_page(headers, fallback_page=2) == 3 - assert freshservice._extract_next_page(httpx.Headers({}), fallback_page=2) is None - - -def test_extract_page_items_infers_single_list_field() -> None: - assert freshservice._extract_page_items( - {"tickets": [{"id": 1}], "meta": {"page": 1}}, items_key=None - ) == [{"id": 1}] - - -def test_extract_page_items_requires_items_key_for_ambiguous_response() -> None: - with pytest.raises(ValueError, match="did not contain a list field"): - freshservice._extract_page_items({"meta": {"page": 1}}, items_key=None) - with pytest.raises(ValueError, match="multiple list fields"): - freshservice._extract_page_items( - {"tickets": [], "requesters": []}, items_key=None - ) - - -def test_extract_page_items_requires_present_list_items_key() -> None: - with pytest.raises(ValueError, match="field `requesters` is missing"): - freshservice._extract_page_items({"tickets": []}, items_key="requesters") - with pytest.raises(ValueError, match="field `tickets` is not a list"): - freshservice._extract_page_items({"tickets": {}}, items_key="tickets") - - -@pytest.mark.anyio -async def test_create_ticket_calls_tickets_endpoint( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - - async def fake_request(**kwargs: Any) -> dict[str, Any]: - calls.append(kwargs) - return {"ticket": {"id": 123}} - - monkeypatch.setattr(freshservice, "_request_freshservice_api", fake_request) - - result = await freshservice.create_ticket( - ticket={"subject": "Test", "priority": 1, "status": 2}, - base_url="https://example.freshservice.com", - ) - - assert result == {"ticket": {"id": 123}} - assert calls == [ - { - "method": "POST", - "path": "/tickets", - "json_body": {"subject": "Test", "priority": 1, "status": 2}, - "base_url": "https://example.freshservice.com", - } - ] - - -@pytest.mark.anyio -@pytest.mark.parametrize( - ("call", "expected"), - [ - ( - lambda: freshservice.get_ticket( - ticket_id=123, - include="conversations", - base_url="https://example.freshservice.com", - ), - { - "method": "GET", - "path": "/tickets/123", - "query_params": {"include": "conversations"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.update_ticket( - ticket_id=123, - ticket={"priority": 2}, - base_url="https://example.freshservice.com", - ), - { - "method": "PUT", - "path": "/tickets/123", - "json_body": {"priority": 2}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.delete_ticket( - ticket_id=123, - base_url="https://example.freshservice.com", - ), - { - "method": "DELETE", - "path": "/tickets/123", - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.create_ticket_note( - ticket_id=123, - note={"body": "note"}, - base_url="https://example.freshservice.com", - ), - { - "method": "POST", - "path": "/tickets/123/notes", - "json_body": {"body": "note"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.reply_to_ticket( - ticket_id=123, - reply={"body": "reply"}, - base_url="https://example.freshservice.com", - ), - { - "method": "POST", - "path": "/tickets/123/reply", - "json_body": {"body": "reply"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.get_requester( - requester_id=123, - base_url="https://example.freshservice.com", - ), - { - "method": "GET", - "path": "/requesters/123", - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.create_requester( - requester={"primary_email": "user@example.com"}, - base_url="https://example.freshservice.com", - ), - { - "method": "POST", - "path": "/requesters", - "json_body": {"primary_email": "user@example.com"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.update_requester( - requester_id=123, - requester={"first_name": "A"}, - base_url="https://example.freshservice.com", - ), - { - "method": "PUT", - "path": "/requesters/123", - "json_body": {"first_name": "A"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.get_agent( - agent_id=123, - base_url="https://example.freshservice.com", - ), - { - "method": "GET", - "path": "/agents/123", - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.get_group( - group_id=123, - base_url="https://example.freshservice.com", - ), - { - "method": "GET", - "path": "/groups/123", - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.get_change( - change_id=123, - base_url="https://example.freshservice.com", - ), - { - "method": "GET", - "path": "/changes/123", - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.create_change( - change={"subject": "Change"}, - base_url="https://example.freshservice.com", - ), - { - "method": "POST", - "path": "/changes", - "json_body": {"subject": "Change"}, - "base_url": "https://example.freshservice.com", - }, - ), - ( - lambda: freshservice.update_change( - change_id=123, - change={"priority": 1}, - base_url="https://example.freshservice.com", - ), - { - "method": "PUT", - "path": "/changes/123", - "json_body": {"priority": 1}, - "base_url": "https://example.freshservice.com", - }, - ), - ], -) -async def test_resource_wrappers_use_documented_contracts( - monkeypatch: pytest.MonkeyPatch, - call: Callable[[], Any], - expected: dict[str, Any], -) -> None: - calls: list[dict[str, Any]] = [] - - async def fake_request(**kwargs: Any) -> dict[str, Any]: - calls.append(kwargs) - return {"ok": True} - - monkeypatch.setattr(freshservice, "_request_freshservice_api", fake_request) - - result = await call() - - assert result == {"ok": True} - assert calls == [expected] - - -@pytest.mark.anyio -@pytest.mark.parametrize( - ("call", "expected_path", "expected_items_key"), - [ - (freshservice.list_tickets, "/tickets", "tickets"), - (freshservice.list_requesters, "/requesters", "requesters"), - (freshservice.list_agents, "/agents", "agents"), - (freshservice.list_groups, "/groups", "groups"), - (freshservice.list_changes, "/changes", "changes"), - ], -) -async def test_list_wrappers_use_documented_contracts( - monkeypatch: pytest.MonkeyPatch, - call: Callable[..., Any], - expected_path: str, - expected_items_key: str, -) -> None: - calls: list[dict[str, Any]] = [] - - async def fake_paginated(**kwargs: Any) -> freshservice.FreshservicePaginatedResult: - calls.append(kwargs) - return {"items": [], "pages": 0, "next_page": None} - - monkeypatch.setattr(freshservice, "_call_paginated_endpoint", fake_paginated) - - result = await call( - query_params={"updated_since": "2026-01-01"}, - page=2, - per_page=10, - max_pages=3, - base_url="https://example.freshservice.com", - ) - - assert result == {"items": [], "pages": 0, "next_page": None} - assert calls == [ - { - "path": expected_path, - "query_params": {"updated_since": "2026-01-01"}, - "items_key": expected_items_key, - "page": 2, - "per_page": 10, - "max_pages": 3, - "base_url": "https://example.freshservice.com", - } - ] diff --git a/tests/registry/test_google_api.py b/tests/registry/test_google_api.py deleted file mode 100644 index d970064996..0000000000 --- a/tests/registry/test_google_api.py +++ /dev/null @@ -1,505 +0,0 @@ -from typing import Any - -import pytest -from tracecat_registry import SecretNotFoundError -from tracecat_registry.integrations import google_api - - -class FakeRequest: - def __init__(self, response: Any) -> None: - self._response = response - - def execute(self) -> Any: - return self._response - - -class FakeMethodResource: - def __init__(self, calls: list[dict[str, Any]], responses: list[Any]): - self._calls = calls - self._responses = responses - - def list(self, **params: Any) -> FakeRequest: - self._calls.append(params) - return FakeRequest(self._responses.pop(0)) - - -class FakeValuesResource: - def __init__(self, calls: list[dict[str, Any]], responses: list[Any]): - self._calls = calls - self._responses = responses - - def get(self, **params: Any) -> FakeRequest: - self._calls.append(params) - return FakeRequest(self._responses.pop(0)) - - -class FakeSpreadsheetsResource: - def __init__(self, calls: list[dict[str, Any]], responses: list[Any]): - self._calls = calls - self._responses = responses - - def values(self) -> FakeValuesResource: - return FakeValuesResource(self._calls, self._responses) - - -class FakeService: - def __init__(self, calls: list[dict[str, Any]], responses: list[Any]): - self._calls = calls - self._responses = responses - - def files(self) -> FakeMethodResource: - return FakeMethodResource(self._calls, self._responses) - - def spreadsheets(self) -> FakeSpreadsheetsResource: - return FakeSpreadsheetsResource(self._calls, self._responses) - - -def test_call_api_prefers_oauth_token(monkeypatch: pytest.MonkeyPatch) -> None: - calls: list[dict[str, Any]] = [] - built: dict[str, Any] = {} - - def build(*args: Any, **kwargs: Any) -> FakeService: - built["args"] = args - built["kwargs"] = kwargs - return FakeService(calls, [{"files": [{"id": "1"}]}]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - params={"pageSize": 10}, - ) - - assert result == {"files": [{"id": "1"}]} - assert calls == [{"pageSize": 10}] - assert built["args"] == ("drive", "v3") - assert built["kwargs"]["credentials"].token == "service-token" - # Default preserves the client library behaviour (bundled static discovery). - assert built["kwargs"]["static_discovery"] is None - # No document is fetched on this path, and the cache must stay disabled: - # googleapiclient checks the cache before it falls back to the bundled - # document, so a cached runtime-fetched doc could otherwise shadow it. - assert built["kwargs"]["cache_discovery"] is False - assert built["kwargs"]["cache"] is None - - -def test_call_api_forwards_static_discovery(monkeypatch: pytest.MonkeyPatch) -> None: - calls: list[dict[str, Any]] = [] - built: dict[str, Any] = {} - - def build(*args: Any, **kwargs: Any) -> FakeService: - built["kwargs"] = kwargs - return FakeService(calls, [{"files": [{"id": "1"}]}]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - google_api.call_api( - service_name="securitycenter", - version="v2", - resource="files", - method_name="list", - static_discovery=False, - ) - - # Non-bundled API versions must be fetched at runtime, and only that path - # gets the cache so the fetched document cannot shadow a bundled one. - assert built["kwargs"]["static_discovery"] is False - assert built["kwargs"]["cache_discovery"] is True - assert built["kwargs"]["cache"] is google_api._discovery_cache - - -def test_discovery_cache_reuses_fetched_document() -> None: - """Runtime-fetched discovery documents are cached per process. - - Discovery documents are large (securitycenter v2 is ~420 KB). Executor - workers are reused across actions, so without this the document would be - re-downloaded on every single call. - """ - cache = google_api._DiscoveryCache() - url = "https://securitycenter.googleapis.com/$discovery/rest?version=v2" - - assert cache.get(url) is None # cold - cache.set(url, '{"name": "securitycenter"}') - assert cache.get(url) == '{"name": "securitycenter"}' # warm: no refetch - # Distinct documents do not collide. - assert cache.get("https://sheets.googleapis.com/$discovery/rest?version=v4") is None - - -def test_call_api_uses_service_account_json_fallback( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - captured: dict[str, Any] = {} - credentials = object() - - def from_service_account_info(info: dict[str, Any], scopes: list[str]) -> object: - captured["info"] = info - captured["scopes"] = scopes - return credentials - - def build(*args: Any, **kwargs: Any) -> FakeService: - captured["build_args"] = args - captured["build_kwargs"] = kwargs - return FakeService(calls, [{"values": [["a"]]}]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: ( - None if key == "GOOGLE_SERVICE_TOKEN" else '{"type":"service_account"}' - ), - ) - monkeypatch.setattr( - google_api.secrets, "get", lambda _key: '{"type":"service_account"}' - ) - monkeypatch.setattr( - google_api.service_account.Credentials, - "from_service_account_info", - from_service_account_info, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_api( - service_name="sheets", - version="v4", - resource="spreadsheets.values", - method_name="get", - params={"spreadsheetId": "sheet-id", "range": "A1:B2"}, - ) - - assert result == {"values": [["a"]]} - assert calls == [{"spreadsheetId": "sheet-id", "range": "A1:B2"}] - assert captured["info"] == {"type": "service_account"} - assert captured["scopes"] == google_api.DEFAULT_SCOPES - assert captured["build_kwargs"]["credentials"] is credentials - - -def test_call_api_uses_service_account_when_scopes_override_oauth_token( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - captured: dict[str, Any] = {} - credentials = object() - - def from_service_account_info(info: dict[str, Any], scopes: list[str]) -> object: - captured["info"] = info - captured["scopes"] = scopes - return credentials - - def build(*_args: Any, **kwargs: Any) -> FakeService: - captured["build_kwargs"] = kwargs - return FakeService(calls, [{"files": [{"id": "1"}]}]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: ( - "service-token" - if key == "GOOGLE_SERVICE_TOKEN" - else '{"type":"service_account"}' - ), - ) - monkeypatch.setattr( - google_api.secrets, "get", lambda _key: '{"type":"service_account"}' - ) - monkeypatch.setattr( - google_api.service_account.Credentials, - "from_service_account_info", - from_service_account_info, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - scopes=["scope-from-input"], - ) - - assert result == {"files": [{"id": "1"}]} - assert captured["info"] == {"type": "service_account"} - assert captured["scopes"] == ["scope-from-input"] - assert captured["build_kwargs"]["credentials"] is credentials - - -def test_call_api_applies_service_account_subject( - monkeypatch: pytest.MonkeyPatch, -) -> None: - class FakeCredentials: - def with_subject(self, subject: str) -> "FakeCredentials": - captured["subject"] = subject - return self - - captured: dict[str, Any] = {} - credentials = FakeCredentials() - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: ( - None if key == "GOOGLE_SERVICE_TOKEN" else '{"type":"service_account"}' - ), - ) - monkeypatch.setattr( - google_api.secrets, "get", lambda _key: '{"type":"service_account"}' - ) - monkeypatch.setattr( - google_api.service_account.Credentials, - "from_service_account_info", - lambda _info, scopes: credentials, - ) - - result = google_api._get_google_credentials(subject="user@example.test") - - assert result is credentials - assert captured["subject"] == "user@example.test" - - -def test_call_api_rejects_service_account_overrides_without_service_account_json( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - - with pytest.raises(SecretNotFoundError, match="GOOGLE_API_CREDENTIALS"): - google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - subject="user@example.test", - ) - - -def test_call_api_rejects_invalid_service_account_json( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: None if key == "GOOGLE_SERVICE_TOKEN" else "{invalid", - ) - monkeypatch.setattr(google_api.secrets, "get", lambda _key: "{invalid") - - with pytest.raises(ValueError, match="not a valid JSON string"): - google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - ) - - -def test_call_api_requires_oauth_or_service_account_json( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr(google_api.secrets, "get_or_default", lambda _key: None) - - with pytest.raises(SecretNotFoundError, match="GOOGLE_SERVICE_TOKEN"): - google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - ) - - -def test_call_api_returns_non_dict_response(monkeypatch: pytest.MonkeyPatch) -> None: - calls: list[dict[str, Any]] = [] - - def build(*_args: Any, **_kwargs: Any) -> FakeService: - return FakeService(calls, [["not-a-dict"]]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - ) - - assert result == ["not-a-dict"] - - -def test_call_paginated_api_returns_pages(monkeypatch: pytest.MonkeyPatch) -> None: - calls: list[dict[str, Any]] = [] - - def build(*_args: Any, **_kwargs: Any) -> FakeService: - return FakeService( - calls, - [ - {"files": [{"id": "1"}], "nextPageToken": "next-token"}, - {"files": [{"id": "2"}]}, - ], - ) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_paginated_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - params={"pageSize": 1}, - ) - - assert result == [ - {"files": [{"id": "1"}], "nextPageToken": "next-token"}, - {"files": [{"id": "2"}]}, - ] - assert calls == [{"pageSize": 1}, {"pageSize": 1, "pageToken": "next-token"}] - - -def test_call_paginated_api_stops_at_max_pages( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - - def build(*_args: Any, **_kwargs: Any) -> FakeService: - return FakeService( - calls, - [ - {"files": [{"id": "1"}], "nextPageToken": "token-1"}, - {"files": [{"id": "2"}], "nextPageToken": "token-2"}, - {"files": [{"id": "3"}]}, - ], - ) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_paginated_api( - service_name="drive", - version="v3", - resource="files", - method_name="list", - params={"pageSize": 1}, - max_pages=2, - ) - - assert result == [ - {"files": [{"id": "1"}], "nextPageToken": "token-1"}, - {"files": [{"id": "2"}], "nextPageToken": "token-2"}, - ] - # No third request is made once max_pages is reached. - assert calls == [{"pageSize": 1}, {"pageSize": 1, "pageToken": "token-1"}] - - -def test_call_paginated_api_supports_custom_token_fields( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - - def build(*_args: Any, **_kwargs: Any) -> FakeService: - return FakeService( - calls, - [ - { - "voidedPurchases": [{"orderId": "1"}], - "tokenPagination": {"nextPageToken": "next-token"}, - }, - { - "voidedPurchases": [{"orderId": "2"}], - "tokenPagination": {}, - }, - ], - ) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - result = google_api.call_paginated_api( - service_name="androidpublisher", - version="v3", - resource="files", - method_name="list", - params={"packageName": "com.example.app"}, - page_token_param="token", - next_page_token_path="tokenPagination.nextPageToken", - ) - - assert result == [ - { - "voidedPurchases": [{"orderId": "1"}], - "tokenPagination": {"nextPageToken": "next-token"}, - }, - { - "voidedPurchases": [{"orderId": "2"}], - "tokenPagination": {}, - }, - ] - assert calls == [ - {"packageName": "com.example.app"}, - {"packageName": "com.example.app", "token": "next-token"}, - ] - - -def test_call_paginated_api_forwards_static_discovery( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """The paginated helper must forward `static_discovery` like `call_api` does. - - Templates for non-bundled API versions (e.g. `securitycenter` v2) rely on - this; without it the client cannot build the service at all. - """ - calls: list[dict[str, Any]] = [] - built: dict[str, Any] = {} - - def build(*args: Any, **kwargs: Any) -> FakeService: - built["kwargs"] = kwargs - return FakeService(calls, [{"files": [{"id": "1"}]}]) - - monkeypatch.setattr( - google_api.secrets, - "get_or_default", - lambda key: "service-token" if key == "GOOGLE_SERVICE_TOKEN" else None, - ) - monkeypatch.setattr(google_api, "build", build) - - google_api.call_paginated_api( - service_name="securitycenter", - version="v2", - resource="files", - method_name="list", - static_discovery=False, - ) - - assert built["kwargs"]["static_discovery"] is False - assert built["kwargs"]["cache_discovery"] is True - assert built["kwargs"]["cache"] is google_api._discovery_cache diff --git a/tests/registry/test_google_scc_templates.py b/tests/registry/test_google_scc_templates.py deleted file mode 100644 index c8e33440dd..0000000000 --- a/tests/registry/test_google_scc_templates.py +++ /dev/null @@ -1,292 +0,0 @@ -"""Regression tests for the `tools.google_scc` templates. - -The templates wrap the generic `tools.google_api` actions, deriving the -discovery-client resource path (`organizations.sources.findings`, etc.) from -the scope or finding name via inline Python steps. These tests pin the exact -request wiring — service, version, resource, and method — without spinning up -the executor or calling Google. -""" - -from __future__ import annotations - -from pathlib import Path -from typing import Any - -import pytest - -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATES_DIR = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/google_scc" -) - -# SCC v2 always returns location-qualified finding names. These mirror the shape -# of a real `list_findings` response, which is what gets fed to the write-back -# actions. -ORG_FINDING = "organizations/123456789/sources/5678/locations/global/findings/abc123" -PROJECT_FINDING = ( - "projects/example-project/sources/5678/locations/global/findings/abc123" -) -# The non-location shape is still accepted (v1-style / other callers). -ORG_FINDING_NO_LOCATION = "organizations/123456789/sources/5678/findings/abc123" - - -def load_template(filename: str) -> TemplateAction: - return TemplateAction.from_yaml(TEMPLATES_DIR / filename) - - -def get_step(template: TemplateAction, ref: str): - return next(s for s in template.definition.steps if s.ref == ref) - - -def get_script(template: TemplateAction, ref: str): - namespace: dict[str, Any] = {} - exec(get_step(template, ref).args["script"], namespace) # noqa: S102 - return namespace["main"] - - -@pytest.mark.parametrize( - "filename", - [ - "list_findings.yml", - "set_finding_state.yml", - "mute_finding.yml", - "list_sources.yml", - ], -) -def test_template_parses_with_expected_namespace(filename: str) -> None: - template = load_template(filename) - assert template.definition.namespace == "tools.google_scc" - assert template.definition.name == filename.removesuffix(".yml") - - -@pytest.mark.parametrize( - ("filename", "call_ref", "expected_action", "expected_method"), - [ - ( - "list_findings.yml", - "list_findings", - "tools.google_api.call_paginated_api", - "list", - ), - ("set_finding_state.yml", "set_state", "tools.google_api.call_api", "setState"), - ("mute_finding.yml", "set_mute", "tools.google_api.call_api", "setMute"), - ( - "list_sources.yml", - "list_sources", - "tools.google_api.call_paginated_api", - "list", - ), - ], -) -def test_request_wiring( - filename: str, call_ref: str, expected_action: str, expected_method: str -) -> None: - template = load_template(filename) - step = get_step(template, call_ref) - assert step.action == expected_action - assert step.args["service_name"] == "securitycenter" - assert step.args["version"] == "v2" - assert step.args["method_name"] == expected_method - # v2 is not bundled in the client's static discovery docs, so every SCC - # call must fetch the discovery document at runtime. - assert step.args["static_discovery"] is False - - -class TestListFindingsBuildRequest: - @pytest.fixture(scope="class") - def build_request(self): - return get_script(load_template("list_findings.yml"), "build_request") - - def test_organization_scope_defaults(self, build_request) -> None: - result = build_request("organizations/123456789", "-", None, None, 100) - assert result == { - "resource": "organizations.sources.findings", - "params": { - "parent": "organizations/123456789/sources/-", - "pageSize": 100, - }, - } - - def test_project_scope_with_filter_and_order(self, build_request) -> None: - result = build_request( - "projects/example-project", - "-", - 'state="ACTIVE"', - "event_time desc", - 50, - ) - assert result["resource"] == "projects.sources.findings" - assert result["params"] == { - "parent": "projects/example-project/sources/-", - "pageSize": 50, - "filter": 'state="ACTIVE"', - "orderBy": "event_time desc", - } - - def test_folder_scope(self, build_request) -> None: - result = build_request("folders/987", "-", None, None, 100) - assert result["resource"] == "folders.sources.findings" - assert result["params"]["parent"] == "folders/987/sources/-" - - @pytest.mark.parametrize( - "bad_scope", - [ - "", # empty - "organizations", # no id - "organizations/", # empty id - "projects", # no id - "billingAccounts/123", # not a valid SCC scope type - "__class__/x", # would otherwise be traversed via getattr - ], - ) - def test_rejects_invalid_scope(self, build_request, bad_scope: str) -> None: - # The scope type becomes part of the discovery-client resource path, so - # it must be validated before it is resolved via getattr. - with pytest.raises(ValueError, match="scope"): - build_request(bad_scope, "-", None, None, 100) - - -class TestListFindingsFlatten: - @pytest.fixture(scope="class") - def flatten(self): - return get_script(load_template("list_findings.yml"), "flatten") - - def test_flattens_results_across_pages(self, flatten) -> None: - pages = [ - { - "listFindingsResults": [ - {"finding": {"name": ORG_FINDING}, "resource": {"name": "vm-1"}} - ], - "nextPageToken": "t", - }, - { - "listFindingsResults": [ - {"finding": {"name": PROJECT_FINDING}, "resource": {"name": "vm-2"}} - ] - }, - ] - result = flatten(pages) - assert [r["finding"]["name"] for r in result["findings"]] == [ - ORG_FINDING, - PROJECT_FINDING, - ] - # Last page carries no nextPageToken -> the result set is complete. - assert result["truncated"] is False - assert result["next_page_token"] is None - - def test_empty_pages_return_empty_list(self, flatten) -> None: - # Pages with no findings omit `listFindingsResults` entirely. - result = flatten([{"totalSize": 0}]) - assert result["findings"] == [] - assert result["truncated"] is False - assert result["total_size"] == 0 - - def test_no_pages_at_all(self, flatten) -> None: - result = flatten([]) - assert result["findings"] == [] - assert result["truncated"] is False - - def test_truncated_result_is_flagged(self, flatten) -> None: - # `max_pages` stops the fetch early, so the final page still carries a - # nextPageToken. A silently truncated list would let a triage workflow - # believe it had seen every finding. - pages = [ - { - "listFindingsResults": [ - {"finding": {"name": ORG_FINDING}, "resource": {"name": "vm-1"}} - ], - "nextPageToken": "more-findings-exist", - "totalSize": 20000, - } - ] - result = flatten(pages) - assert result["truncated"] is True - assert result["next_page_token"] == "more-findings-exist" - assert result["total_size"] == 20000 - assert len(result["findings"]) == 1 - - -@pytest.mark.parametrize("filename", ["set_finding_state.yml", "mute_finding.yml"]) -class TestFindingResourceDerivation: - def test_location_qualified_organization_finding(self, filename: str) -> None: - # This is the shape SCC v2 actually returns from list_findings. The - # non-location resource rejects it via its `name` pattern, so it must - # resolve to the `.locations.` variant. - build_resource = get_script(load_template(filename), "build_resource") - assert build_resource(ORG_FINDING) == "organizations.sources.locations.findings" - - def test_location_qualified_project_finding(self, filename: str) -> None: - build_resource = get_script(load_template(filename), "build_resource") - assert build_resource(PROJECT_FINDING) == "projects.sources.locations.findings" - - def test_non_location_finding_uses_plain_resource(self, filename: str) -> None: - build_resource = get_script(load_template(filename), "build_resource") - assert ( - build_resource(ORG_FINDING_NO_LOCATION) == "organizations.sources.findings" - ) - - @pytest.mark.parametrize( - "bad_name", - [ - "", - "organizations/123", # not a finding name - "organizations/123/sources/5678", # missing /findings/ - "billingAccounts/123/sources/1/findings/abc", # invalid scope type - "__class__/sources/1/findings/abc", # would be traversed via getattr - ], - ) - def test_rejects_invalid_finding_name(self, filename: str, bad_name: str) -> None: - # The scope type becomes part of the discovery-client resource path, so - # it must be validated before it is resolved via getattr. - build_resource = get_script(load_template(filename), "build_resource") - with pytest.raises(ValueError, match="finding_name"): - build_resource(bad_name) - - -def test_set_state_passes_name_and_state_body() -> None: - template = load_template("set_finding_state.yml") - params = get_step(template, "set_state").args["params"] - assert params["name"] == "${{ inputs.finding_name }}" - assert params["body"] == {"state": "${{ inputs.state }}"} - - -def test_mute_passes_name_and_mute_body() -> None: - template = load_template("mute_finding.yml") - params = get_step(template, "set_mute").args["params"] - assert params["name"] == "${{ inputs.finding_name }}" - assert params["body"] == {"mute": "${{ inputs.mute }}"} - - -def test_list_sources_build_request() -> None: - build_request = get_script(load_template("list_sources.yml"), "build_request") - assert build_request("organizations/123456789", 100) == { - "resource": "organizations.sources", - "params": {"parent": "organizations/123456789", "pageSize": 100}, - } - - -def test_list_sources_flatten() -> None: - flatten = get_script(load_template("list_sources.yml"), "flatten") - pages = [ - {"sources": [{"name": "organizations/123456789/sources/1"}]}, - {"sources": [{"name": "organizations/123456789/sources/2"}]}, - {}, - ] - result = flatten(pages) - assert [s["name"] for s in result["sources"]] == [ - "organizations/123456789/sources/1", - "organizations/123456789/sources/2", - ] - assert result["truncated"] is False - - -def test_list_findings_expects_contract() -> None: - expects = load_template("list_findings.yml").definition.expects - assert expects["source"].default == "-" - assert expects["page_size"].default == 100 - assert expects["filter"].type == "str | None" - # Bounded by default: an unbounded fetch materializes every page in memory - # and returns them as one action result, which can OOM the executor on a - # real org scope. Callers opt into a full fetch explicitly (null). - assert expects["max_pages"].default == 10 diff --git a/tests/registry/test_kubernetes_sdk.py b/tests/registry/test_kubernetes_sdk.py deleted file mode 100644 index 021dd381e7..0000000000 --- a/tests/registry/test_kubernetes_sdk.py +++ /dev/null @@ -1,264 +0,0 @@ -from typing import Any - -import pytest -import yaml -from kubernetes import config as kube_config -from tracecat_registry.integrations import kubernetes_sdk - - -def _kubeconfig(**user_overrides: Any) -> str: - user = {"token": "kube-token", **user_overrides} - return yaml.safe_dump( - { - "apiVersion": "v1", - "kind": "Config", - "clusters": [ - { - "name": "cluster", - "cluster": { - "server": "https://kubernetes.example.com", - "certificate-authority-data": "Y2E=", - }, - } - ], - "contexts": [ - { - "name": "ctx", - "context": {"cluster": "cluster", "user": "user"}, - }, - { - "name": "secret-context", - "context": {"cluster": "cluster", "user": "user"}, - }, - ], - "current-context": "ctx", - "users": [{"name": "user", "user": user}], - } - ) - - -def test_call_api_uses_isolated_kubeconfig_and_passes_api_client(monkeypatch) -> None: - calls: dict[str, Any] = {} - - class Configuration: - pass - - class ApiClient: - def __init__(self, *, configuration: Configuration) -> None: - calls["api_client_configuration"] = configuration - - def sanitize_for_serialization(self, result: Any) -> Any: - calls["sanitized"] = result - return {"items": result["items"]} - - class Loader: - def __init__(self, *, config_dict: dict[str, Any], active_context: str) -> None: - calls["config_dict"] = config_dict - calls["active_context"] = active_context - - def load_and_set(self, configuration: Configuration) -> None: - calls["loader_configuration"] = configuration - - class CoreV1Api: - def __init__(self, *, api_client: ApiClient) -> None: - calls["api_client"] = api_client - - def list_namespaced_pod(self, **params: Any) -> dict[str, Any]: - calls["params"] = params - return {"items": [{"metadata": {"name": "pod-a"}}]} - - monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) - monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) - monkeypatch.setattr(kubernetes_sdk.client, "CoreV1Api", CoreV1Api) - monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) - monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) - monkeypatch.setattr( - kubernetes_sdk.secrets, - "get_or_default", - lambda key: "secret-context" if key == "KUBECONFIG_CONTEXT" else None, - ) - - result = kubernetes_sdk.call_api( - api_class="CoreV1Api", - method_name="list_namespaced_pod", - params={"namespace": "default"}, - ) - - assert result == {"items": [{"metadata": {"name": "pod-a"}}]} - assert calls["config_dict"]["users"][0]["user"]["token"] == "kube-token" - assert calls["active_context"] == "secret-context" - assert calls["api_client"].__class__ is ApiClient - assert calls["api_client_configuration"] is calls["loader_configuration"] - assert calls["params"] == {"namespace": "default"} - - -def test_build_api_client_uses_secret_context(monkeypatch) -> None: - calls: dict[str, Any] = {} - - class Configuration: - pass - - class ApiClient: - def __init__(self, *, configuration: Configuration) -> None: - calls["api_client_configuration"] = configuration - - class Loader: - def __init__( - self, *, config_dict: dict[str, Any], active_context: str | None - ) -> None: - calls["active_context"] = active_context - - def load_and_set(self, configuration: Configuration) -> None: - calls["loader_configuration"] = configuration - - monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) - monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) - monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) - monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) - monkeypatch.setattr( - kubernetes_sdk.secrets, - "get_or_default", - lambda key: "secret-context" if key == "KUBECONFIG_CONTEXT" else None, - ) - - kubernetes_sdk._build_api_client() - - assert calls["active_context"] == "secret-context" - assert calls["api_client_configuration"] is calls["loader_configuration"] - - -@pytest.mark.parametrize("blank_context", ["", " "]) -def test_build_api_client_normalizes_blank_secret_context( - monkeypatch, blank_context: str -) -> None: - """A blank KUBECONFIG_CONTEXT secret must fall back to `current-context`.""" - calls: dict[str, Any] = {} - - class Configuration: - pass - - class ApiClient: - def __init__(self, *, configuration: Configuration) -> None: - pass - - class Loader: - def __init__( - self, *, config_dict: dict[str, Any], active_context: str | None - ) -> None: - calls["active_context"] = active_context - - def load_and_set(self, configuration: Configuration) -> None: - pass - - monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) - monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) - monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) - monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) - monkeypatch.setattr( - kubernetes_sdk.secrets, - "get_or_default", - lambda key: blank_context if key == "KUBECONFIG_CONTEXT" else None, - ) - - kubernetes_sdk._build_api_client() - - assert calls["active_context"] is None - - -@pytest.mark.parametrize( - ("cluster_overrides", "user_overrides", "message"), - [ - ( - {"certificate-authority": "/var/run/secrets/kubernetes.io/ca.crt"}, - {}, - "certificate-authority", - ), - ({}, {"client-certificate": "/tmp/client.crt"}, "client-certificate"), - ({}, {"client-key": "/tmp/client.key"}, "client-key"), - ({}, {"tokenFile": "/var/run/secrets/kubernetes.io/token"}, "tokenFile"), - ({}, {"exec": {"command": "kubectl"}}, "exec"), - ({}, {"auth-provider": {"name": "gcp"}}, "auth-provider"), - ], -) -def test_validate_rejects_file_backed_and_dynamic_credentials( - cluster_overrides: dict[str, Any], - user_overrides: dict[str, Any], - message: str, -) -> None: - config = yaml.safe_load(_kubeconfig(**user_overrides)) - config["clusters"][0]["cluster"].update(cluster_overrides) - - with pytest.raises(ValueError, match=message): - kubernetes_sdk._validate_no_executor_credentials(config) - - -def test_validate_ignores_inactive_context_credentials() -> None: - """Unused contexts with exec/file-backed credentials must not be rejected.""" - config = yaml.safe_load(_kubeconfig()) - config["clusters"].append( - { - "name": "other-cluster", - "cluster": { - "server": "https://other.example.com", - "certificate-authority-data": "Y2E=", - }, - } - ) - config["users"].append( - {"name": "other-user", "user": {"exec": {"command": "kubectl"}}} - ) - config["contexts"].append( - { - "name": "other-ctx", - "context": {"cluster": "other-cluster", "user": "other-user"}, - } - ) - - # current-context is the safe inline-token "ctx"; the unsafe context is inactive. - kubernetes_sdk._validate_no_executor_credentials(config) - - # Selecting the unsafe context explicitly is still rejected. - with pytest.raises(ValueError, match="exec"): - kubernetes_sdk._validate_no_executor_credentials( - config, active_context="other-ctx" - ) - - -def test_validate_rejects_unknown_active_context() -> None: - config = yaml.safe_load(_kubeconfig()) - with pytest.raises(ValueError, match="not found"): - kubernetes_sdk._validate_no_executor_credentials( - config, active_context="missing" - ) - - -def test_never_calls_ambient_kubernetes_credential_loaders(monkeypatch) -> None: - def fail(*args: Any, **kwargs: Any) -> None: - raise AssertionError("ambient Kubernetes credential loader was called") - - class Configuration: - set_default = staticmethod(fail) - - class ApiClient: - def __init__(self, *, configuration: Configuration) -> None: - pass - - class Loader: - def __init__( - self, *, config_dict: dict[str, Any], active_context: str | None - ) -> None: - pass - - def load_and_set(self, configuration: Configuration) -> None: - pass - - monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) - monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) - monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) - monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) - monkeypatch.setattr(kubernetes_sdk.secrets, "get_or_default", lambda key: None) - monkeypatch.setattr(kube_config, "load_kube_config", fail, raising=False) - monkeypatch.setattr(kube_config, "new_client_from_config", fail, raising=False) - monkeypatch.setattr(kube_config, "load_incluster_config", fail) - - kubernetes_sdk._build_api_client() diff --git a/tests/registry/test_misp_templates.py b/tests/registry/test_misp_templates.py deleted file mode 100644 index 5696a40533..0000000000 --- a/tests/registry/test_misp_templates.py +++ /dev/null @@ -1,306 +0,0 @@ -"""Contract tests for the agent-oriented MISP enrichment actions.""" - -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path -from typing import Any - -import pytest -from tracecat_registry import RegistrySecret - -from tracecat.dsl.schemas import TemplateExecutionContext -from tracecat.expressions.eval import eval_templated_object -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATE_ROOT = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/misp" -) - - -@dataclass(frozen=True, slots=True) -class CatalogEntry: - name: str - method: str - endpoint: str - - @property - def action(self) -> str: - return f"tools.misp.{self.name}" - - -CATALOG = ( - CatalogEntry("search_events", "POST", "/events/restSearch"), - CatalogEntry("search_attributes", "POST", "/attributes/restSearch"), - CatalogEntry("search_objects", "POST", "/objects/restSearch"), - CatalogEntry("search_event_index", "POST", "/events/index"), - CatalogEntry("get_attribute", "GET", "/attributes/view/"), - CatalogEntry("get_object", "GET", "/objects/view/"), - CatalogEntry("search_tags", "POST", "/tags/search"), - CatalogEntry("list_taxonomies", "GET", "/taxonomies/index"), - CatalogEntry("get_taxonomy", "GET", "/taxonomies/view/"), - CatalogEntry("search_galaxies", "POST", "/galaxies"), - CatalogEntry("get_galaxy", "GET", "/galaxies/view/"), - CatalogEntry( - "search_galaxy_clusters", - "POST", - "/galaxy_clusters/index/", - ), -) - -APPROVED_ACTION_NAMES = frozenset( - "search_events search_attributes search_objects search_event_index " - "get_event get_attribute get_object search_tags list_taxonomies get_taxonomy " - "search_galaxies get_galaxy search_galaxy_clusters search_feeds".split() -) - - -@pytest.fixture(scope="module") -def templates() -> dict[str, tuple[TemplateAction, Path]]: - loaded: dict[str, tuple[TemplateAction, Path]] = {} - for path in sorted(TEMPLATE_ROOT.rglob("*.yml")): - template = TemplateAction.from_yaml(path) - action = template.definition.action - if action in loaded: - pytest.fail(f"Duplicate action in {loaded[action][1]} and {path}") - loaded[action] = (template, path) - return loaded - - -def http_step(template: TemplateAction): - return next( - step for step in template.definition.steps if step.action == "core.http_request" - ) - - -def execute_script(template: TemplateAction, ref: str, **inputs: Any) -> Any: - step = next(step for step in template.definition.steps if step.ref == ref) - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"](**inputs) - - -def test_catalog_matches_official_mcp_surface( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - expected = {f"tools.misp.{name}" for name in APPROVED_ACTION_NAMES} - assert len(expected) == 14 - assert set(templates) == expected - - -@pytest.mark.parametrize("entry", CATALOG, ids=lambda entry: entry.action) -def test_action_method_and_endpoint( - entry: CatalogEntry, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = templates[entry.action] - definition = template.definition - request = http_step(template) - - assert definition.namespace == "tools.misp" - assert definition.name == entry.name - assert definition.display_group == "MISP" - assert definition.doc_url is not None - assert definition.doc_url.startswith("https://www.misp-project.org/") - assert request.args["method"] == entry.method - assert entry.endpoint in request.args["url"] - - -def test_common_connection_contract( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - for action, (template, _) in templates.items(): - definition = template.definition - assert definition.secrets is not None - assert len(definition.secrets) == 1 - secret = definition.secrets[0] - assert isinstance(secret, RegistrySecret), action - assert secret.name == "misp", action - assert secret.keys == ["MISP_API_KEY"], action - - assert definition.expects["base_url"].type == "str | None", action - assert definition.expects["base_url"].default is None, action - assert definition.expects["verify_ssl"].type == "bool", action - assert definition.expects["verify_ssl"].default is True, action - assert "payload" not in definition.expects, action - assert "params" not in definition.expects, action - - request = http_step(template) - assert request.args["verify_ssl"] == "${{ inputs.verify_ssl }}", action - assert request.args["headers"]["Authorization"] == ( - "${{ SECRETS.misp.MISP_API_KEY }}" - ), action - assert "inputs.base_url || VARS.misp.base_url" in request.args["url"], action - assert definition.returns == "${{ steps.request.result.data }}", action - - -def test_catalog_contains_only_read_only_enrichment_paths( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - forbidden_path_fragments = ( - "/add", - "/delete", - "/edit", - "/publish", - "/unpublish", - "/enable", - "/disable", - "/update", - "/servers", - "/users", - "/auth", - "/admin", - "/sharing_groups", - "/sightings", - ) - for action, (template, path) in templates.items(): - source = path.read_text() - assert not any(fragment in source for fragment in forbidden_path_fragments), ( - action - ) - method = http_step(template).args["method"] - assert method in {"GET", "POST", "${{ steps.build_request.result.method }}"} - - -@pytest.mark.parametrize( - ("action", "identifier", "path"), - ( - ("tools.misp.get_event", "event_id", "/events/view/42"), - ("tools.misp.get_attribute", "attribute_id", "/attributes/view/42"), - ("tools.misp.get_object", "object_id", "/objects/view/42"), - ("tools.misp.get_taxonomy", "taxonomy_id", "/taxonomies/view/42"), - ("tools.misp.get_galaxy", "galaxy_id", "/galaxies/view/42"), - ( - "tools.misp.search_galaxy_clusters", - "galaxy_id", - "/galaxy_clusters/index/42", - ), - ), -) -def test_numeric_identifiers_are_encoded_as_strings( - action: str, - identifier: str, - path: str, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = templates[action] - inputs = { - identifier: 42, - "base_url": "https://misp.example.com", - "include_correlations": False, - "include_sightings": False, - } - context = TemplateExecutionContext(inputs=inputs, steps={}) - - url = eval_templated_object(http_step(template).args["url"], operand=context) - - assert url == f"https://misp.example.com{path}" - - -def test_search_feeds_matches_official_list_or_search_behavior( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = templates["tools.misp.search_feeds"] - - assert execute_script(template, "build_request", value=None) == { - "method": "GET", - "path": "/feeds/index", - "payload": None, - } - assert execute_script(template, "build_request", value="example.test") == { - "method": "POST", - "path": "/feeds/searchCaches", - "payload": {"value": "example.test"}, - } - - -def test_get_event_matches_official_direct_or_enriched_behavior( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, path = templates["tools.misp.get_event"] - - assert execute_script( - template, - "build_request", - event_id=42, - include_correlations=False, - include_sightings=False, - ) == {"method": "GET", "payload": None} - assert execute_script( - template, - "build_request", - event_id=42, - include_correlations=True, - include_sightings=False, - ) == { - "method": "POST", - "payload": { - "eventid": 42, - "includeCorrelations": True, - "includeSightings": False, - "limit": 1, - }, - } - source = path.read_text() - assert "/events/view/" in source - assert "/events/restSearch" in source - - -@pytest.mark.parametrize( - "action", - ("tools.misp.get_event", "tools.misp.search_feeds"), -) -def test_dual_method_actions_leave_content_type_to_http_client( - action: str, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = templates[action] - - assert "Content-Type" not in http_step(template).args["headers"] - - -@pytest.mark.parametrize( - ("sort", "desc", "expected"), - ( - ("date", None, None), - ("date", False, "asc"), - ("date", True, "desc"), - (None, True, None), - ), -) -def test_event_index_preserves_optional_sort_direction( - sort: str | None, - desc: bool | None, - expected: str | None, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, _ = templates["tools.misp.search_event_index"] - context = TemplateExecutionContext( - inputs={"sort": sort, "desc": desc}, - steps={}, - ) - direction = http_step(template).args["payload"]["direction"] - - assert eval_templated_object(direction, operand=context) == expected - - -def test_search_defaults_are_bounded_and_agent_friendly( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - expected_limits = { - "tools.misp.search_events": 25, - "tools.misp.search_attributes": 50, - "tools.misp.search_objects": 25, - "tools.misp.search_event_index": 25, - } - for action, limit in expected_limits.items(): - definition = templates[action][0].definition - assert definition.expects["limit"].default == limit - assert definition.expects["page"].default == 1 - - assert ( - templates["tools.misp.search_galaxy_clusters"][0] - .definition.expects["context"] - .default - is None - ) diff --git a/tests/registry/test_okta_sdk.py b/tests/registry/test_okta_sdk.py deleted file mode 100644 index 8a464b5581..0000000000 --- a/tests/registry/test_okta_sdk.py +++ /dev/null @@ -1,314 +0,0 @@ -from datetime import UTC, datetime -from types import SimpleNamespace -from typing import Any -from unittest.mock import AsyncMock - -import pytest -from pydantic import BaseModel, ConfigDict, Field -from tracecat_registry import SecretNotFoundError -from tracecat_registry.integrations import okta_sdk - - -class FakeOktaModel(BaseModel): - model_config = ConfigDict(populate_by_name=True) - - id: str - last_updated: datetime = Field(alias="lastUpdated") - - -def _secret_getter(values: dict[str, str]) -> Any: - return lambda key, default=None: values.get(key, default) - - -def test_build_config_prefers_oauth_service_token( - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - okta_sdk.secrets, - "get_or_default", - _secret_getter( - { - "OKTA_BASE_URL": "https://example.okta.com/", - "OKTA_SERVICE_TOKEN": "service-token", - "OKTA_API_TOKEN": "api-token", - } - ), - ) - - config = okta_sdk._build_okta_config() - - assert config["orgUrl"] == "https://example.okta.com" - assert config["authorizationMode"] == "Bearer" - assert config["token"] == "service-token" - - -def test_build_config_supports_ssws(monkeypatch: pytest.MonkeyPatch) -> None: - values = { - "OKTA_BASE_URL": "https://example.okta.com", - "OKTA_API_TOKEN": "api-token", - } - monkeypatch.setattr(okta_sdk.secrets, "get_or_default", _secret_getter(values)) - monkeypatch.setattr(okta_sdk.secrets, "get", lambda key: values[key]) - - config = okta_sdk._build_okta_config(auth_mode="ssws") - - assert config["authorizationMode"] == "SSWS" - assert config["token"] == "api-token" - - -def test_build_config_supports_oauth_alias(monkeypatch: pytest.MonkeyPatch) -> None: - values = { - "OKTA_BASE_URL": "https://example.okta.com", - "OKTA_ACCESS_TOKEN": "access-token", - } - monkeypatch.setattr(okta_sdk.secrets, "get_or_default", _secret_getter(values)) - monkeypatch.setattr(okta_sdk.secrets, "get", lambda key: values[key]) - - config = okta_sdk._build_okta_config(auth_mode="oauth") - - assert config["authorizationMode"] == "Bearer" - assert config["token"] == "access-token" - - -def test_okta_secret_form_accepts_all_sdk_auth_keys() -> None: - assert okta_sdk.okta_secret.optional_keys == [ - "OKTA_BASE_URL", - "OKTA_API_TOKEN", - "OKTA_ACCESS_TOKEN", - "OKTA_SERVICE_TOKEN", - "OKTA_CLIENT_ID", - "OKTA_PRIVATE_KEY", - "OKTA_SCOPES", - "OKTA_KID", - "OKTA_DPOP_ENABLED", - "OKTA_DPOP_KEY_ROTATION_INTERVAL", - ] - assert [secret.name for secret in okta_sdk.OKTA_SDK_SECRETS] == [ - "okta", - "okta_oauth", - ] - - -def test_build_config_supports_private_key(monkeypatch: pytest.MonkeyPatch) -> None: - values = { - "OKTA_BASE_URL": "https://example.okta.com", - "OKTA_CLIENT_ID": "client-id", - "OKTA_PRIVATE_KEY": "pem", - "OKTA_SCOPES": "okta.users.read, okta.groups.read", - "OKTA_KID": "kid", - } - monkeypatch.setattr(okta_sdk.secrets, "get_or_default", _secret_getter(values)) - monkeypatch.setattr(okta_sdk.secrets, "get", lambda key: values[key]) - - config = okta_sdk._build_okta_config(auth_mode="private_key") - - assert config["authorizationMode"] == "PrivateKey" - assert config["clientId"] == "client-id" - assert config["privateKey"] == "pem" - assert config["scopes"] == ["okta.users.read", "okta.groups.read"] - assert config["kid"] == "kid" - - -def test_build_config_supports_private_key_dpop( - monkeypatch: pytest.MonkeyPatch, -) -> None: - values = { - "OKTA_BASE_URL": "https://example.okta.com", - "OKTA_CLIENT_ID": "client-id", - "OKTA_PRIVATE_KEY": "pem", - "OKTA_SCOPES": "okta.users.read", - "OKTA_DPOP_ENABLED": "true", - "OKTA_DPOP_KEY_ROTATION_INTERVAL": "7200", - } - monkeypatch.setattr(okta_sdk.secrets, "get_or_default", _secret_getter(values)) - monkeypatch.setattr(okta_sdk.secrets, "get", lambda key: values[key]) - - config = okta_sdk._build_okta_config(auth_mode="private_key") - - assert config["authorizationMode"] == "PrivateKey" - assert config["dpopEnabled"] is True - assert config["dpopKeyRotationInterval"] == 7200 - - -def test_build_config_requires_auth_source(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr( - okta_sdk.secrets, - "get_or_default", - _secret_getter({"OKTA_BASE_URL": "https://example.okta.com"}), - ) - - with pytest.raises(SecretNotFoundError, match="one auth source"): - okta_sdk._build_okta_config() - - -def test_jsonable_preserves_okta_model_readonly_fields() -> None: - model = FakeOktaModel( - id="00u123", - lastUpdated=datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC), - ) - - assert okta_sdk._jsonable(model) == { - "id": "00u123", - "lastUpdated": "2026-01-02T03:04:05Z", - } - - -@pytest.mark.anyio -async def test_call_method_uses_sdk_method(monkeypatch: pytest.MonkeyPatch) -> None: - client = SimpleNamespace( - list_users=AsyncMock( - return_value=( - [ - FakeOktaModel( - id="00u123", - lastUpdated=datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC), - ) - ], - None, - None, - ) - ) - ) - monkeypatch.setattr(okta_sdk, "_build_okta_client", lambda **_kwargs: client) - - result = await okta_sdk.call_method( - method_name="list_users", - params={"limit": 1}, - base_url="https://example.okta.com", - ) - - assert result == [{"id": "00u123", "lastUpdated": "2026-01-02T03:04:05Z"}] - client.list_users.assert_awaited_once_with(limit=1) - - -def test_add_group_request_preserves_custom_profile_attributes() -> None: - # Custom (schema-unknown) profile attributes must survive into the request. - # The SDK only routes unknown profile fields into `additional_properties` - # when built via `AddGroupRequest.from_dict()`; a raw dict coerced by - # `@validate_call` would silently drop them. - from okta.models.add_group_request import AddGroupRequest - - request = AddGroupRequest.from_dict( - { - "profile": { - "name": "Engineers", - "description": "Eng team", - "costCenter": "CC-42", - } - } - ) - assert request is not None - - body = request.to_dict() - assert body["profile"] == { - "name": "Engineers", - "description": "Eng team", - "costCenter": "CC-42", - } - - -@pytest.mark.anyio -async def test_add_group_passes_request_with_custom_attributes( - monkeypatch: pytest.MonkeyPatch, -) -> None: - from okta.models.add_group_request import AddGroupRequest - - client = SimpleNamespace(add_group=AsyncMock(return_value=({"id": "00g1"}, None))) - monkeypatch.setattr(okta_sdk, "_build_okta_client", lambda **_kwargs: client) - - await okta_sdk.add_group( - group={"profile": {"name": "Engineers", "costCenter": "CC-42"}}, - base_url="https://example.okta.com", - ) - - client.add_group.assert_awaited_once() - sent_group = client.add_group.await_args.kwargs["group"] - assert isinstance(sent_group, AddGroupRequest) - assert sent_group.to_dict()["profile"]["costCenter"] == "CC-42" - - -@pytest.mark.anyio -async def test_replace_group_passes_request_with_custom_attributes( - monkeypatch: pytest.MonkeyPatch, -) -> None: - from okta.models.add_group_request import AddGroupRequest - - client = SimpleNamespace( - replace_group=AsyncMock(return_value=({"id": "00g1"}, None)) - ) - monkeypatch.setattr(okta_sdk, "_build_okta_client", lambda **_kwargs: client) - - await okta_sdk.replace_group( - group_id="00g1", - group={"profile": {"name": "Engineers", "costCenter": "CC-42"}}, - base_url="https://example.okta.com", - ) - - client.replace_group.assert_awaited_once() - await_kwargs = client.replace_group.await_args.kwargs - assert await_kwargs["group_id"] == "00g1" - sent_group = await_kwargs["group"] - assert isinstance(sent_group, AddGroupRequest) - assert sent_group.to_dict()["profile"]["costCenter"] == "CC-42" - - -@pytest.mark.anyio -async def test_call_method_rejects_private_method() -> None: - with pytest.raises(ValueError, match="cannot start"): - await okta_sdk.call_method(method_name="_private") - - -@pytest.mark.anyio -async def test_call_method_raises_two_tuple_sdk_error( - monkeypatch: pytest.MonkeyPatch, -) -> None: - client = SimpleNamespace( - delete_user=AsyncMock(return_value=(None, ValueError("bad"))) - ) - monkeypatch.setattr(okta_sdk, "_build_okta_client", lambda **_kwargs: client) - - with pytest.raises(ValueError, match="bad"): - await okta_sdk.call_method(method_name="delete_user", params={"id": "00u123"}) - - -@pytest.mark.anyio -async def test_call_paginated_method_follows_link_headers( - monkeypatch: pytest.MonkeyPatch, -) -> None: - first_response = SimpleNamespace( - headers={ - "Link": '; rel="next"' - } - ) - second_response = SimpleNamespace(headers={}) - client = SimpleNamespace( - list_users=AsyncMock( - side_effect=[ - ([{"id": "00u1"}], first_response, None), - ([{"id": "00u2"}], second_response, None), - ] - ) - ) - monkeypatch.setattr(okta_sdk, "_build_okta_client", lambda **_kwargs: client) - - result = await okta_sdk.call_paginated_method( - method_name="list_users", - params={"q": "alice"}, - limit=1, - base_url="https://example.okta.com", - ) - - assert result == { - "items": [{"id": "00u1"}, {"id": "00u2"}], - "pages": 2, - "next_after": None, - } - assert client.list_users.await_args_list[0].kwargs == { - "q": "alice", - "limit": 1, - } - assert client.list_users.await_args_list[1].kwargs == { - "q": "alice", - "limit": 1, - "after": "abc", - } diff --git a/tests/registry/test_opensearch_templates.py b/tests/registry/test_opensearch_templates.py deleted file mode 100644 index 256fd4c769..0000000000 --- a/tests/registry/test_opensearch_templates.py +++ /dev/null @@ -1,289 +0,0 @@ -"""Catalog and request-contract tests for OpenSearch template actions.""" - -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path -from typing import Any - -import pytest -from tracecat_registry import RegistrySecret - -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATE_ROOT = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/opensearch" -) -AUTHORIZATION = ( - 'Basic ${{ FN.to_base64(SECRETS.opensearch.OPENSEARCH_USERNAME + ":" + ' - "SECRETS.opensearch.OPENSEARCH_PASSWORD) }}" -) - - -@dataclass(frozen=True, slots=True) -class CatalogEntry: - name: str - method: str - endpoint_fragment: str - - @property - def action(self) -> str: - return f"tools.opensearch.{self.name}" - - -CATALOG = ( - CatalogEntry("list_indexes", "GET", "_cat/indices"), - CatalogEntry("get_mapping", "GET", "_mapping"), - CatalogEntry("search_events", "POST", "_search"), - CatalogEntry("multi_search", "POST", "_msearch"), - CatalogEntry("get_document", "GET", "_doc"), - CatalogEntry("multi_get_documents", "POST", "_mget"), - CatalogEntry("count_events", "POST", "_count"), - CatalogEntry("ppl_query", "POST", "_plugins/_ppl"), - CatalogEntry( - "list_security_analytics_alerts", - "GET", - "_plugins/_security_analytics/alerts", - ), - CatalogEntry( - "acknowledge_security_analytics_alerts", - "POST", - "_plugins/_security_analytics/detectors", - ), - CatalogEntry( - "list_security_analytics_findings", - "GET", - "_plugins/_security_analytics/findings/_search", - ), - CatalogEntry( - "search_detectors", - "POST", - "_plugins/_security_analytics/detectors/_search", - ), - CatalogEntry( - "search_detection_rules", - "POST", - "_plugins/_security_analytics/rules/_search", - ), - CatalogEntry( - "list_monitor_alerts", - "GET", - "_plugins/_alerting/monitors/alerts", - ), - CatalogEntry( - "acknowledge_monitor_alerts", - "POST", - "_plugins/_alerting/monitors", - ), -) - - -@pytest.fixture(scope="module") -def templates() -> dict[str, tuple[TemplateAction, Path]]: - loaded: dict[str, tuple[TemplateAction, Path]] = {} - for path in sorted(TEMPLATE_ROOT.rglob("*.yml")): - template = TemplateAction.from_yaml(path) - loaded[template.definition.action] = (template, path) - return loaded - - -def http_step(template: TemplateAction): - return next( - step for step in template.definition.steps if step.action == "core.http_request" - ) - - -def execute_script(template: TemplateAction, ref: str): - step = next(step for step in template.definition.steps if step.ref == ref) - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"] - - -def get_template( - templates: dict[str, tuple[TemplateAction, Path]], name: str -) -> TemplateAction: - return templates[f"tools.opensearch.{name}"][0] - - -def test_catalog_is_exact(templates: dict[str, tuple[TemplateAction, Path]]) -> None: - expected = {entry.action for entry in CATALOG} - assert len(CATALOG) == 15 - assert len(expected) == 15 - assert set(templates) == expected - - -@pytest.mark.parametrize("entry", CATALOG, ids=lambda entry: entry.action) -def test_metadata_method_and_endpoint( - entry: CatalogEntry, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, path = templates[entry.action] - definition = template.definition - - assert definition.namespace == "tools.opensearch" - assert definition.name == entry.name - assert definition.display_group == "OpenSearch" - assert definition.doc_url is not None - assert definition.doc_url.startswith("https://docs.opensearch.org/") - assert http_step(template).args["method"] == entry.method - assert entry.endpoint_fragment in path.read_text() - - -def test_common_authentication_and_connection_contract( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - for action, (template, _) in templates.items(): - definition = template.definition - assert definition.secrets is not None - assert len(definition.secrets) == 1 - secret = definition.secrets[0] - assert isinstance(secret, RegistrySecret), action - assert secret.name == "opensearch", action - assert secret.keys == [ - "OPENSEARCH_USERNAME", - "OPENSEARCH_PASSWORD", - ], action - - assert definition.expects["base_url"].type == "str | None", action - assert definition.expects["base_url"].default is None, action - assert definition.expects["verify_ssl"].type == "bool", action - assert definition.expects["verify_ssl"].default is True, action - - request = http_step(template) - assert request.args["verify_ssl"] == "${{ inputs.verify_ssl }}", action - assert request.args["headers"]["Authorization"] == AUTHORIZATION, action - assert "inputs.base_url || VARS.opensearch.base_url" in request.args["url"] - assert definition.returns == "${{ steps.request.result.data }}", action - - -@pytest.mark.parametrize( - ("name", "suffix"), - ( - ("search_events", "_search"), - ("multi_search", "_msearch"), - ("multi_get_documents", "_mget"), - ("count_events", "_count"), - ), -) -def test_optional_index_paths( - name: str, - suffix: str, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - build_path = execute_script(get_template(templates, name), "build_path") - assert build_path(None) == f"/{suffix}" - assert build_path("logs-*/events") == f"/logs-*%2Fevents/{suffix}" - - -def test_index_discovery_and_mapping_paths( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - list_indexes = get_template(templates, "list_indexes") - build_path = execute_script(list_indexes, "build_path") - build_params = execute_script(list_indexes, "build_params") - assert build_path(None) == "/_cat/indices" - assert build_path("logs-*/events") == "/_cat/indices/logs-*%2Fevents" - assert build_params(None) == {"format": "json"} - assert build_params({"health": "yellow", "format": "yaml"}) == { - "health": "yellow", - "format": "json", - } - - get_mapping = get_template(templates, "get_mapping") - mapping_path = execute_script(get_mapping, "build_path") - assert mapping_path("logs-*/events") == "/logs-*%2Fevents/_mapping" - - -def test_document_identifiers_are_url_encoded( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - _, path = templates["tools.opensearch.get_document"] - source = path.read_text() - assert "FN.url_encode(inputs.index)" in source - assert "FN.url_encode(inputs.document_id)" in source - - -def test_search_payload_has_bounded_default( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template = get_template(templates, "search_events") - build_payload = execute_script(template, "build_payload") - assert build_payload({"query": {"match_all": {}}}, 100) == { - "query": {"match_all": {}}, - "size": 100, - } - explicit = {"query": {"match_all": {}}, "size": 0} - assert build_payload(explicit, 100) is explicit - assert http_step(template).args["params"] == "${{ inputs.params }}" - - -def test_multi_search_uses_native_ndjson( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template = get_template(templates, "multi_search") - normalize = execute_script(template, "normalize_ndjson") - ndjson = '{}\n{"query":{"match_all":{}}}' - assert normalize(ndjson) == f"{ndjson}\n" - assert normalize(f"{ndjson}\n") == f"{ndjson}\n" - - request = http_step(template) - assert request.args["content"] == "${{ steps.normalize_ndjson.result }}" - assert "payload" not in request.args - assert request.args["headers"]["Content-Type"] == "application/x-ndjson" - - -@pytest.mark.parametrize( - "name", - ( - "multi_get_documents", - "count_events", - "search_detectors", - "search_detection_rules", - ), -) -def test_api_native_payloads_pass_through( - name: str, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template = get_template(templates, name) - assert template.definition.expects["payload"].type == "dict[str, Any]" - assert http_step(template).args["payload"] == "${{ inputs.payload }}" - - -def test_ppl_query_contract( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template = get_template(templates, "ppl_query") - assert template.definition.expects["format"].default == "jdbc" - request = http_step(template) - assert request.args["params"] == {"format": "${{ inputs.format }}"} - assert request.args["payload"] == {"query": "${{ inputs.query }}"} - - -def test_detection_rule_partition_is_explicit( - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template = get_template(templates, "search_detection_rules") - assert template.definition.expects["pre_packaged"].default is True - assert http_step(template).args["params"] == { - "pre_packaged": "${{ inputs.pre_packaged }}" - } - - -@pytest.mark.parametrize( - ("name", "identifier"), - ( - ("acknowledge_security_analytics_alerts", "detector_id"), - ("acknowledge_monitor_alerts", "monitor_id"), - ), -) -def test_acknowledgement_contracts( - name: str, - identifier: str, - templates: dict[str, tuple[TemplateAction, Path]], -) -> None: - template, path = templates[f"tools.opensearch.{name}"] - request = http_step(template) - assert request.args["payload"] == {"alerts": "${{ inputs.alert_ids }}"} - assert f"FN.url_encode(inputs.{identifier})" in path.read_text() diff --git a/tests/registry/test_scanner_templates.py b/tests/registry/test_scanner_templates.py deleted file mode 100644 index a697c4bb35..0000000000 --- a/tests/registry/test_scanner_templates.py +++ /dev/null @@ -1,220 +0,0 @@ -from pathlib import Path -from typing import Any - -import pytest -from tracecat_registry import RegistrySecret - -from tracecat.registry.actions.schemas import ActionStep, TemplateAction - -TEMPLATE_ROOT = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/scanner" -) - -EXPECTED_ACTIONS = { - "tools.scanner.cancel_query", - "tools.scanner.create_detection_rule", - "tools.scanner.create_event_sink", - "tools.scanner.delete_detection_rule", - "tools.scanner.delete_event_sink", - "tools.scanner.get_detection_rule", - "tools.scanner.get_event_sink", - "tools.scanner.get_query_progress", - "tools.scanner.list_detection_rules", - "tools.scanner.list_event_sinks", - "tools.scanner.run_detection_rule_yaml_tests", - "tools.scanner.run_query", - "tools.scanner.start_query", - "tools.scanner.update_detection_rule", - "tools.scanner.update_event_sink", - "tools.scanner.validate_detection_rule_yaml", -} - - -def _load_template(filename: str) -> TemplateAction: - return TemplateAction.from_yaml(TEMPLATE_ROOT / filename) - - -def _step(action: TemplateAction, ref: str) -> ActionStep: - return next(step for step in action.definition.steps if step.ref == ref) - - -def _run_python_step(step: ActionStep, **inputs: Any) -> Any: - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"](**inputs) - - -@pytest.fixture(scope="module") -def templates() -> list[TemplateAction]: - return [ - TemplateAction.from_yaml(path) for path in sorted(TEMPLATE_ROOT.glob("*.yml")) - ] - - -def test_scanner_template_inventory(templates: list[TemplateAction]) -> None: - assert {template.definition.action for template in templates} == EXPECTED_ACTIONS - - -def test_scanner_templates_share_auth_and_base_url_contract( - templates: list[TemplateAction], -) -> None: - for template in templates: - assert template.definition.secrets is not None - secret = template.definition.secrets[0] - assert isinstance(secret, RegistrySecret) - assert secret.name == "scanner" - assert secret.keys == ["SCANNER_API_KEY"] - - base_url = template.definition.expects["base_url"] - assert base_url.type == "str | None" - assert base_url.default is None - - http_step = next( - step - for step in template.definition.steps - if step.action == "core.http_request" - ) - assert http_step.args["url"].startswith( - "${{ inputs.base_url || VARS.scanner.base_url }}" - ) - assert http_step.args["headers"]["Authorization"] == ( - "Bearer ${{ SECRETS.scanner.SCANNER_API_KEY }}" - ) - - -def test_query_payload_omits_none_and_preserves_false() -> None: - template = _load_template("run_query.yml") - payload = _run_python_step( - _step(template, "build_query_payload"), - query="error | count", - start_time="2026-07-13T00:00:00Z", - end_time="2026-07-13T01:00:00Z", - max_rows=None, - max_bytes=None, - scan_back_to_front=False, - ) - - assert payload == { - "query": "error | count", - "start_time": "2026-07-13T00:00:00Z", - "end_time": "2026-07-13T01:00:00Z", - "scan_back_to_front": False, - } - - -def test_detection_rule_pagination_uses_documented_parameter_names() -> None: - template = _load_template("list_detection_rules.yml") - params = _run_python_step( - _step(template, "build_pagination_params"), - tenant_id="00000000-0000-0000-0000-000000000001", - page_size=25, - page_token="next-page", - ) - - assert params == { - "tenant_id": "00000000-0000-0000-0000-000000000001", - "pagination[page_size]": 25, - "pagination[page_token]": "next-page", - } - - -def test_create_detection_rule_matches_documented_contract() -> None: - template = _load_template("create_detection_rule.yml") - assert set(template.definition.expects) == { - "tenant_id", - "name", - "description", - "time_range_s", - "run_frequency_s", - "enabled_state_override", - "severity", - "query_text", - "event_sink_ids", - "tags", - "sync_key", - "base_url", - } - - payload = _run_python_step( - _step(template, "build_detection_rule"), - tenant_id="00000000-0000-0000-0000-000000000001", - name="Example detection", - description="Detect an example event", - time_range_s=300, - run_frequency_s=300, - enabled_state_override="Active", - severity="Information", - query_text="error | count", - event_sink_ids=[], - tags=None, - sync_key=None, - ) - - assert payload["event_sink_ids"] == [] - assert "tags" not in payload - assert "sync_key" not in payload - - -@pytest.mark.parametrize( - ("filename", "step_ref", "resource_id_name"), - [ - ( - "update_detection_rule.yml", - "build_detection_rule_update", - "detection_rule_id", - ), - ("update_event_sink.yml", "build_event_sink_update", "event_sink_id"), - ], -) -def test_update_payload_uses_explicit_resource_id( - filename: str, - step_ref: str, - resource_id_name: str, -) -> None: - template = _load_template(filename) - payload = _run_python_step( - _step(template, step_ref), - **{resource_id_name: "resource-id", "updates": {"id": "wrong-id"}}, - ) - - assert payload == {"id": "resource-id"} - - -def test_list_event_sinks_uses_only_documented_tenant_parameter() -> None: - template = _load_template("list_event_sinks.yml") - assert set(template.definition.expects) == {"tenant_id", "base_url"} - assert _step(template, "list_event_sinks").args["params"] == { - "tenant_id": "${{ inputs.tenant_id }}" - } - - -@pytest.mark.parametrize( - ("filename", "step_ref", "path"), - [ - ( - "validate_detection_rule_yaml.yml", - "validate_detection_rule_yaml", - "/v1/detection_rule_yaml/validate", - ), - ( - "run_detection_rule_yaml_tests.yml", - "run_detection_rule_yaml_tests", - "/v1/detection_rule_yaml/run_tests", - ), - ], -) -def test_detection_yaml_templates_send_raw_content( - filename: str, - step_ref: str, - path: str, -) -> None: - template = _load_template(filename) - args = _step(template, step_ref).args - - assert args["url"].endswith(path) - assert args["method"] == "POST" - assert args["content"] == "${{ inputs.yaml_text }}" - assert args["headers"]["Content-Type"] == "application/x-yaml" - assert "payload" not in args - assert "form_data" not in args - assert "files" not in args diff --git a/tests/registry/test_sentinel_one_templates.py b/tests/registry/test_sentinel_one_templates.py deleted file mode 100644 index d2f38683b5..0000000000 --- a/tests/registry/test_sentinel_one_templates.py +++ /dev/null @@ -1,226 +0,0 @@ -from collections.abc import Callable, Mapping -from pathlib import Path -from typing import cast - -import pytest - -from tracecat.registry.actions.schemas import TemplateAction - -TEMPLATE_ROOT = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/sentinel_one" -) - - -def load_template(filename: str) -> TemplateAction: - return TemplateAction.from_yaml(TEMPLATE_ROOT / filename) - - -@pytest.mark.parametrize( - ("filename", "action_name", "endpoint", "data_key", "data_input"), - [ - ( - "update_alert_analyst_verdict.yml", - "tools.sentinel_one.update_alert_analyst_verdict", - "/web/api/v2.1/cloud-detection/alerts/analyst-verdict", - "analystVerdict", - "${{ inputs.analyst_verdict }}", - ), - ( - "update_alert_incident_status.yml", - "tools.sentinel_one.update_alert_incident_status", - "/web/api/v2.1/cloud-detection/alerts/incident", - "incidentStatus", - "${{ inputs.incident_status }}", - ), - ], -) -def test_alert_lifecycle_template_contract( - filename: str, - action_name: str, - endpoint: str, - data_key: str, - data_input: str, -) -> None: - template = load_template(filename) - definition = template.definition - - assert definition.action == action_name - assert definition.expects["alert_ids"].type == "list[str]" - assert definition.expects["base_url"].default is None - - step = definition.steps[0] - assert step.action == "core.http_request" - assert step.args["method"] == "POST" - assert step.args["url"] == ( - "${{ inputs.base_url || VARS.sentinel_one.base_url }}" + endpoint - ) - assert step.args["payload"] == { - "filter": {"ids": "${{ inputs.alert_ids }}"}, - "data": {data_key: data_input}, - } - - -@pytest.mark.parametrize("filename", ["powerquery.yml", "submit_powerquery.yml"]) -def test_powerquery_submission_contract(filename: str) -> None: - definition = load_template(filename).definition - - assert definition.expects["query_priority"].default == "LOW" - assert definition.expects["result_type"].default == "TABLE" - assert definition.expects["frequency"].default == "LOW" - assert definition.expects["timeout_seconds"].default == 60 - - build_step = definition.steps[0] - namespace: dict[str, object] = {} - exec(build_step.args["script"], namespace) - build_payload = cast(Callable[..., object], namespace["main"]) - result = build_payload( - start_time="24h", - end_time="0s", - query_priority="LOW", - tenant=True, - account_ids=None, - query_origin=None, - query="| group count() by event.type", - result_type="TABLE", - frequency="LOW", - ) - - assert isinstance(result, Mapping) - payload = result["payload"] - assert isinstance(payload, Mapping) - assert payload["queryType"] == "PQ" - assert payload["queryPriority"] == "LOW" - assert payload["pq"] == { - "query": "| group count() by event.type", - "resultType": "TABLE", - "frequency": "LOW", - } - assert payload["tenant"] is True - assert "accountIds" not in payload - - submit_step = definition.steps[1] - assert submit_step.action == "core.http_request" - assert submit_step.args["method"] == "POST" - assert submit_step.args["url"].endswith("/sdl/v2/api/queries") - assert submit_step.args["timeout"] == "${{ inputs.timeout_seconds }}" - - -def test_powerquery_poll_contract() -> None: - definition = load_template("powerquery.yml").definition - - assert definition.expects["poll_interval"].default == 5 - assert definition.expects["poll_max_attempts"].default == 60 - - poll_step = definition.steps[2] - assert poll_step.action == "core.http_poll" - assert poll_step.args["method"] == "GET" - assert poll_step.args["params"] == {"lastStepSeen": 0} - assert poll_step.args["headers"]["X-Dataset-Query-Forward-Tag"] == ( - '${{ steps.submit_query.result.headers["x-dataset-query-forward-tag"] }}' - ) - - delete_step = definition.steps[3] - assert delete_step.action == "core.http_request" - assert delete_step.args["method"] == "DELETE" - assert delete_step.args["url"].endswith( - "/sdl/v2/api/queries/${{ steps.submit_query.result.data.id }}" - ) - assert delete_step.args["headers"]["X-Dataset-Query-Forward-Tag"] == ( - '${{ steps.submit_query.result.headers["x-dataset-query-forward-tag"] }}' - ) - - -@pytest.mark.parametrize( - ("filename", "method", "endpoint"), - [ - ( - "get_powerquery_results.yml", - "GET", - "/sdl/v2/api/queries/${{ inputs.query_id }}", - ), - ( - "delete_powerquery.yml", - "DELETE", - "/sdl/v2/api/queries/${{ inputs.query_id }}", - ), - ], -) -def test_powerquery_followup_contract( - filename: str, method: str, endpoint: str -) -> None: - definition = load_template(filename).definition - step = definition.steps[0] - - assert definition.expects["timeout_seconds"].default == 60 - assert step.args["method"] == method - assert step.args["url"] == ( - "${{ inputs.base_url || VARS.sentinel_one.base_url }}" + endpoint - ) - assert step.args["headers"]["Authorization"].startswith("Bearer ") - assert step.args["headers"]["X-Dataset-Query-Forward-Tag"] == ( - "${{ inputs.forward_tag }}" - ) - - -def test_purple_ai_graphql_contract() -> None: - definition = load_template("purple_ai.yml").definition - - assert definition.expects["timeout_seconds"].default == 120 - - build_step = definition.steps[0] - namespace: dict[str, object] = {} - exec(build_step.args["script"], namespace) - build_query = cast(Callable[..., object], namespace["main"]) - query = build_query( - base_url="https://console.example.test", - console_id=None, - tenant_id="tenant-id", - account_id="account-id", - site_id=None, - version="test-version", - start_time=1, - end_time=2, - ) - - assert isinstance(query, str) - assert "query SimpleTestQuery($input: String!)" in query - assert "purpleLaunchQuery" in query - assert query.count("tenantDetails:") == 2 - assert query.count("userTime:") == 2 - assert "displayedTimeRange: { start: 1, end: 2 }" in query - - request_step = definition.steps[1] - assert request_step.args["url"].endswith("/web/api/v2.1/graphql") - assert request_step.args["headers"]["Authorization"].startswith("ApiToken ") - assert request_step.args["payload"] == { - "query": "${{ steps.build_query.result }}", - "variables": {"input": "${{ inputs.question }}"}, - } - assert request_step.args["timeout"] == "${{ inputs.timeout_seconds }}" - - -def test_graphql_passthrough_contract() -> None: - definition = load_template("graphql.yml").definition - step = definition.steps[0] - - assert definition.expects["endpoint"].default == "/web/api/v2.1/graphql" - assert definition.expects["auth_scheme"].default == "ApiToken" - assert definition.expects["timeout_seconds"].default == 30 - assert step.args["method"] == "POST" - assert step.args["payload"] == { - "query": "${{ inputs.query }}", - "variables": "${{ inputs.variables }}", - } - - -def test_inventory_search_contract() -> None: - definition = load_template("list_inventory.yml").definition - step = definition.steps[0] - - assert step.args["url"].endswith("/web/api/v2.1/xdr/assets") - assert step.args["method"] == "POST" - assert step.args["headers"]["Authorization"].startswith("Bearer ") - assert step.args["payload"] == { - "filter": '${{ FN.merge([inputs.filters, {"limit": inputs.limit, "skip": inputs.skip}]) }}' - } - assert step.args["timeout"] == 30 diff --git a/tests/registry/test_slack_sdk.py b/tests/registry/test_slack_sdk.py deleted file mode 100644 index 34c0f76d59..0000000000 --- a/tests/registry/test_slack_sdk.py +++ /dev/null @@ -1,172 +0,0 @@ -from types import SimpleNamespace -from unittest.mock import AsyncMock - -import pytest -from tracecat_registry.integrations import slack_sdk - - -@pytest.mark.anyio -async def test_call_method_uses_sdk_method_when_available(monkeypatch) -> None: - client = SimpleNamespace( - chat_postMessage=AsyncMock(return_value=SimpleNamespace(data={"ok": True})), - api_call=AsyncMock(), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - result = await slack_sdk.call_method( - sdk_method="chat_postMessage", - params={"channel": "C123", "text": "hello"}, - ) - - assert result == {"ok": True} - client.chat_postMessage.assert_awaited_once_with(channel="C123", text="hello") - client.api_call.assert_not_awaited() - - -@pytest.mark.anyio -async def test_call_method_falls_back_to_raw_web_api_method(monkeypatch) -> None: - client = SimpleNamespace( - api_call=AsyncMock(return_value=SimpleNamespace(data={"ok": True})), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - result = await slack_sdk.call_method( - sdk_method="assistant.threads.setStatus", - params={ - "channel_id": "C123", - "thread_ts": "1700000000.001", - "status": "is thinking...", - }, - ) - - assert result == {"ok": True} - client.api_call.assert_awaited_once_with( - api_method="assistant.threads.setStatus", - json={ - "channel_id": "C123", - "thread_ts": "1700000000.001", - "status": "is thinking...", - }, - ) - - -class FakePaginator: - """Mimics an awaited AsyncSlackResponse that yields one response per page.""" - - def __init__(self, pages: list[dict]) -> None: - self._pages = pages - self.pages_fetched = 0 - - def __aiter__(self): - return self._iterate() - - async def _iterate(self): - for page_data in self._pages: - self.pages_fetched += 1 - yield SimpleNamespace(data=page_data) - - -@pytest.mark.anyio -async def test_call_paginated_method_stops_at_limit(monkeypatch) -> None: - paginator = FakePaginator( - [ - {"messages": [{"ts": "1"}, {"ts": "2"}]}, - {"messages": [{"ts": "3"}, {"ts": "4"}]}, - {"messages": [{"ts": "5"}, {"ts": "6"}]}, - ] - ) - client = SimpleNamespace( - conversations_history=AsyncMock(return_value=paginator), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - result = await slack_sdk.call_paginated_method( - sdk_method="conversations_history", - params={"channel": "C123"}, - key="messages", - limit=3, - ) - - assert result == [{"ts": "1"}, {"ts": "2"}, {"ts": "3"}] - # Pagination must stop once the limit is reached, not exhaust all pages. - assert paginator.pages_fetched == 2 - client.conversations_history.assert_awaited_once_with(channel="C123", limit=3) - - -@pytest.mark.anyio -async def test_call_paginated_method_returns_all_items_below_limit( - monkeypatch, -) -> None: - paginator = FakePaginator( - [ - {"members": ["U1", "U2"]}, - {"members": ["U3"]}, - ] - ) - client = SimpleNamespace( - conversations_members=AsyncMock(return_value=paginator), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - result = await slack_sdk.call_paginated_method( - sdk_method="conversations_members", - params={"channel": "C123"}, - key="members", - limit=10, - ) - - assert result == ["U1", "U2", "U3"] - assert paginator.pages_fetched == 2 - - -@pytest.mark.anyio -async def test_call_paginated_method_without_key_caps_pages(monkeypatch) -> None: - paginator = FakePaginator( - [ - {"ok": True, "page": 1}, - {"ok": True, "page": 2}, - {"ok": True, "page": 3}, - ] - ) - client = SimpleNamespace( - conversations_history=AsyncMock(return_value=paginator), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - result = await slack_sdk.call_paginated_method( - sdk_method="conversations_history", - params={"channel": "C123"}, - limit=2, - ) - - assert result == [{"ok": True, "page": 1}, {"ok": True, "page": 2}] - assert paginator.pages_fetched == 2 - - -@pytest.mark.anyio -async def test_call_paginated_method_raises_on_missing_key(monkeypatch) -> None: - paginator = FakePaginator([{"members": ["U1"]}]) - client = SimpleNamespace( - conversations_history=AsyncMock(return_value=paginator), - ) - - monkeypatch.setattr(slack_sdk.secrets, "get", lambda _key: "xoxb-token") - monkeypatch.setattr(slack_sdk, "AsyncWebClient", lambda token: client) - - with pytest.raises(ValueError, match="not found in data"): - await slack_sdk.call_paginated_method( - sdk_method="conversations_history", - params={"channel": "C123"}, - key="messages", - limit=10, - ) diff --git a/tests/unit/test_microsoft_sentinel_templates.py b/tests/unit/test_microsoft_sentinel_templates.py deleted file mode 100644 index 61a01caebe..0000000000 --- a/tests/unit/test_microsoft_sentinel_templates.py +++ /dev/null @@ -1,113 +0,0 @@ -from pathlib import Path -from typing import Any - -from tracecat.dsl.schemas import MaterializedTaskResult, TemplateExecutionContext -from tracecat.expressions.eval import eval_templated_object -from tracecat.registry.actions.schemas import ActionStep, TemplateAction - -TEMPLATE_ROOT = Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/microsoft_sentinel" -) - -SUBSCRIPTION_ID = "00000000-1111-2222-3333-444444444444" -RESOURCE_GROUP_NAME = "example-resource-group" -WORKSPACE_NAME = "example-workspace" -INCIDENT_ID = "55555555-6666-7777-8888-999999999999" -INCIDENT_ARM_ID = ( - f"/subscriptions/{SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP_NAME}/" - f"providers/Microsoft.OperationalInsights/workspaces/{WORKSPACE_NAME}/" - f"providers/Microsoft.SecurityInsights/Incidents/{INCIDENT_ID}" -) - - -def _load_template(path: str) -> TemplateAction: - return TemplateAction.from_yaml(TEMPLATE_ROOT / path) - - -def _step(action: TemplateAction, ref: str) -> ActionStep: - return next(step for step in action.definition.steps if step.ref == ref) - - -def _run_python_step( - step: ActionStep, context: TemplateExecutionContext -) -> dict[str, Any]: - inputs = eval_templated_object(step.args["inputs"], operand=context) - namespace: dict[str, Any] = {} - exec(step.args["script"], namespace) # noqa: S102 - return namespace["main"](**inputs) - - -def _materialized_result(result: Any) -> MaterializedTaskResult: - return { - "result": result, - "result_typename": type(result).__name__, - "error": None, - "error_typename": None, - "interaction": None, - "interaction_id": None, - "interaction_type": None, - } - - -def _render_http_url( - template_path: str, - http_ref: str, - inputs: dict[str, Any], -) -> str: - action = _load_template(template_path) - context = TemplateExecutionContext(inputs=inputs, steps={}) - - normalize_step = _step(action, "normalize_ids") - normalized = _run_python_step(normalize_step, context) - context["steps"]["normalize_ids"] = _materialized_result(normalized) - - http_step = _step(action, http_ref) - return eval_templated_object(http_step.args["url"], operand=context) - - -def _sentinel_inputs(incident_id: str) -> dict[str, Any]: - return { - "base_url": "https://management.azure.com", - "subscription_id": SUBSCRIPTION_ID, - "resource_group_name": RESOURCE_GROUP_NAME, - "workspace_name": WORKSPACE_NAME, - "incident_id": incident_id, - } - - -def test_get_incident_normalizes_full_arm_id() -> None: - url = _render_http_url( - "incidents/get_incident.yml", - "get_incident", - _sentinel_inputs(INCIDENT_ARM_ID), - ) - - assert url == ( - f"https://management.azure.com/subscriptions/{SUBSCRIPTION_ID}" - f"/resourceGroups/{RESOURCE_GROUP_NAME}" - f"/providers/Microsoft.OperationalInsights/workspaces/{WORKSPACE_NAME}" - f"/providers/Microsoft.SecurityInsights/incidents/{INCIDENT_ID}" - ) - assert url.count("/subscriptions/") == 1 - - -def test_list_incident_comments_normalizes_full_arm_id() -> None: - url = _render_http_url( - "incidents/list_incident_comments.yml", - "list_comments", - _sentinel_inputs(INCIDENT_ARM_ID), - ) - - assert url.endswith(f"/incidents/{INCIDENT_ID}/comments") - assert url.count("/subscriptions/") == 1 - - -def test_get_incident_keeps_plain_incident_id() -> None: - url = _render_http_url( - "incidents/get_incident.yml", - "get_incident", - _sentinel_inputs(INCIDENT_ID), - ) - - assert url.endswith(f"/incidents/{INCIDENT_ID}") - assert url.count("/subscriptions/") == 1 From 34c84abf646c9a8a924e16047a3d6e84fb6bee79 Mon Sep 17 00:00:00 2001 From: Chris Lo <46541035+topher-lo@users.noreply.github.com> Date: Wed, 5 Aug 2026 17:53:09 -0700 Subject: [PATCH 2/3] test(registry): retain Kubernetes boundary coverage --- packages/tracecat-registry/AGENTS.md | 8 +- tests/registry/test_kubernetes_sdk.py | 257 ++++++++++++++++++++++++++ 2 files changed, 262 insertions(+), 3 deletions(-) create mode 100644 tests/registry/test_kubernetes_sdk.py diff --git a/packages/tracecat-registry/AGENTS.md b/packages/tracecat-registry/AGENTS.md index ac64fbe7a0..6a99cff65c 100644 --- a/packages/tracecat-registry/AGENTS.md +++ b/packages/tracecat-registry/AGENTS.md @@ -60,9 +60,11 @@ do not break their public inputs or outputs without an explicitly planned migrat - Add live or sandbox provider tests only when a reliable environment exists and the user explicitly chooses that coverage during planning. - Generic registry and template validation remains required. Narrow unit tests are - allowed for Tracecat-owned security or protocol mechanics such as credential - isolation, private-method blocking, serialization limits, and reusable pagination - machinery. + allowed for Tracecat-owned platform security or protocol boundaries, such as + credential isolation, preventing host filesystem or subprocess access, blocking + ambient credential discovery, network-target restrictions, and shared protocol + machinery. Provider-local dispatch, validation, pagination, or serialization is + not a platform-boundary exception merely because Tracecat implements it. ## Template design diff --git a/tests/registry/test_kubernetes_sdk.py b/tests/registry/test_kubernetes_sdk.py new file mode 100644 index 0000000000..2b30571c1b --- /dev/null +++ b/tests/registry/test_kubernetes_sdk.py @@ -0,0 +1,257 @@ +from typing import Any + +import pytest +import yaml +from kubernetes import config as kube_config +from tracecat_registry.integrations import kubernetes_sdk + + +def _kubeconfig(**user_overrides: Any) -> str: + user = {"token": "kube-token", **user_overrides} + return yaml.safe_dump( + { + "apiVersion": "v1", + "kind": "Config", + "clusters": [ + { + "name": "cluster", + "cluster": { + "server": "https://kubernetes.example.com", + "certificate-authority-data": "Y2E=", + }, + } + ], + "contexts": [ + { + "name": "ctx", + "context": {"cluster": "cluster", "user": "user"}, + }, + { + "name": "secret-context", + "context": {"cluster": "cluster", "user": "user"}, + }, + ], + "current-context": "ctx", + "users": [{"name": "user", "user": user}], + } + ) + + +def test_call_api_injects_isolated_api_client(monkeypatch) -> None: + """The public wrapper must not instantiate an SDK API with ambient config.""" + calls: dict[str, Any] = {} + + class Configuration: + pass + + class ApiClient: + def __init__(self, *, configuration: Configuration) -> None: + calls["api_client_configuration"] = configuration + + def sanitize_for_serialization(self, result: Any) -> Any: + return result + + class Loader: + def __init__(self, *, config_dict: dict[str, Any], active_context: str) -> None: + calls["config_dict"] = config_dict + calls["active_context"] = active_context + + def load_and_set(self, configuration: Configuration) -> None: + calls["loader_configuration"] = configuration + + class BoundaryApi: + def __init__(self, *, api_client: ApiClient) -> None: + calls["injected_api_client"] = api_client + + def execute(self) -> None: + return None + + monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) + monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) + monkeypatch.setattr( + kubernetes_sdk.client, "BoundaryApi", BoundaryApi, raising=False + ) + monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) + monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) + monkeypatch.setattr( + kubernetes_sdk.secrets, + "get_or_default", + lambda key: "secret-context" if key == "KUBECONFIG_CONTEXT" else None, + ) + + kubernetes_sdk.call_api(api_class="BoundaryApi", method_name="execute") + + assert calls["config_dict"]["users"][0]["user"]["token"] == "kube-token" + assert calls["active_context"] == "secret-context" + assert calls["injected_api_client"].__class__ is ApiClient + assert calls["api_client_configuration"] is calls["loader_configuration"] + + +def test_build_api_client_uses_secret_context(monkeypatch) -> None: + calls: dict[str, Any] = {} + + class Configuration: + pass + + class ApiClient: + def __init__(self, *, configuration: Configuration) -> None: + calls["api_client_configuration"] = configuration + + class Loader: + def __init__( + self, *, config_dict: dict[str, Any], active_context: str | None + ) -> None: + calls["active_context"] = active_context + + def load_and_set(self, configuration: Configuration) -> None: + calls["loader_configuration"] = configuration + + monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) + monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) + monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) + monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) + monkeypatch.setattr( + kubernetes_sdk.secrets, + "get_or_default", + lambda key: "secret-context" if key == "KUBECONFIG_CONTEXT" else None, + ) + + kubernetes_sdk._build_api_client() + + assert calls["active_context"] == "secret-context" + assert calls["api_client_configuration"] is calls["loader_configuration"] + + +@pytest.mark.parametrize("blank_context", ["", " "]) +def test_build_api_client_normalizes_blank_secret_context( + monkeypatch, blank_context: str +) -> None: + """A blank KUBECONFIG_CONTEXT secret must fall back to `current-context`.""" + calls: dict[str, Any] = {} + + class Configuration: + pass + + class ApiClient: + def __init__(self, *, configuration: Configuration) -> None: + pass + + class Loader: + def __init__( + self, *, config_dict: dict[str, Any], active_context: str | None + ) -> None: + calls["active_context"] = active_context + + def load_and_set(self, configuration: Configuration) -> None: + pass + + monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) + monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) + monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) + monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) + monkeypatch.setattr( + kubernetes_sdk.secrets, + "get_or_default", + lambda key: blank_context if key == "KUBECONFIG_CONTEXT" else None, + ) + + kubernetes_sdk._build_api_client() + + assert calls["active_context"] is None + + +@pytest.mark.parametrize( + ("cluster_overrides", "user_overrides", "message"), + [ + ( + {"certificate-authority": "/var/run/secrets/kubernetes.io/ca.crt"}, + {}, + "certificate-authority", + ), + ({}, {"client-certificate": "/tmp/client.crt"}, "client-certificate"), + ({}, {"client-key": "/tmp/client.key"}, "client-key"), + ({}, {"tokenFile": "/var/run/secrets/kubernetes.io/token"}, "tokenFile"), + ({}, {"exec": {"command": "kubectl"}}, "exec"), + ({}, {"auth-provider": {"name": "gcp"}}, "auth-provider"), + ], +) +def test_validate_rejects_file_backed_and_dynamic_credentials( + cluster_overrides: dict[str, Any], + user_overrides: dict[str, Any], + message: str, +) -> None: + config = yaml.safe_load(_kubeconfig(**user_overrides)) + config["clusters"][0]["cluster"].update(cluster_overrides) + + with pytest.raises(ValueError, match=message): + kubernetes_sdk._validate_no_executor_credentials(config) + + +def test_validate_ignores_inactive_context_credentials() -> None: + """Unsafe unused contexts cannot affect the selected isolated context.""" + config = yaml.safe_load(_kubeconfig()) + config["clusters"].append( + { + "name": "other-cluster", + "cluster": { + "server": "https://other.example.com", + "certificate-authority-data": "Y2E=", + }, + } + ) + config["users"].append( + {"name": "other-user", "user": {"exec": {"command": "kubectl"}}} + ) + config["contexts"].append( + { + "name": "other-ctx", + "context": {"cluster": "other-cluster", "user": "other-user"}, + } + ) + + kubernetes_sdk._validate_no_executor_credentials(config) + + with pytest.raises(ValueError, match="exec"): + kubernetes_sdk._validate_no_executor_credentials( + config, active_context="other-ctx" + ) + + +def test_validate_rejects_unknown_active_context() -> None: + config = yaml.safe_load(_kubeconfig()) + with pytest.raises(ValueError, match="not found"): + kubernetes_sdk._validate_no_executor_credentials( + config, active_context="missing" + ) + + +def test_never_calls_ambient_kubernetes_credential_loaders(monkeypatch) -> None: + def fail(*args: Any, **kwargs: Any) -> None: + raise AssertionError("ambient Kubernetes credential loader was called") + + class Configuration: + set_default = staticmethod(fail) + + class ApiClient: + def __init__(self, *, configuration: Configuration) -> None: + pass + + class Loader: + def __init__( + self, *, config_dict: dict[str, Any], active_context: str | None + ) -> None: + pass + + def load_and_set(self, configuration: Configuration) -> None: + pass + + monkeypatch.setattr(kubernetes_sdk.client, "Configuration", Configuration) + monkeypatch.setattr(kubernetes_sdk.client, "ApiClient", ApiClient) + monkeypatch.setattr(kubernetes_sdk, "KubeConfigLoader", Loader) + monkeypatch.setattr(kubernetes_sdk.secrets, "get", lambda key: _kubeconfig()) + monkeypatch.setattr(kubernetes_sdk.secrets, "get_or_default", lambda key: None) + monkeypatch.setattr(kube_config, "load_kube_config", fail, raising=False) + monkeypatch.setattr(kube_config, "new_client_from_config", fail, raising=False) + monkeypatch.setattr(kube_config, "load_incluster_config", fail) + + kubernetes_sdk._build_api_client() From 14a1f36e360d0595b90b63f2f436dfd595f7da49 Mon Sep 17 00:00:00 2001 From: Chris Lo <46541035+topher-lo@users.noreply.github.com> Date: Wed, 5 Aug 2026 18:14:17 -0700 Subject: [PATCH 3/3] test(registry): remove provider-specific test coupling --- tests/unit/test_aws_assume_role.py | 79 +----------------------------- tests/unit/test_dsl_common.py | 46 +++++++++++------ 2 files changed, 32 insertions(+), 93 deletions(-) diff --git a/tests/unit/test_aws_assume_role.py b/tests/unit/test_aws_assume_role.py index 90f6d4032e..cf16029534 100644 --- a/tests/unit/test_aws_assume_role.py +++ b/tests/unit/test_aws_assume_role.py @@ -1,8 +1,7 @@ from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import MagicMock, patch import pytest -import tracecat_registry.integrations.amazon_s3 as amazon_s3 import tracecat_registry.integrations.aws_boto3 as aws_boto3 from tracecat_registry import SecretNotFoundError @@ -293,79 +292,3 @@ async def test_get_session_region_override_takes_precedence() -> None: aws_secret_access_key="secret_test", region_name="ap-south-2", ) - - -class _AsyncAwsClient: - def __init__(self) -> None: - self.calls: list[tuple[str, dict[str, object]]] = [] - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc, tb): - return None - - async def list_buckets(self) -> dict[str, object]: - self.calls.append(("list_buckets", {})) - return {"Buckets": []} - - async def list_objects_v2(self, **kwargs: object) -> dict[str, object]: - self.calls.append(("list_objects_v2", kwargs)) - return {"Contents": []} - - -class _AsyncAwsSession: - def __init__(self, client: _AsyncAwsClient) -> None: - self.client_instance = client - self.client_calls: list[tuple[str, str | None]] = [] - - def client(self, service_name: str, endpoint_url: str | None = None): - self.client_calls.append((service_name, endpoint_url)) - return self.client_instance - - -@pytest.mark.anyio -async def test_call_api_accepts_region_override() -> None: - client = _AsyncAwsClient() - session = _AsyncAwsSession(client) - - with patch.object( - aws_boto3, "get_session", AsyncMock(return_value=session) - ) as get_session: - result = await aws_boto3.call_api( - service_name="s3", - method_name="list_buckets", - endpoint_url="https://s3.example.test", - region_name="custom-region-1", - ) - - get_session.assert_awaited_once_with(region_name="custom-region-1") - assert session.client_calls == [("s3", "https://s3.example.test")] - assert client.calls == [("list_buckets", {})] - assert result == {"Buckets": []} - - -@pytest.mark.anyio -async def test_s3_list_objects_accepts_region_override() -> None: - client = _AsyncAwsClient() - session = _AsyncAwsSession(client) - - with patch.object( - aws_boto3, "get_session", AsyncMock(return_value=session) - ) as get_session: - result = await amazon_s3.list_objects( - bucket="example-bucket", - prefix="logs/", - endpoint_url="https://s3.example.test", - region_name="custom-region-1", - ) - - get_session.assert_awaited_once_with(region_name="custom-region-1") - assert session.client_calls == [("s3", "https://s3.example.test")] - assert client.calls == [ - ( - "list_objects_v2", - {"Bucket": "example-bucket", "Prefix": "logs/", "MaxKeys": 1000}, - ) - ] - assert result == {"Contents": []} diff --git a/tests/unit/test_dsl_common.py b/tests/unit/test_dsl_common.py index fd79776554..79f2ed2719 100644 --- a/tests/unit/test_dsl_common.py +++ b/tests/unit/test_dsl_common.py @@ -3,7 +3,6 @@ from __future__ import annotations import uuid -from pathlib import Path from types import SimpleNamespace from typing import Any, cast @@ -16,8 +15,12 @@ UDFNode, UDFNodeData, ) -from tracecat.expressions.expectations import create_expectation_model -from tracecat.registry.actions.schemas import TemplateAction +from tracecat.expressions.expectations import ExpectedField, create_expectation_model +from tracecat.registry.actions.schemas import ( + ActionStep, + TemplateAction, + TemplateActionDefinition, +) def _make_trigger_node(trigger_id: str | None = None) -> TriggerNode: @@ -134,28 +137,41 @@ def test_normalize_with_multiple_nodes(self) -> None: assert node_ids == {action1_uuid, action2_uuid} -def test_build_action_statements_preserves_sentinel_date_like_api_version() -> None: +def test_build_action_statements_preserves_date_like_string_input() -> None: """Date-like string inputs should stay strings for template arg validation.""" - template = TemplateAction.from_yaml( - Path( - "packages/tracecat-registry/tracecat_registry/templates/tools/" - "microsoft_sentinel/incidents/get_incident.yml" - ) + template = TemplateAction( + type="action", + definition=TemplateActionDefinition( + title="Date-like string input", + description="Synthetic template for input parsing regression coverage.", + name="date_like_string", + namespace="testing", + display_group="Testing", + expects={ + "api_version": ExpectedField( + type="str", + description="A string that YAML can parse as a date.", + ), + }, + steps=[ + ActionStep( + ref="identity", + action="core.transform.reshape", + args={"value": "${{ inputs.api_version }}"}, + ) + ], + returns="${{ steps.identity.result }}", + ), ) expectation_model = create_expectation_model(template.definition.expects) action = cast( Any, SimpleNamespace( id=uuid.uuid4(), - ref="get_incident", + ref="date_like_string", type=template.definition.action, inputs=""" -subscription_id: sub-123 -resource_group_name: rg -workspace_name: workspace -incident_id: incident-123 api_version: 2025-09-01 -base_url: https://management.azure.com """, control_flow={}, upstream_edges=[],