Skip to content

Commit c1d7a34

Browse files
prhaclaude
prha
and
claude
committed
Add GraphQL tests for partitioned asset check executions
Added comprehensive tests for retrieving asset check executions for partitioned assets via GraphQL: 1. test_partitioned_asset_check_executions - Tests basic partition support for asset checks - Creates check executions for multiple partitions - Verifies both IN_PROGRESS and completed (SUCCEEDED/FAILED) statuses - Validates evaluation metadata and descriptions per partition 2. test_partitioned_asset_check_executions_with_partition_filter - Tests the partition filter parameter in assetCheckExecutions query - Creates checks for different time-based partitions (2024-01, 2024-02) - Verifies filtering works correctly to isolate specific partition executions - Tests both filtered and unfiltered queries These tests ensure the GraphQL API correctly handles partitioned asset checks, which is essential for querying check status on a per-partition basis. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 9346b97 commit c1d7a34

1 file changed

Lines changed: 230 additions & 0 deletions

File tree

‎python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_asset_checks.py‎

Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1443,3 +1443,233 @@ def test_launch_multi_asset_job_without_subset(self, graphql_context: WorkspaceR
14431443
{"name": "my_check", "assetKey": {"path": ["one"]}},
14441444
{"name": "my_other_check", "assetKey": {"path": ["one"]}},
14451445
]
1446+
1447+
def test_partitioned_asset_check_executions(self, graphql_context: WorkspaceRequestContext):
1448+
"""Test retrieving asset check executions for partitioned assets."""
1449+
run_id_one, run_id_two, run_id_three = [make_new_run_id() for _ in range(3)]
1450+
1451+
# Create runs for different partitions
1452+
create_run_for_test(graphql_context.instance, run_id=run_id_one)
1453+
create_run_for_test(graphql_context.instance, run_id=run_id_two)
1454+
create_run_for_test(graphql_context.instance, run_id=run_id_three)
1455+
1456+
# Store planned events for partition "a"
1457+
graphql_context.instance.event_log_storage.store_event(
1458+
_planned_event(
1459+
run_id_one,
1460+
AssetCheckEvaluationPlanned(
1461+
asset_key=AssetKey(["partitioned_asset"]),
1462+
check_name="partitioned_check",
1463+
partition="a",
1464+
),
1465+
)
1466+
)
1467+
1468+
# Store planned events for partition "b"
1469+
graphql_context.instance.event_log_storage.store_event(
1470+
_planned_event(
1471+
run_id_two,
1472+
AssetCheckEvaluationPlanned(
1473+
asset_key=AssetKey(["partitioned_asset"]),
1474+
check_name="partitioned_check",
1475+
partition="b",
1476+
),
1477+
)
1478+
)
1479+
1480+
# Query without partition filter - should get all executions
1481+
res = execute_dagster_graphql(
1482+
graphql_context,
1483+
GET_ASSET_CHECK_HISTORY,
1484+
variables={
1485+
"assetKey": {"path": ["partitioned_asset"]},
1486+
"checkName": "partitioned_check",
1487+
},
1488+
)
1489+
1490+
executions = res.data["assetCheckExecutions"]
1491+
assert len(executions) == 2
1492+
run_ids = {execution["runId"] for execution in executions}
1493+
assert run_ids == {run_id_one, run_id_two}
1494+
1495+
# All should be in progress
1496+
for execution in executions:
1497+
assert execution["status"] == "IN_PROGRESS"
1498+
1499+
# Store evaluation for partition "a" - passed
1500+
evaluation_timestamp_a = time.time()
1501+
graphql_context.instance.event_log_storage.store_event(
1502+
_evaluation_event(
1503+
run_id_one,
1504+
AssetCheckEvaluation(
1505+
asset_key=AssetKey(["partitioned_asset"]),
1506+
check_name="partitioned_check",
1507+
passed=True,
1508+
metadata={"partition": MetadataValue.text("a")},
1509+
severity=AssetCheckSeverity.WARN,
1510+
description="Check passed for partition a",
1511+
),
1512+
timestamp=evaluation_timestamp_a,
1513+
)
1514+
)
1515+
1516+
# Store evaluation for partition "b" - failed
1517+
evaluation_timestamp_b = time.time()
1518+
graphql_context.instance.event_log_storage.store_event(
1519+
_evaluation_event(
1520+
run_id_two,
1521+
AssetCheckEvaluation(
1522+
asset_key=AssetKey(["partitioned_asset"]),
1523+
check_name="partitioned_check",
1524+
passed=False,
1525+
metadata={"partition": MetadataValue.text("b")},
1526+
severity=AssetCheckSeverity.ERROR,
1527+
description="Check failed for partition b",
1528+
),
1529+
timestamp=evaluation_timestamp_b,
1530+
)
1531+
)
1532+
1533+
# Query again - should see completed evaluations
1534+
res = execute_dagster_graphql(
1535+
graphql_context,
1536+
GET_ASSET_CHECK_HISTORY,
1537+
variables={
1538+
"assetKey": {"path": ["partitioned_asset"]},
1539+
"checkName": "partitioned_check",
1540+
},
1541+
)
1542+
1543+
executions = res.data["assetCheckExecutions"]
1544+
# Should have completed executions (with evaluation)
1545+
completed_executions = [e for e in executions if e["evaluation"] is not None]
1546+
assert len(completed_executions) == 2
1547+
1548+
# Find partition a execution
1549+
exec_a = next(e for e in completed_executions if e["runId"] == run_id_one)
1550+
assert exec_a["status"] == "SUCCEEDED"
1551+
assert exec_a["evaluation"]["severity"] == "WARN"
1552+
assert exec_a["evaluation"]["description"] == "Check passed for partition a"
1553+
1554+
# Find partition b execution
1555+
exec_b = next(e for e in completed_executions if e["runId"] == run_id_two)
1556+
assert exec_b["status"] == "FAILED"
1557+
assert exec_b["evaluation"]["severity"] == "ERROR"
1558+
assert exec_b["evaluation"]["description"] == "Check failed for partition b"
1559+
1560+
def test_partitioned_asset_check_executions_with_partition_filter(
1561+
self, graphql_context: WorkspaceRequestContext
1562+
):
1563+
"""Test retrieving asset check executions with partition filter."""
1564+
# Define query with partition filter
1565+
GET_ASSET_CHECK_HISTORY_WITH_PARTITION = """
1566+
query GetAssetChecksQuery($assetKey: AssetKeyInput!, $checkName: String!, $partition: String) {
1567+
assetCheckExecutions(assetKey: $assetKey, checkName: $checkName, limit: 10, partition: $partition) {
1568+
runId
1569+
status
1570+
evaluation {
1571+
severity
1572+
description
1573+
}
1574+
}
1575+
}
1576+
"""
1577+
1578+
run_id_one, run_id_two = [make_new_run_id() for _ in range(2)]
1579+
1580+
create_run_for_test(graphql_context.instance, run_id=run_id_one)
1581+
create_run_for_test(graphql_context.instance, run_id=run_id_two)
1582+
1583+
# Store evaluations for multiple partitions
1584+
graphql_context.instance.event_log_storage.store_event(
1585+
_planned_event(
1586+
run_id_one,
1587+
AssetCheckEvaluationPlanned(
1588+
asset_key=AssetKey(["partitioned_asset"]),
1589+
check_name="partitioned_check",
1590+
partition="2024-01",
1591+
),
1592+
)
1593+
)
1594+
1595+
graphql_context.instance.event_log_storage.store_event(
1596+
_evaluation_event(
1597+
run_id_one,
1598+
AssetCheckEvaluation(
1599+
asset_key=AssetKey(["partitioned_asset"]),
1600+
check_name="partitioned_check",
1601+
passed=True,
1602+
severity=AssetCheckSeverity.WARN,
1603+
description="Check for January",
1604+
),
1605+
)
1606+
)
1607+
1608+
graphql_context.instance.event_log_storage.store_event(
1609+
_planned_event(
1610+
run_id_two,
1611+
AssetCheckEvaluationPlanned(
1612+
asset_key=AssetKey(["partitioned_asset"]),
1613+
check_name="partitioned_check",
1614+
partition="2024-02",
1615+
),
1616+
)
1617+
)
1618+
1619+
graphql_context.instance.event_log_storage.store_event(
1620+
_evaluation_event(
1621+
run_id_two,
1622+
AssetCheckEvaluation(
1623+
asset_key=AssetKey(["partitioned_asset"]),
1624+
check_name="partitioned_check",
1625+
passed=True,
1626+
severity=AssetCheckSeverity.WARN,
1627+
description="Check for February",
1628+
),
1629+
)
1630+
)
1631+
1632+
# Query with partition filter for "2024-01"
1633+
res = execute_dagster_graphql(
1634+
graphql_context,
1635+
GET_ASSET_CHECK_HISTORY_WITH_PARTITION,
1636+
variables={
1637+
"assetKey": {"path": ["partitioned_asset"]},
1638+
"checkName": "partitioned_check",
1639+
"partition": "2024-01",
1640+
},
1641+
)
1642+
1643+
executions = res.data["assetCheckExecutions"]
1644+
assert len(executions) == 1
1645+
assert executions[0]["runId"] == run_id_one
1646+
assert executions[0]["evaluation"]["description"] == "Check for January"
1647+
1648+
# Query with partition filter for "2024-02"
1649+
res = execute_dagster_graphql(
1650+
graphql_context,
1651+
GET_ASSET_CHECK_HISTORY_WITH_PARTITION,
1652+
variables={
1653+
"assetKey": {"path": ["partitioned_asset"]},
1654+
"checkName": "partitioned_check",
1655+
"partition": "2024-02",
1656+
},
1657+
)
1658+
1659+
executions = res.data["assetCheckExecutions"]
1660+
assert len(executions) == 1
1661+
assert executions[0]["runId"] == run_id_two
1662+
assert executions[0]["evaluation"]["description"] == "Check for February"
1663+
1664+
# Query without partition filter - should get both
1665+
res = execute_dagster_graphql(
1666+
graphql_context,
1667+
GET_ASSET_CHECK_HISTORY_WITH_PARTITION,
1668+
variables={
1669+
"assetKey": {"path": ["partitioned_asset"]},
1670+
"checkName": "partitioned_check",
1671+
},
1672+
)
1673+
1674+
executions = res.data["assetCheckExecutions"]
1675+
assert len(executions) == 2

0 commit comments

Comments
 (0)