wehub-resource-sync
589 lines
19 KiB
Python
589 lines
19 KiB
Python
import copy
|
|
import sys
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from ray_release.config import (
|
|
CLOUD_ID_TO_NAME,
|
|
_substitute_variable,
|
|
get_test_cloud_name,
|
|
load_schema_file,
|
|
parse_test_definition,
|
|
read_and_validate_release_test_collection,
|
|
validate_cluster_compute,
|
|
validate_test,
|
|
)
|
|
from ray_release.exception import ReleaseTestConfigError
|
|
from ray_release.test import Test
|
|
|
|
_TEST_COLLECTION_FILES = [
|
|
"release/release_tests.yaml",
|
|
"release/release_data_tests.yaml",
|
|
"release/release_multimodal_inference_benchmarks_tests.yaml",
|
|
"release/ray_release/tests/test_collection_data.yaml",
|
|
]
|
|
|
|
VALID_TEST = {
|
|
"name": "validation_test",
|
|
"group": "validation_group",
|
|
"working_dir": "validation_dir",
|
|
"python": "3.10",
|
|
"frequency": "nightly",
|
|
"team": "release",
|
|
"cluster": {
|
|
"byod": {"type": "gpu"},
|
|
"cluster_compute": "tpl_cpu_small.yaml",
|
|
"autosuspend_mins": 10,
|
|
},
|
|
"run": {
|
|
"timeout": 100,
|
|
"script": "python validate.py",
|
|
"wait_for_nodes": {"num_nodes": 2, "timeout": 100},
|
|
"type": "client",
|
|
},
|
|
"smoke_test": {"run": {"timeout": 20}, "frequency": "nightly"},
|
|
"alert": "default",
|
|
}
|
|
|
|
|
|
def test_parse_test_definition():
|
|
"""
|
|
Unit test for the ray_release.config.parse_test_definition function. In particular,
|
|
we check that the code correctly parse a test definition that have the 'variations'
|
|
field.
|
|
"""
|
|
test_definitions = yaml.safe_load(
|
|
"""
|
|
- name: sample_test
|
|
working_dir: sample_dir
|
|
frequency: nightly
|
|
team: sample
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
cluster_compute: compute.yaml
|
|
run:
|
|
timeout: 100
|
|
script: python script.py
|
|
variations:
|
|
- __suffix__: aws
|
|
- __suffix__: gce
|
|
cluster:
|
|
cluster_compute: compute_gce.yaml
|
|
"""
|
|
)
|
|
# Check that parsing returns two tests, one for each variation (aws and gce). Check
|
|
# that both tests are valid, and their fields are populated correctly
|
|
tests = parse_test_definition(test_definitions)
|
|
aws_test = tests[0]
|
|
gce_test = tests[1]
|
|
schema = load_schema_file()
|
|
assert not validate_test(aws_test, schema)
|
|
assert not validate_test(gce_test, schema)
|
|
assert aws_test["name"] == "sample_test.aws"
|
|
assert gce_test["cluster"]["cluster_compute"] == "compute_gce.yaml"
|
|
assert gce_test["cluster"]["byod"]["type"] == "gpu"
|
|
invalid_test_definition = test_definitions[0]
|
|
# Intentionally make the test definition invalid by create an empty 'variations'
|
|
# field. Check that the parser throws exception at runtime
|
|
invalid_test_definition["variations"] = []
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
parse_test_definition([invalid_test_definition])
|
|
# Intentionally make the test definition invalid by making one 'variation' entry
|
|
# missing the __suffix__ entry. Check that the parser throws exception at runtime
|
|
invalid_test_definition["variations"] = [{"__suffix__": "aws"}, {}]
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
parse_test_definition([invalid_test_definition])
|
|
|
|
|
|
def test_parse_test_definition_with_python_version():
|
|
"""
|
|
Unit test for the ray_release.config.parse_test_definition function. In particular,
|
|
we check that the code correctly parse a test definition that have the 'variations' & 'python'
|
|
field.
|
|
"""
|
|
test_definitions = yaml.safe_load(
|
|
"""
|
|
- name: sample_test
|
|
working_dir: sample_dir
|
|
frequency: nightly
|
|
team: sample
|
|
python: "3.10"
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
cluster_compute: compute.yaml
|
|
run:
|
|
timeout: 100
|
|
script: python script.py
|
|
variations:
|
|
- __suffix__: aws
|
|
- __suffix__: gce
|
|
cluster:
|
|
cluster_compute: compute_gce.yaml
|
|
"""
|
|
)
|
|
# Check that parsing returns two tests, one for each variation (aws and gce). Check
|
|
# that both tests are valid, and their fields are populated correctly
|
|
tests = parse_test_definition(test_definitions)
|
|
aws_test = tests[0]
|
|
gce_test = tests[1]
|
|
schema = load_schema_file()
|
|
assert not validate_test(aws_test, schema)
|
|
assert not validate_test(gce_test, schema)
|
|
assert aws_test["name"] == "sample_test.aws"
|
|
assert gce_test["cluster"]["cluster_compute"] == "compute_gce.yaml"
|
|
assert gce_test["cluster"]["byod"]["type"] == "gpu"
|
|
invalid_test_definition = test_definitions[0]
|
|
# Intentionally make the test definition invalid by create an empty 'variations'
|
|
# field. Check that the parser throws exception at runtime
|
|
invalid_test_definition["variations"] = []
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
parse_test_definition([invalid_test_definition])
|
|
# Intentionally make the test definition invalid by making one 'variation' entry
|
|
# missing the __suffix__ entry. Check that the parser throws exception at runtime
|
|
invalid_test_definition["variations"] = [{"__suffix__": "aws"}, {}]
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
parse_test_definition([invalid_test_definition])
|
|
|
|
|
|
def test_parse_test_definition_with_defaults():
|
|
test_definitions = yaml.safe_load(
|
|
"""
|
|
- name: DEFAULTS
|
|
working_dir: default_working_dir
|
|
- name: sample_test_with_default_working_dir
|
|
frequency: nightly
|
|
team: sample
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
cluster_compute: compute.yaml
|
|
run:
|
|
timeout: 100
|
|
script: python script.py
|
|
- name: sample_test_with_overridden_working_dir
|
|
working_dir: overridden_working_dir
|
|
frequency: nightly
|
|
team: sample
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
cluster_compute: compute.yaml
|
|
run:
|
|
timeout: 100
|
|
script: python script.py
|
|
"""
|
|
)
|
|
test_with_default, test_with_override = parse_test_definition(test_definitions)
|
|
schema = load_schema_file()
|
|
assert not validate_test(test_with_default, schema)
|
|
assert not validate_test(test_with_override, schema)
|
|
assert test_with_default["working_dir"] == "default_working_dir"
|
|
assert test_with_override["working_dir"] == "overridden_working_dir"
|
|
|
|
|
|
def test_parse_test_definition_with_matrix_and_variations_raises():
|
|
# Matrix and variations are mutually exclusive.
|
|
test_definitions = yaml.safe_load(
|
|
"""
|
|
- name: test
|
|
frequency: nightly
|
|
team: team
|
|
working_dir: sample_dir
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
cluster_compute: "{{os}}.yaml"
|
|
matrix:
|
|
setup:
|
|
os: [windows, linux]
|
|
run:
|
|
timeout: 100
|
|
script: python script.py
|
|
variations:
|
|
- __suffix__: amd64
|
|
- __suffix__: arm64
|
|
"""
|
|
)
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
parse_test_definition(test_definitions)
|
|
|
|
|
|
def test_parse_test_definition_with_matrix_and_adjustments():
|
|
test_definitions = yaml.safe_load(
|
|
"""
|
|
- name: "test-{{compute}}-{{arg}}"
|
|
matrix:
|
|
setup:
|
|
compute: [fixed, autoscaling]
|
|
arg: [0, 1]
|
|
adjustments:
|
|
- with:
|
|
# Only run arg 2 with fixed compute
|
|
compute: fixed
|
|
arg: 2
|
|
frequency: nightly
|
|
team: team
|
|
working_dir: sample_dir
|
|
cluster:
|
|
byod:
|
|
type: gpu
|
|
runtime_env:
|
|
- SCALING_MODE={{compute}}
|
|
cluster_compute: "{{compute}}.yaml"
|
|
run:
|
|
timeout: 100
|
|
script: python script.py --arg "{{arg}}"
|
|
"""
|
|
)
|
|
tests = parse_test_definition(test_definitions)
|
|
schema = load_schema_file()
|
|
|
|
assert len(tests) == 5 # 4 from matrix, 1 from adjustments
|
|
assert not any(validate_test(test, schema) for test in tests)
|
|
for i, (compute, arg) in enumerate(
|
|
[
|
|
("fixed", 0),
|
|
("fixed", 1),
|
|
("autoscaling", 0),
|
|
("autoscaling", 1),
|
|
("fixed", 2),
|
|
]
|
|
):
|
|
assert tests[i]["name"] == f"test-{compute}-{arg}"
|
|
assert tests[i]["cluster"]["cluster_compute"] == f"{compute}.yaml"
|
|
assert tests[i]["cluster"]["byod"]["runtime_env"] == [f"SCALING_MODE={compute}"]
|
|
|
|
|
|
class TestSubstituteVariable:
|
|
def test_does_not_mutate_original(self):
|
|
test_definition = {"name": "test-{{arg}}"}
|
|
|
|
substituted = _substitute_variable(test_definition, "arg", "1")
|
|
|
|
assert substituted is not test_definition
|
|
assert test_definition == {"name": "test-{{arg}}"}
|
|
|
|
def test_substitute_variable_in_string(self):
|
|
test_definition = {"name": "test-{{arg}}"}
|
|
|
|
substituted = _substitute_variable(test_definition, "arg", "1")
|
|
|
|
assert substituted == {"name": "test-1"}
|
|
|
|
def test_substitute_variable_in_list(self):
|
|
test_definition = {"items": ["item-{{arg}}"]}
|
|
|
|
substituted = _substitute_variable(test_definition, "arg", "1")
|
|
|
|
assert substituted == {"items": ["item-1"]}
|
|
|
|
def test_substitute_variable_in_dict(self):
|
|
test_definition = {"outer": {"inner": "item-{{arg}}"}}
|
|
|
|
substituted = _substitute_variable(test_definition, "arg", "1")
|
|
|
|
assert substituted == {"outer": {"inner": "item-1"}}
|
|
|
|
|
|
def test_schema_validation():
|
|
test = VALID_TEST.copy()
|
|
|
|
schema = load_schema_file()
|
|
|
|
assert not validate_test(Test(**test), schema)
|
|
|
|
# Remove some optional arguments
|
|
del test["alert"]
|
|
del test["python"]
|
|
del test["run"]["wait_for_nodes"]
|
|
del test["cluster"]["autosuspend_mins"]
|
|
|
|
assert not validate_test(Test(**test), schema)
|
|
|
|
# Add some faulty arguments
|
|
|
|
# Faulty frequency
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["frequency"] = "invalid"
|
|
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty job type
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["run"]["type"] = "invalid"
|
|
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty file manager type
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["run"]["file_manager"] = "invalid"
|
|
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty smoke test
|
|
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
del invalid_test["smoke_test"]["frequency"]
|
|
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty Python version
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["python"] = "invalid"
|
|
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty BYOD type
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["cluster"]["byod"]["type"] = "invalid"
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
# Faulty BYOD and Python version match
|
|
invalid_test = Test(**copy.deepcopy(VALID_TEST))
|
|
invalid_test["cluster"]["byod"]["type"] = "gpu"
|
|
invalid_test["python"] = "3.11"
|
|
assert validate_test(invalid_test, schema)
|
|
|
|
|
|
def test_compute_config_invalid_ebs():
|
|
compute_config = {
|
|
"aws": {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {
|
|
"VolumeSize": 1000,
|
|
},
|
|
}
|
|
]
|
|
}
|
|
}
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["aws"]["BlockDeviceMappings"][0]["Ebs"][
|
|
"DeleteOnTermination"
|
|
] = False
|
|
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["aws"]["BlockDeviceMappings"][0]["Ebs"]["DeleteOnTermination"] = True
|
|
|
|
assert not validate_cluster_compute(compute_config)
|
|
|
|
compute_config["head_node_type"] = {}
|
|
compute_config["head_node_type"]["aws_advanced_configurations"] = {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {
|
|
"VolumeSize": 1000,
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["head_node_type"]["aws_advanced_configurations"][
|
|
"BlockDeviceMappings"
|
|
][0]["Ebs"]["DeleteOnTermination"] = False
|
|
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["head_node_type"]["aws_advanced_configurations"][
|
|
"BlockDeviceMappings"
|
|
][0]["Ebs"]["DeleteOnTermination"] = True
|
|
|
|
assert not validate_cluster_compute(compute_config)
|
|
|
|
compute_config["worker_node_types"] = [{}]
|
|
compute_config["worker_node_types"][0]["aws_advanced_configurations"] = {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {
|
|
"VolumeSize": 1000,
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["worker_node_types"][0]["aws_advanced_configurations"][
|
|
"BlockDeviceMappings"
|
|
][0]["Ebs"]["DeleteOnTermination"] = False
|
|
|
|
assert validate_cluster_compute(compute_config)
|
|
|
|
compute_config["worker_node_types"][0]["aws_advanced_configurations"][
|
|
"BlockDeviceMappings"
|
|
][0]["Ebs"]["DeleteOnTermination"] = True
|
|
|
|
assert not validate_cluster_compute(compute_config)
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_valid():
|
|
"""New schema: empty config is valid (all fields optional)."""
|
|
assert not validate_cluster_compute({}, is_new_schema=True)
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_valid_with_fields():
|
|
"""New schema: config with new-schema keys is valid."""
|
|
compute_config = {
|
|
"cloud": "my_cloud",
|
|
"head_node": {"instance_type": "m5.4xlarge"},
|
|
"worker_nodes": [{"instance_type": "m5.xlarge", "min_nodes": 1}],
|
|
}
|
|
assert not validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_rejects_legacy_keys():
|
|
"""New schema: config with legacy keys is rejected."""
|
|
compute_config = {
|
|
"cloud_id": "cld_123",
|
|
"head_node_type": {"instance_type": "m5.4xlarge"},
|
|
}
|
|
error = validate_cluster_compute(compute_config, is_new_schema=True)
|
|
assert error is not None
|
|
assert "legacy schema keys" in error
|
|
assert "anyscale_sdk_2026=true" in error
|
|
|
|
|
|
def test_validate_cluster_compute_legacy_rejects_new_keys():
|
|
"""Legacy schema: config with new-schema keys is rejected."""
|
|
compute_config = {
|
|
"cloud_id": "cld_123",
|
|
"head_node": {"instance_type": "m5.4xlarge"},
|
|
}
|
|
error = validate_cluster_compute(compute_config, is_new_schema=False)
|
|
assert error is not None
|
|
assert "new schema keys" in error
|
|
assert "anyscale_sdk_2026=false" in error
|
|
|
|
|
|
def test_validate_cluster_compute_legacy_rejects_empty():
|
|
"""Legacy schema: empty config is rejected (no legacy keys)."""
|
|
error = validate_cluster_compute({}, is_new_schema=False)
|
|
assert error is not None
|
|
assert "does not have legacy schema keys" in error
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_ebs_top_level():
|
|
"""New schema: EBS DeleteOnTermination is checked in top-level advanced_instance_config."""
|
|
compute_config = {
|
|
"advanced_instance_config": {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {"VolumeSize": 1000},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
# Missing DeleteOnTermination should fail
|
|
assert validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
# Set DeleteOnTermination to True should pass
|
|
compute_config["advanced_instance_config"]["BlockDeviceMappings"][0]["Ebs"][
|
|
"DeleteOnTermination"
|
|
] = True
|
|
assert not validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_ebs_head_node():
|
|
"""New schema: EBS DeleteOnTermination is checked in head_node.advanced_instance_config."""
|
|
compute_config = {
|
|
"head_node": {
|
|
"instance_type": "m5.4xlarge",
|
|
"advanced_instance_config": {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {"VolumeSize": 1000},
|
|
}
|
|
]
|
|
},
|
|
},
|
|
}
|
|
# Missing DeleteOnTermination should fail
|
|
assert validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
# Set DeleteOnTermination to True should pass
|
|
compute_config["head_node"]["advanced_instance_config"]["BlockDeviceMappings"][0][
|
|
"Ebs"
|
|
]["DeleteOnTermination"] = True
|
|
assert not validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
|
|
def test_validate_cluster_compute_new_schema_ebs_worker_nodes():
|
|
"""New schema: EBS checks in worker_nodes[*].advanced_instance_config."""
|
|
compute_config = {
|
|
"worker_nodes": [
|
|
{
|
|
"instance_type": "m5.xlarge",
|
|
"advanced_instance_config": {
|
|
"BlockDeviceMappings": [
|
|
{
|
|
"DeviceName": "/dev/sda1",
|
|
"Ebs": {"VolumeSize": 500},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
}
|
|
assert validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
compute_config["worker_nodes"][0]["advanced_instance_config"][
|
|
"BlockDeviceMappings"
|
|
][0]["Ebs"]["DeleteOnTermination"] = True
|
|
assert not validate_cluster_compute(compute_config, is_new_schema=True)
|
|
|
|
|
|
def test_get_test_cloud_name_from_cluster_cloud():
|
|
"""get_test_cloud_name() returns cluster.cloud when set."""
|
|
test = Test(
|
|
{
|
|
"name": "test",
|
|
"cluster": {"cluster_compute": "tpl.yaml", "cloud": "my_cloud"},
|
|
}
|
|
)
|
|
assert get_test_cloud_name(test) == "my_cloud"
|
|
|
|
|
|
def test_get_test_cloud_name_from_cloud_id_mapping():
|
|
"""get_test_cloud_name() falls back to CLOUD_ID_TO_NAME mapping."""
|
|
for cloud_id, expected_name in CLOUD_ID_TO_NAME.items():
|
|
test = Test(
|
|
{
|
|
"name": "test",
|
|
"cluster": {"cluster_compute": "tpl.yaml", "cloud_id": cloud_id},
|
|
}
|
|
)
|
|
assert get_test_cloud_name(test) == expected_name
|
|
|
|
|
|
def test_get_test_cloud_name_unknown_cloud_id():
|
|
"""get_test_cloud_name() raises ReleaseTestConfigError for unknown cloud_id."""
|
|
test = Test(
|
|
{
|
|
"name": "test",
|
|
"cluster": {"cluster_compute": "tpl.yaml", "cloud_id": "cld_unknown"},
|
|
}
|
|
)
|
|
with pytest.raises(ReleaseTestConfigError):
|
|
get_test_cloud_name(test)
|
|
|
|
|
|
def test_load_and_validate_test_collection_file():
|
|
tests = read_and_validate_release_test_collection(_TEST_COLLECTION_FILES)
|
|
assert [test for test in tests if test.get_name() == "test_name"]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(pytest.main(["-v", __file__]))
|