"""Unit tests for the Dagster integration module. Mirrors the test_rabbitmq.py pattern: config layer + GraphQL client + validation exercised against httpx.MockTransport, no real Dagster instance required. """ from __future__ import annotations import json from typing import Any import httpx import pytest from pydantic import ValidationError from integrations.dagster import ( DagsterConfig, DagsterValidationResult, _compute_run_durations, _extract_step_failures, build_dagster_config, dagster_extract_params, dagster_is_available, get_run_logs, list_assets_with_materialization, list_runs, list_schedule_ticks, list_sensor_ticks, validate_dagster_config, ) from integrations.dagster.client import DagsterClient # --- helpers --------------------------------------------------------------- def _mock_client( handler: object | None = None, *, responses: list[dict[str, Any]] | None = None, raise_on_request: BaseException | None = None, api_token: str = "", ) -> httpx.Client: """Build an httpx.Client whose transport returns canned JSON or raises. Pass either ``handler`` (a callable taking ``httpx.Request`` and returning ``httpx.Response``) OR ``responses`` (a list of dicts; each call returns the next dict as a 200 JSON body). ``raise_on_request`` overrides both and raises the given exception per request. """ if raise_on_request is not None: def _raise(_request: httpx.Request) -> httpx.Response: raise raise_on_request return httpx.Client( transport=httpx.MockTransport(_raise), headers=_headers_for(api_token), ) if handler is None and responses is not None: queue = list(responses) def _from_queue(_request: httpx.Request) -> httpx.Response: payload = queue.pop(0) if queue else {"data": {}} return httpx.Response(200, json=payload) handler = _from_queue assert handler is not None, "must pass handler or responses" return httpx.Client( transport=httpx.MockTransport(handler), # type: ignore[arg-type] headers=_headers_for(api_token), ) def _headers_for(api_token: str) -> dict[str, str]: headers = {"Content-Type": "application/json"} if api_token: headers["Dagster-Cloud-Api-Token"] = api_token return headers def _read_request_body(request: httpx.Request) -> dict[str, Any]: return json.loads(request.content) # --- TestDagsterConfig ----------------------------------------------------- class TestDagsterConfig: def test_endpoint_required(self) -> None: with pytest.raises(ValidationError): DagsterConfig() # type: ignore[call-arg] def test_defaults_api_token_and_integration_id_to_empty(self) -> None: config = DagsterConfig(endpoint="http://localhost:3000") assert config.api_token == "" assert config.integration_id == "" def test_accepts_all_fields(self) -> None: config = DagsterConfig( endpoint="https://acme.eu.dagster.cloud/prod", api_token="t0k3n", integration_id="cloud-prod", ) assert config.endpoint == "https://acme.eu.dagster.cloud/prod" assert config.api_token == "t0k3n" assert config.integration_id == "cloud-prod" class TestBuildDagsterConfig: def test_from_dict_constructs_config(self) -> None: config = build_dagster_config({"endpoint": "http://localhost:3000", "api_token": "abc"}) assert isinstance(config, DagsterConfig) assert config.endpoint == "http://localhost:3000" assert config.api_token == "abc" def test_from_none_raises_validation_error(self) -> None: with pytest.raises(ValidationError): build_dagster_config(None) # --- TestDagsterIsAvailable / TestDagsterExtractParams --------------------- class TestDagsterIsAvailable: def test_empty_sources_returns_false(self) -> None: assert dagster_is_available({}) is False def test_sources_without_endpoint_returns_false(self) -> None: assert dagster_is_available({"dagster": {}}) is False def test_sources_with_endpoint_returns_true(self) -> None: assert dagster_is_available({"dagster": {"endpoint": "http://x"}}) is True class TestDagsterExtractParams: def test_returns_endpoint_and_token(self) -> None: sources = {"dagster": {"endpoint": "http://x", "api_token": "t"}} assert dagster_extract_params(sources) == {"endpoint": "http://x", "api_token": "t"} def test_missing_keys_default_to_empty(self) -> None: assert dagster_extract_params({}) == {"endpoint": "", "api_token": ""} # --- TestDagsterClientUrlNormalisation ------------------------------------- class TestDagsterClientUrlNormalisation: @pytest.mark.parametrize( "raw_endpoint", [ "https://example.com/prod", "https://example.com/prod/", "https://example.com/prod/graphql", "https://example.com/prod/graphql/", ], ) def test_endpoint_collapses_to_canonical_base(self, raw_endpoint: str) -> None: # We use a mock client so no network call happens. Only normalised # state is inspected. client = DagsterClient( endpoint=raw_endpoint, http_client=_mock_client(responses=[{"data": {}}]) ) assert client.endpoint == "https://example.com/prod" assert client._graphql_url == "https://example.com/prod/graphql" # --- TestDagsterClientHeaders ---------------------------------------------- class TestDagsterClientHeaders: def test_token_set_includes_auth_header(self) -> None: # When no http_client is injected, the default httpx.Client carries # our default headers. Inspect them directly. client = DagsterClient(endpoint="http://x", api_token="t0k3n") assert client._client.headers.get("Dagster-Cloud-Api-Token") == "t0k3n" assert client._client.headers.get("Content-Type") == "application/json" def test_token_empty_omits_auth_header(self) -> None: client = DagsterClient(endpoint="http://x") assert "Dagster-Cloud-Api-Token" not in client._client.headers assert client._client.headers.get("Content-Type") == "application/json" # --- TestPing -------------------------------------------------------------- class TestPing: def test_happy_path_returns_data(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client(responses=[{"data": {"version": "1.2.3"}}]), ) result = client.ping() assert result == {"data": {"version": "1.2.3"}} def test_http_500_returns_error(self) -> None: def _handler(_request: httpx.Request) -> httpx.Response: return httpx.Response(500, text="server error") client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) result = client.ping() assert "error" in result assert "HTTP 500" in result["error"] assert "server error" in result["error"] def test_network_error_returns_error(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client(raise_on_request=httpx.ConnectError("refused")), ) result = client.ping() assert "error" in result assert "Request to Dagster failed" in result["error"] def test_non_json_response_returns_error(self) -> None: def _handler(_request: httpx.Request) -> httpx.Response: return httpx.Response(200, text="not json") client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) result = client.ping() assert "error" in result assert "Invalid JSON" in result["error"] def test_graphql_errors_field_returns_error(self) -> None: def _handler(_request: httpx.Request) -> httpx.Response: return httpx.Response( 200, json={"errors": [{"message": "Field 'version' is not allowed"}]} ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) result = client.ping() assert "error" in result assert "Field 'version' is not allowed" in result["error"] # --- TestListRuns ---------------------------------------------------------- class TestListRuns: def test_sends_query_with_limit_no_status(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "runsOrError": { "__typename": "Runs", "results": [], "count": 0, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) result = client.list_runs(limit=5) assert "data" in result # Variables: limit set, statuses absent variables = captured[0]["variables"] assert variables == {"limit": 5} def test_sends_status_filter_as_single_element_list(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response(200, json={"data": {"runsOrError": {"__typename": "Runs"}}}) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.list_runs(limit=10, status="FAILURE") variables = captured[0]["variables"] assert variables == {"limit": 10, "statuses": ["FAILURE"]} def test_invalid_filter_union_member_propagates(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "runsOrError": { "__typename": "InvalidPipelineRunsFilterError", "message": "bad filter", } } } ] ), ) result = client.list_runs(limit=1) assert result["data"]["runsOrError"]["__typename"] == "InvalidPipelineRunsFilterError" def test_python_error_union_member_propagates(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "runsOrError": { "__typename": "PythonError", "message": "internal failure", } } } ] ), ) result = client.list_runs(limit=1) assert result["data"]["runsOrError"]["__typename"] == "PythonError" def test_sends_pipeline_name_filter_when_job_name_passed(self) -> None: """``job_name`` is sent as ``RunsFilter.pipelineName`` (Dagster's back-compat alias for jobName) so list_runs scopes to one pipeline.""" captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response(200, json={"data": {"runsOrError": {"__typename": "Runs"}}}) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.list_runs(limit=5, job_name="user_analytics_pipeline") variables = captured[0]["variables"] assert variables == { "limit": 5, "pipelineName": "user_analytics_pipeline", } def test_sends_pipeline_name_alongside_status_when_both_passed(self) -> None: """job_name and status are independent dimensions of the filter; both must propagate when set together.""" captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response(200, json={"data": {"runsOrError": {"__typename": "Runs"}}}) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.list_runs(limit=10, status="FAILURE", job_name="user_analytics_pipeline") variables = captured[0]["variables"] assert variables == { "limit": 10, "statuses": ["FAILURE"], "pipelineName": "user_analytics_pipeline", } # --- TestGetRunLogs -------------------------------------------------------- class TestGetRunLogs: def test_sends_runid_variable(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "logsForRun": { "__typename": "EventConnection", "events": [], "cursor": None, "hasMore": False, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.get_run_logs(run_id="abc-123") variables = captured[0]["variables"] assert variables["runId"] == "abc-123" def test_sends_after_cursor_variable_when_cursor_passed(self) -> None: """Pin the GraphQL variable name: Dagster's logsForRun field expects ``afterCursor``, not ``cursor``. A wrong name silently passes our mock-based pagination tests but fails against a real Dagster server with ``Unknown argument 'cursor' on field 'logsForRun'``.""" captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "logsForRun": { "__typename": "EventConnection", "events": [], "cursor": None, "hasMore": False, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.get_run_logs(run_id="abc-123", cursor="page-2-cursor") variables = captured[0]["variables"] # Must be `afterCursor` per Dagster's schema, never `cursor`. assert variables.get("afterCursor") == "page-2-cursor" assert "cursor" not in variables def test_run_not_found_union_member_propagates(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "logsForRun": { "__typename": "RunNotFoundError", "message": "Run not found", } } } ] ), ) result = client.get_run_logs(run_id="missing") assert result["data"]["logsForRun"]["__typename"] == "RunNotFoundError" # --- TestExtractStepFailures ------------------------------------------------- class TestExtractStepFailures: """The integration enriches get_run_logs responses with a structured `summary.failures` list so the agent's diagnosis stage can identify multi-failure runs without walking the raw event list.""" def test_empty_events_yields_zero_failure_count(self) -> None: summary = _extract_step_failures({"events": []}) assert summary == {"failure_count": 0, "failures": []} def test_non_failure_events_are_filtered_out(self) -> None: events = [ {"__typename": "RunStartEvent", "stepKey": None}, {"__typename": "StepWorkerStartedEvent", "stepKey": "op_a"}, {"__typename": "EngineEvent", "stepKey": None}, ] summary = _extract_step_failures({"events": events}) assert summary == {"failure_count": 0, "failures": []} def test_single_step_failure_extracts_cause(self) -> None: events = [ { "__typename": "ExecutionStepFailureEvent", "stepKey": "failing_op", "timestamp": "1779906298920", "error": { "className": "DagsterExecutionStepExecutionError", "cause": { "className": "ValueError", "message": "ValueError: intentional failure", }, }, } ] summary = _extract_step_failures({"events": events}) assert summary["failure_count"] == 1 assert summary["failures"] == [ { "step_key": "failing_op", "timestamp": "1779906298920", "wrapper_class": "DagsterExecutionStepExecutionError", "exception_class": "ValueError", "cause_message": "ValueError: intentional failure", } ] def test_multiple_concurrent_step_failures_all_surfaced(self) -> None: """Multi-step parallel failure case: BOTH step failures must appear in the summary, not just the first one. This is the agent-side clarity our integration owes for parallel-execution runs.""" events = [ { "__typename": "ExecutionStepFailureEvent", "stepKey": "extract_orders", "timestamp": "1", "error": { "className": "DagsterExecutionStepExecutionError", "cause": { "className": "TimeoutError", "message": "Postgres pool exhausted", }, }, }, { "__typename": "ExecutionStepFailureEvent", "stepKey": "extract_customers", "timestamp": "2", "error": { "className": "DagsterExecutionStepExecutionError", "cause": { "className": "PermissionError", "message": "Cassandra denied", }, }, }, { "__typename": "RunFailureEvent", "stepKey": None, "error": None, }, ] summary = _extract_step_failures({"events": events}) assert summary["failure_count"] == 2 step_keys = [f["step_key"] for f in summary["failures"]] assert step_keys == ["extract_orders", "extract_customers"] classes = [f["exception_class"] for f in summary["failures"]] assert classes == ["TimeoutError", "PermissionError"] def test_run_failure_event_with_null_error_is_not_a_step_failure(self) -> None: """RunFailureEvent (run-level termination) is distinct from ExecutionStepFailureEvent (step-level exception). Run-level events often have ``error: null`` and carry no cause; the summary should only count step-level failures since those have the actionable user-code exception.""" events = [{"__typename": "RunFailureEvent", "stepKey": None, "error": None}] summary = _extract_step_failures({"events": events}) assert summary["failure_count"] == 0 def test_step_failure_event_with_null_error_yields_null_fields(self) -> None: """Defensive path: ``ExecutionStepFailureEvent`` with ``error: None`` still counts as a failure; the wrapper/exception/cause fields fall back to None.""" events = [ { "__typename": "ExecutionStepFailureEvent", "stepKey": "failing_op", "timestamp": "1", "error": None, } ] summary = _extract_step_failures({"events": events}) assert summary["failure_count"] == 1 assert summary["failures"][0] == { "step_key": "failing_op", "timestamp": "1", "wrapper_class": None, "exception_class": None, "cause_message": None, } def test_get_run_logs_response_includes_summary_on_event_connection( self, monkeypatch: pytest.MonkeyPatch ) -> None: """The end-to-end integration helper enriches the EventConnection response with a top-level summary block.""" client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [ { "__typename": "ExecutionStepFailureEvent", "stepKey": "op_x", "timestamp": "1", "error": { "className": "DagsterExecutionStepExecutionError", "cause": { "className": "RuntimeError", "message": "boom", }, }, } ], "cursor": None, "hasMore": False, } } } ] ), ) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(dagster_module, "_client", lambda _cfg: client) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="abc") assert "summary" in result assert result["summary"]["failure_count"] == 1 assert result["summary"]["failures"][0]["exception_class"] == "RuntimeError" def test_get_run_logs_response_skips_summary_on_error_union_members( self, monkeypatch: pytest.MonkeyPatch ) -> None: """When logsForRun returns RunNotFoundError or PythonError instead of EventConnection, there is no event list to summarize. The response should NOT add a misleading empty summary block.""" client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "logsForRun": { "__typename": "RunNotFoundError", "message": "Run not found", } } } ] ), ) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(dagster_module, "_client", lambda _cfg: client) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="missing") assert "summary" not in result def test_get_run_logs_paginates_until_has_more_false( self, monkeypatch: pytest.MonkeyPatch ) -> None: """A run spanning multiple pages: every page's failures must roll up into the summary""" def _failure_event(step_key: str, exc_class: str) -> dict[str, Any]: return { "__typename": "ExecutionStepFailureEvent", "stepKey": step_key, "timestamp": "1", "error": { "className": "DagsterExecutionStepExecutionError", "cause": {"className": exc_class, "message": f"{exc_class} on {step_key}"}, }, } pages = [ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [_failure_event("op_a", "TimeoutError")], "cursor": "page-2-cursor", "hasMore": True, } } }, { "data": { "logsForRun": { "__typename": "EventConnection", "events": [_failure_event("op_b", "PermissionError")], "cursor": "end", "hasMore": False, } } }, ] captured_cursors: list[str | None] = [] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: captured_cursors.append(cursor) return pages.pop(0) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="run-1") # Pagination drove 2 calls, with the second using the cursor from the first. assert captured_cursors == [None, "page-2-cursor"] # Both failures are present in the aggregated summary. assert result["summary"]["failure_count"] == 2 assert [f["step_key"] for f in result["summary"]["failures"]] == ["op_a", "op_b"] # Truncation flag is false when we reached has_more=False naturally. assert result["summary"]["truncated"] is False assert result["summary"]["events_examined"] == 2 def test_get_run_logs_caps_non_failure_events_but_keeps_late_failures( self, monkeypatch: pytest.MonkeyPatch ) -> None: """Load-bearing guarantee: a failure that arrives AFTER the non-failure event cap is hit must still be collected and surfaced in the summary.""" from integrations.dagster import MAX_NON_FAILURE_RUN_LOG_EVENTS def _benign(i: int) -> dict[str, Any]: return {"__typename": "EngineEvent", "stepKey": None, "timestamp": str(i)} late_failure = { "__typename": "ExecutionStepFailureEvent", "stepKey": "deep_step", "timestamp": "9999", "error": { "className": "DagsterExecutionStepExecutionError", "cause": {"className": "RuntimeError", "message": "found past the cap"}, }, } # Page 1: enough benign events to fill the cap exactly. # Page 2: one more benign (discarded — cap reached) plus the late failure. pages = [ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [_benign(i) for i in range(MAX_NON_FAILURE_RUN_LOG_EVENTS)], "cursor": "page-2", "hasMore": True, } } }, { "data": { "logsForRun": { "__typename": "EventConnection", "events": [_benign(99999), late_failure], "cursor": "end", "hasMore": False, } } }, ] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return pages.pop(0) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="late-fail") # Failure past the non-failure cap was captured. assert result["summary"]["failure_count"] == 1 assert result["summary"]["failures"][0]["step_key"] == "deep_step" # Truncation flag still raised because non-failure data is incomplete. assert result["summary"]["truncated"] is True # Aggregated events = capped non-failure (1500) + all failures (1) = 1501. assert result["summary"]["events_examined"] == MAX_NON_FAILURE_RUN_LOG_EVENTS + 1 def test_get_run_logs_sliding_window_drops_oldest_non_failures( self, monkeypatch: pytest.MonkeyPatch ) -> None: """When more non-failure events arrive than the window can hold, the OLDEST events are evicted (FIFO). This keeps the surviving context adjacent to failures (which typically land later in the stream), rather than at the run's chronological start.""" from integrations.dagster import MAX_NON_FAILURE_RUN_LOG_EVENTS def _benign(i: int) -> dict[str, Any]: return {"__typename": "EngineEvent", "stepKey": None, "timestamp": str(i)} # Page 1 fills the window exactly; page 2 adds 2 more, forcing the # 2 oldest to evict. pages = [ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [_benign(i) for i in range(MAX_NON_FAILURE_RUN_LOG_EVENTS)], "cursor": "page-2", "hasMore": True, } } }, { "data": { "logsForRun": { "__typename": "EventConnection", "events": [ _benign(MAX_NON_FAILURE_RUN_LOG_EVENTS), _benign(MAX_NON_FAILURE_RUN_LOG_EVENTS + 1), ], "cursor": "end", "hasMore": False, } } }, ] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return pages.pop(0) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="sliding") events = result["data"]["logsForRun"]["events"] timestamps = [e["timestamp"] for e in events] # The kept window is exactly the cap size. assert len(events) == MAX_NON_FAILURE_RUN_LOG_EVENTS # Two newest events are present (most recent context preserved). assert timestamps[-1] == str(MAX_NON_FAILURE_RUN_LOG_EVENTS + 1) assert timestamps[-2] == str(MAX_NON_FAILURE_RUN_LOG_EVENTS) # Two oldest events were evicted from the window. assert "0" not in timestamps assert "1" not in timestamps # Truncation flag set because the window overflowed. assert result["summary"]["truncated"] is True def test_get_run_logs_aggregated_events_sorted_chronologically( self, monkeypatch: pytest.MonkeyPatch ) -> None: """Aggregated events must preserve chronological order even when a non-failure event (e.g. a downstream skip) lands AFTER the upstream failure that caused it.""" page = { "data": { "logsForRun": { "__typename": "EventConnection", "events": [ # T=100: upstream step started {"__typename": "EngineEvent", "stepKey": "start", "timestamp": "100"}, # T=200: upstream step FAILED (kept in failure_events) { "__typename": "ExecutionStepFailureEvent", "stepKey": "upstream_op", "timestamp": "200", "error": { "className": "DagsterExecutionStepExecutionError", "cause": {"className": "ValueError", "message": "boom"}, }, }, # T=300: downstream step skipped (kept in non_failure_events) { "__typename": "EngineEvent", "stepKey": "downstream_op", "timestamp": "300", }, ], "cursor": "end", "hasMore": False, } } } def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return page from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="ordering") events = result["data"]["logsForRun"]["events"] timestamps = [int(e["timestamp"]) for e in events] # Must be chronological (100, 200, 300), NOT non-failures-first (100, 300, 200). assert timestamps == [100, 200, 300] def test_get_run_logs_mid_pagination_error_preserves_collected_failures( self, monkeypatch: pytest.MonkeyPatch ) -> None: """If page N>1 returns an error, failures already collected from earlier pages should be preserved.""" failure_event = { "__typename": "ExecutionStepFailureEvent", "stepKey": "page_1_op", "timestamp": "100", "error": { "className": "DagsterExecutionStepExecutionError", "cause": {"className": "ValueError", "message": "boom"}, }, } pages = [ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [failure_event], "cursor": "page-2", "hasMore": True, } } }, {"error": "Request to Dagster failed: timeout after 10s"}, ] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return pages.pop(0) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="partial") # Page-1 failure preserved assert result["summary"]["failure_count"] == 1 assert result["summary"]["failures"][0]["step_key"] == "page_1_op" # Truncation flag set because fetch was incomplete. assert result["summary"]["truncated"] is True # fetch_error signals partial fetch with the original error message. assert result["summary"]["fetch_error"] == ("Request to Dagster failed: timeout after 10s") def test_get_run_logs_first_page_error_still_propagates_unchanged( self, monkeypatch: pytest.MonkeyPatch ) -> None: """First-page errors have no accumulated data to save. so the raw envelope propagates unchanged.""" def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return {"error": "Request to Dagster failed: connection refused"} from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="dead") assert result == {"error": "Request to Dagster failed: connection refused"} assert "summary" not in result def test_get_run_logs_mid_pagination_non_event_response_preserves_collected_failures( self, monkeypatch: pytest.MonkeyPatch ) -> None: """If page N>1 returns a non-EventConnection failures already collected from earlier pages should be preserved.""" failure_event = { "__typename": "ExecutionStepFailureEvent", "stepKey": "page_1_op", "timestamp": "100", "error": { "className": "DagsterExecutionStepExecutionError", "cause": {"className": "ValueError", "message": "boom"}, }, } pages = [ { "data": { "logsForRun": { "__typename": "EventConnection", "events": [failure_event], "cursor": "page-2", "hasMore": True, } } }, { "data": { "logsForRun": { "__typename": "RunNotFoundError", "message": "Run not found", } } }, ] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: return pages.pop(0) from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="vanished") # Page-1 failure preserved assert result["summary"]["failure_count"] == 1 assert result["summary"]["failures"][0]["step_key"] == "page_1_op" assert result["summary"]["truncated"] is True # fetch_error names the unexpected typename. assert "RunNotFoundError" in result["summary"]["fetch_error"] assert "page 2" in result["summary"]["fetch_error"] def test_get_run_logs_page_cap_safety_net(self, monkeypatch: pytest.MonkeyPatch) -> None: """A server that says ``hasMore=true`` forever must terminate at ``MAX_RUN_LOG_PAGES``; ``truncated`` is set to signal incomplete fetch.""" from integrations.dagster import MAX_RUN_LOG_PAGES pages_served = [0] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: pages_served[0] += 1 return { "data": { "logsForRun": { "__typename": "EventConnection", "events": [ {"__typename": "EngineEvent", "stepKey": None, "timestamp": "1"} ], "cursor": f"page-{pages_served[0]}", "hasMore": True, } } } from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="runaway") assert pages_served[0] == MAX_RUN_LOG_PAGES assert result["summary"]["truncated"] is True assert result["data"]["logsForRun"]["hasMore"] is True def test_get_run_logs_exits_when_has_more_true_but_cursor_missing( self, monkeypatch: pytest.MonkeyPatch ) -> None: """Defensive: a buggy server that says ``hasMore=true`` but returns no cursor should exit the loop safely and signal partial fetch""" called = [0] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: called[0] += 1 return { "data": { "logsForRun": { "__typename": "EventConnection", "events": [], "cursor": None, "hasMore": True, } } } from integrations import dagster as dagster_module from integrations.dagster import get_run_logs as helper_get_run_logs monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) monkeypatch.setattr( dagster_module, "_client", lambda cfg: DagsterClient(endpoint=cfg.endpoint) ) result = helper_get_run_logs(DagsterConfig(endpoint="http://x"), run_id="buggy") # Loop exited safely after one call (no infinite loop). assert called[0] == 1 # Truncation signal raised because pagination ended mid-stream. assert result["summary"]["truncated"] is True assert "hasMore=true" in result["summary"]["fetch_error"] assert "no cursor" in result["summary"]["fetch_error"] # --- TestComputeRunDurations ----------------------------------------------- class TestComputeRunDurations: """The integration adds ``duration_seconds`` per run row so the agent does not have to derive it from ``startTime``/``endTime`` itself.""" def test_completed_run_yields_finite_duration(self) -> None: result = { "data": { "runsOrError": { "__typename": "Runs", "results": [ {"runId": "r1", "startTime": 100.0, "endTime": 142.5}, ], } } } enriched = _compute_run_durations(result) assert enriched["data"]["runsOrError"]["results"][0]["duration_seconds"] == 42.5 def test_in_flight_run_yields_null_duration(self) -> None: result = { "data": { "runsOrError": { "__typename": "Runs", "results": [ {"runId": "r2", "startTime": 100.0, "endTime": None}, ], } } } enriched = _compute_run_durations(result) assert enriched["data"]["runsOrError"]["results"][0]["duration_seconds"] is None def test_queued_run_with_no_timestamps_yields_null(self) -> None: result = { "data": { "runsOrError": { "__typename": "Runs", "results": [ {"runId": "r3", "startTime": None, "endTime": None}, ], } } } enriched = _compute_run_durations(result) assert enriched["data"]["runsOrError"]["results"][0]["duration_seconds"] is None def test_error_union_member_skipped(self) -> None: """``InvalidPipelineRunsFilterError`` and ``PythonError`` have no ``results`` array; the enrichment is a no-op.""" result = { "data": { "runsOrError": { "__typename": "InvalidPipelineRunsFilterError", "message": "bad filter", } } } enriched = _compute_run_durations(result) assert enriched == result def test_python_error_union_member_is_no_op(self) -> None: result = { "data": { "runsOrError": { "__typename": "PythonError", "message": "internal failure", } } } enriched = _compute_run_durations(result) assert enriched == result def test_multi_row_response_enriches_every_row(self) -> None: result = { "data": { "runsOrError": { "__typename": "Runs", "results": [ {"runId": "r1", "startTime": 100.0, "endTime": 150.0}, {"runId": "r2", "startTime": 200.0, "endTime": None}, {"runId": "r3", "startTime": 300.0, "endTime": 305.5}, ], } } } enriched = _compute_run_durations(result) durations = [r["duration_seconds"] for r in enriched["data"]["runsOrError"]["results"]] assert durations == [50.0, None, 5.5] # --- TestListAssetsWithMaterialization ------------------------------------- class TestListAssetsWithMaterialization: def test_sends_limit_variable_and_parses_asset_connection(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "assetsOrError": { "__typename": "AssetConnection", "nodes": [], "cursor": None, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) result = client.list_assets_with_materialization(limit=7) assert captured[0]["variables"] == {"limit": 7} assert result["data"]["assetsOrError"]["__typename"] == "AssetConnection" # --- TestListSensorTicks --------------------------------------------------- class TestListSensorTicks: def test_sends_sensor_selector_with_all_three_coordinates(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "sensorOrError": { "__typename": "Sensor", "name": "my_sensor", "sensorState": {"ticks": []}, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.list_sensor_ticks( repository_name="repo_x", repository_location_name="loc_y", sensor_name="my_sensor", limit=3, ) variables = captured[0]["variables"] assert variables["sensorSelector"] == { "repositoryName": "repo_x", "repositoryLocationName": "loc_y", "sensorName": "my_sensor", } assert variables["limit"] == 3 def test_sensor_not_found_union_member_propagates(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "sensorOrError": { "__typename": "SensorNotFoundError", "message": "Sensor not found", } } } ] ), ) result = client.list_sensor_ticks( repository_name="r", repository_location_name="l", sensor_name="missing", ) assert result["data"]["sensorOrError"]["__typename"] == "SensorNotFoundError" # --- TestListScheduleTicks ------------------------------------------------- class TestListScheduleTicks: def test_sends_schedule_selector_with_all_three_coordinates(self) -> None: captured: list[dict[str, Any]] = [] def _handler(request: httpx.Request) -> httpx.Response: captured.append(_read_request_body(request)) return httpx.Response( 200, json={ "data": { "scheduleOrError": { "__typename": "Schedule", "name": "my_schedule", "scheduleState": {"ticks": []}, } } }, ) client = DagsterClient(endpoint="http://x", http_client=_mock_client(handler=_handler)) client.list_schedule_ticks( repository_name="repo_x", repository_location_name="loc_y", schedule_name="my_schedule", limit=4, ) variables = captured[0]["variables"] assert variables["scheduleSelector"] == { "repositoryName": "repo_x", "repositoryLocationName": "loc_y", "scheduleName": "my_schedule", } assert variables["limit"] == 4 def test_schedule_not_found_union_member_propagates(self) -> None: client = DagsterClient( endpoint="http://x", http_client=_mock_client( responses=[ { "data": { "scheduleOrError": { "__typename": "ScheduleNotFoundError", "message": "Schedule not found", } } } ] ), ) result = client.list_schedule_ticks( repository_name="r", repository_location_name="l", schedule_name="missing", ) assert result["data"]["scheduleOrError"]["__typename"] == "ScheduleNotFoundError" # --- TestValidateDagsterConfig --------------------------------------------- class TestValidateDagsterConfig: def test_missing_endpoint_returns_not_ok(self) -> None: config = DagsterConfig(endpoint="") result = validate_dagster_config(config) assert isinstance(result, DagsterValidationResult) assert result.ok is False assert "endpoint is required" in result.detail.lower() def test_happy_path_returns_ok_with_version(self, monkeypatch: pytest.MonkeyPatch) -> None: original_init = DagsterClient.__init__ def patched_init( self: DagsterClient, endpoint: str, api_token: str = "", timeout_s: int = 10 ) -> None: original_init( self, endpoint=endpoint, api_token=api_token, timeout_s=timeout_s, http_client=_mock_client(responses=[{"data": {"version": "1.13.6"}}]), ) monkeypatch.setattr(DagsterClient, "__init__", patched_init) result = validate_dagster_config(DagsterConfig(endpoint="http://x")) assert result.ok is True assert "1.13.6" in result.detail def test_probe_failure_returns_not_ok(self, monkeypatch: pytest.MonkeyPatch) -> None: original_init = DagsterClient.__init__ def patched_init( self: DagsterClient, endpoint: str, api_token: str = "", timeout_s: int = 10 ) -> None: original_init( self, endpoint=endpoint, api_token=api_token, timeout_s=timeout_s, http_client=_mock_client(raise_on_request=httpx.ConnectError("refused")), ) monkeypatch.setattr(DagsterClient, "__init__", patched_init) result = validate_dagster_config(DagsterConfig(endpoint="http://x")) assert result.ok is False assert "probe failed" in result.detail.lower() def test_response_without_version_field_returns_not_ok( self, monkeypatch: pytest.MonkeyPatch ) -> None: original_init = DagsterClient.__init__ def patched_init( self: DagsterClient, endpoint: str, api_token: str = "", timeout_s: int = 10 ) -> None: original_init( self, endpoint=endpoint, api_token=api_token, timeout_s=timeout_s, http_client=_mock_client(responses=[{"data": {}}]), ) monkeypatch.setattr(DagsterClient, "__init__", patched_init) result = validate_dagster_config(DagsterConfig(endpoint="http://x")) assert result.ok is False assert "did not return a version" in result.detail.lower() # --- TestIntegrationHelpers ------------------------------------------------ class TestIntegrationHelpers: """The list_runs / get_run_logs / list_assets / list_sensor_ticks helpers in integrations.dagster are thin wrappers that build a DagsterClient and call through. Cover that the routing works.""" def test_list_runs_routes_to_client(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict[str, Any]] = [] def fake_list_runs( self: DagsterClient, *, limit: int = 25, status: str | None = None, job_name: str | None = None, ) -> dict[str, Any]: calls.append({"limit": limit, "status": status, "job_name": job_name}) return {"data": {"routed": True}} monkeypatch.setattr(DagsterClient, "list_runs", fake_list_runs) result = list_runs( DagsterConfig(endpoint="http://x"), limit=3, status="FAILURE", job_name="user_analytics_pipeline", ) assert calls == [{"limit": 3, "status": "FAILURE", "job_name": "user_analytics_pipeline"}] assert result == {"data": {"routed": True}} def test_get_run_logs_routes_to_client(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict[str, Any]] = [] def fake_get_run_logs( self: DagsterClient, *, run_id: str, limit: int = 250, cursor: str | None = None, ) -> dict[str, Any]: calls.append({"run_id": run_id, "limit": limit, "cursor": cursor}) return {"data": {"routed": True}} monkeypatch.setattr(DagsterClient, "get_run_logs", fake_get_run_logs) result = get_run_logs(DagsterConfig(endpoint="http://x"), run_id="abc") # The first (and only) call has cursor=None and the default page size. assert calls == [{"run_id": "abc", "limit": 250, "cursor": None}] # Non-EventConnection payload propagates as-is, no pagination. assert result == {"data": {"routed": True}} def test_list_assets_routes_to_client(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict[str, Any]] = [] def fake(self: DagsterClient, *, limit: int = 25) -> dict[str, Any]: calls.append({"limit": limit}) return {"data": {"routed": True}} monkeypatch.setattr(DagsterClient, "list_assets_with_materialization", fake) result = list_assets_with_materialization(DagsterConfig(endpoint="http://x"), limit=4) assert calls == [{"limit": 4}] assert result == {"data": {"routed": True}} def test_list_sensor_ticks_routes_to_client(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict[str, Any]] = [] def fake( self: DagsterClient, *, repository_name: str, repository_location_name: str, sensor_name: str, limit: int = 25, ) -> dict[str, Any]: calls.append( { "repository_name": repository_name, "repository_location_name": repository_location_name, "sensor_name": sensor_name, "limit": limit, } ) return {"data": {"routed": True}} monkeypatch.setattr(DagsterClient, "list_sensor_ticks", fake) result = list_sensor_ticks( DagsterConfig(endpoint="http://x"), repository_name="r", repository_location_name="loc", sensor_name="s", limit=2, ) assert calls == [ { "repository_name": "r", "repository_location_name": "loc", "sensor_name": "s", "limit": 2, } ] assert result == {"data": {"routed": True}} def test_list_schedule_ticks_routes_to_client(self, monkeypatch: pytest.MonkeyPatch) -> None: calls: list[dict[str, Any]] = [] def fake( self: DagsterClient, *, repository_name: str, repository_location_name: str, schedule_name: str, limit: int = 25, ) -> dict[str, Any]: calls.append( { "repository_name": repository_name, "repository_location_name": repository_location_name, "schedule_name": schedule_name, "limit": limit, } ) return {"data": {"routed": True}} monkeypatch.setattr(DagsterClient, "list_schedule_ticks", fake) result = list_schedule_ticks( DagsterConfig(endpoint="http://x"), repository_name="r", repository_location_name="loc", schedule_name="s", limit=2, ) assert calls == [ { "repository_name": "r", "repository_location_name": "loc", "schedule_name": "s", "limit": 2, } ] assert result == {"data": {"routed": True}}