test_basic.py

# Smoke tests for SkyPilot for basic functionality
# Default options are set in pyproject.toml
# Example usage:
# Run all tests except for AWS and Lambda Cloud
# > pytest tests/smoke_tests/test_basic.py
#
# Terminate failed clusters after test finishes
# > pytest tests/smoke_tests/test_basic.py --terminate-on-failure
#
# Re-run last failed tests
# > pytest --lf
#
# Run one of the smoke tests
# > pytest tests/smoke_tests/test_basic.py::test_minimal
#
# Only run test for AWS + generic tests
# > pytest tests/smoke_tests/test_basic.py --aws
#
# Change cloud for generic tests to aws
# > pytest tests/smoke_tests/test_basic.py --generic-cloud aws

importjson
importos
importpathlib
importsubprocess
importsys
importtempfile
importtextwrap
importtime

importpytest
fromsmoke_testsimportsmoke_tests_utils

importsky
fromskyimportskypilot_config
fromsky.cloudsimportLambda
fromsky.skyletimportconstants
fromsky.skyletimportevents
fromsky.utilsimportcommon_utils


# ---------- Dry run: 2 Tasks in a chain. ----------
@pytest.mark.no_vast#requires GCP and AWS set up
@pytest.mark.no_fluidstack#requires GCP and AWS set up
deftest_example_app():
test=smoke_tests_utils.Test(
'example_app',
 ['python examples/example_app.py'],
 )
smoke_tests_utils.run_one_test(test)


# ---------- A minimal task ----------
deftest_minimal(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'minimal',
 [
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --cloud {generic_cloud}{smoke_tests_utils.LOW_RESOURCE_ARG} tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
# Output validation done.
f'sky logs {name} 1 --status',
f'sky logs {name} --status | grep "Job 1: SUCCEEDED"', # Equivalent.
# Test launch output again on existing cluster
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --cloud {generic_cloud}{smoke_tests_utils.LOW_RESOURCE_ARG} tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 2 --status',
f'sky logs {name} --status | grep "Job 2: SUCCEEDED"', # Equivalent.
# Check the logs downloading
f'log_path=$(sky logs {name} 1 --sync-down | grep "Job 1 logs:" | sed -E "s/^.*Job 1 logs: (.*)\\x1b\\[0m/\\1/g") && echo "$log_path" '
# We need to explicitly expand the log path as it starts with ~, and linux does not automatically
# expand it when having it in a variable.
' && expanded_log_path=$(eval echo "$log_path") && echo "$expanded_log_path" '
' && test -f $expanded_log_path/run.log',
# Ensure the raylet process has the correct file descriptor limit.
f'sky exec {name} "prlimit -n --pid=\$(pgrep -f \'raylet/raylet --raylet_socket_name\') | grep \'"\'1048576 1048576\'"\'"',
f'sky logs {name} 3 --status', # Ensure the job succeeded.
# Install jq for the next test.
f'sky exec {name}\'sudo apt-get update && sudo apt-get install -y jq\'',
# Check the cluster info
f'sky exec {name}\'echo "$SKYPILOT_CLUSTER_INFO" | jq .cluster_name | grep {name}\'',
f'sky logs {name} 5 --status', # Ensure the job succeeded.
f'sky exec {name}\'echo "$SKYPILOT_CLUSTER_INFO" | jq .cloud | grep -i {generic_cloud}\'',
f'sky logs {name} 6 --status', # Ensure the job succeeded.
# Test '-c' for exec
f'sky exec -c {name} echo',
f'sky logs {name} 7 --status',
f'sky exec echo -c {name}',
f'sky logs {name} 8 --status',
f'sky exec -c {name} echo hi test',
f'sky logs {name} 9 | grep "hi test"',
f'sky exec {name} && exit 1 || true',
f'sky exec -c {name} && exit 1 || true',
 ],
f'sky down -y {name}',
smoke_tests_utils.get_timeout(generic_cloud),
 )
smoke_tests_utils.run_one_test(test)


# ---------- Test fast launch ----------
deftest_launch_fast(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()

test=smoke_tests_utils.Test(
'test_launch_fast',
 [
# First launch to create the cluster
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --cloud {generic_cloud} --fast {smoke_tests_utils.LOW_RESOURCE_ARG} tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 1 --status',

# Second launch to test fast launch - should not reprovision
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --fast tests/test_yamls/minimal.yaml) && '
' echo "$s" && '
# Validate that cluster was not re-launched.
'! echo "$s" | grep -A 1 "Launching on" | grep "is up." && '
# Validate that setup was not re-run.
'! echo "$s" | grep -A 1 "Running setup on" | grep "running setup" && '
# Validate that the task ran and finished.
'echo "$s" | grep -A 1 "task run finish" | grep "Job finished (status: SUCCEEDED)"',
f'sky logs {name} 2 --status',
f'sky status -r {name} | grep UP',
 ],
f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(generic_cloud),
 )
smoke_tests_utils.run_one_test(test)


# See cloud exclusion explanations in test_autostop
@pytest.mark.no_fluidstack
@pytest.mark.no_lambda_cloud
@pytest.mark.no_ibm
@pytest.mark.no_kubernetes
deftest_launch_fast_with_autostop(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
# Azure takes ~ 7m15s (435s) to autostop a VM, so here we use 600 to ensure
# the VM is stopped.
autostop_timeout=600ifgeneric_cloud=='azure'else250
test=smoke_tests_utils.Test(
'test_launch_fast_with_autostop',
 [
# First launch to create the cluster with a short autostop
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --cloud {generic_cloud} --fast -i 1 {smoke_tests_utils.LOW_RESOURCE_ARG} tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 1 --status',
f'sky status -r {name} | grep UP',

# Ensure cluster is stopped
smoke_tests_utils.get_cmd_wait_until_cluster_status_contains(
cluster_name=name,
cluster_status=[sky.ClusterStatus.STOPPED],
timeout=autostop_timeout),
# Even the cluster is stopped, cloud platform may take a while to
# delete the VM.
# FIXME(aylei): this can be flaky, sleep longer for now.
f'sleep 60',
# Launch again. Do full output validation - we expect the cluster to re-launch
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --fast -i 1 tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 2 --status',
f'sky status -r {name} | grep UP',
 ],
f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(generic_cloud) +autostop_timeout,
 )
smoke_tests_utils.run_one_test(test)


# We override the AWS config to force the cluster to relaunch, so only run the
# test on AWS.
@pytest.mark.aws
deftest_launch_fast_with_cluster_changes(generic_cloud: str, tmp_path):
name=smoke_tests_utils.get_cluster_name()
tmp_config_path=tmp_path/'sky.yaml'
test=smoke_tests_utils.Test(
'test_launch_fast_with_cluster_changes',
 [
# Initial launch
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --cloud {generic_cloud} --fast {smoke_tests_utils.LOW_RESOURCE_ARG} tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 1 --status',

# Launch again - setup and provisioning should be skipped
f's=$(SKYPILOT_DEBUG=0 sky launch -y -c {name} --fast tests/test_yamls/minimal.yaml) && '
' echo "$s" && '
# Validate that cluster was not re-launched.
'! echo "$s" | grep -A 1 "Launching on" | grep "is up." && '
# Validate that setup was not re-run.
'! echo "$s" | grep -A 1 "Running setup on" | grep "running setup" && '
f'sky logs {name} 2 --status',

# Copy current config as a base.
f'cp ${{{skypilot_config.ENV_VAR_SKYPILOT_CONFIG}:-~/.sky/config.yaml}} {tmp_config_path} && '
# Set config override. This should change the cluster yaml, forcing reprovision/setup
f'echo "aws: {{ remote_identity: test }}" >> {tmp_config_path}',
# Launch and do full output validation. Setup/provisioning should be run.
f's=$(SKYPILOT_DEBUG=0 {skypilot_config.ENV_VAR_SKYPILOT_CONFIG}={tmp_config_path} sky launch -y -c {name} --fast tests/test_yamls/minimal.yaml) && {smoke_tests_utils.VALIDATE_LAUNCH_OUTPUT}',
f'sky logs {name} 3 --status',
f'sky status -r {name} | grep UP',
 ],
f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(generic_cloud),
 )
smoke_tests_utils.run_one_test(test)


# ------------ Test stale job ------------
@pytest.mark.no_fluidstack# FluidStack does not support stopping instances in SkyPilot implementation
@pytest.mark.no_lambda_cloud# Lambda Cloud does not support stopping instances
@pytest.mark.no_kubernetes# Kubernetes does not support stopping instances
@pytest.mark.no_vast# This requires port opening
deftest_stale_job(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'stale_job',
 [
f'sky launch -y -c {name} --cloud {generic_cloud}{smoke_tests_utils.LOW_RESOURCE_ARG} "echo hi"',
f'sky exec {name} -d "echo start; sleep 10000"',
f'sky stop {name} -y',
smoke_tests_utils.get_cmd_wait_until_cluster_status_contains(
cluster_name=name,
cluster_status=[sky.ClusterStatus.STOPPED],
timeout=100),
f'sky start {name} -y',
f'sky logs {name} 1 --status',
f's=$(sky queue {name}); echo "$s"; echo; echo; echo "$s" | grep FAILED_DRIVER',
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast
@pytest.mark.aws
deftest_aws_stale_job_manual_restart():
name=smoke_tests_utils.get_cluster_name()
name_on_cloud=common_utils.make_cluster_name_on_cloud(
name, sky.AWS.max_cluster_name_length())
region='us-east-2'
test=smoke_tests_utils.Test(
'aws_stale_job_manual_restart',
 [
smoke_tests_utils.launch_cluster_for_cloud_cmd('aws', name),
f'sky launch -y -c {name} --cloud aws --region {region}{smoke_tests_utils.LOW_RESOURCE_ARG} "echo hi"',
f'sky exec {name} -d "echo start; sleep 10000"',
# Stop the cluster manually.
smoke_tests_utils.run_cloud_cmd_on_cluster(
name,
cmd=
 (f'id=`aws ec2 describe-instances --region {region} --filters '
f'Name=tag:ray-cluster-name,Values={name_on_cloud} '
f'--query Reservations[].Instances[].InstanceId '
f'--output text` && '
f'aws ec2 stop-instances --region {region} '
f'--instance-ids $id')),
smoke_tests_utils.get_cmd_wait_until_cluster_status_contains(
cluster_name=name,
cluster_status=[sky.ClusterStatus.STOPPED],
timeout=40),
f'sky launch -c {name} -y "echo hi"',
f'sky logs {name} 1 --status',
f'sky logs {name} 3 --status',
# Ensure the skylet updated the stale job status.
smoke_tests_utils.
get_cmd_wait_until_job_status_contains_without_matching_job(
cluster_name=name,
job_status=[sky.JobStatus.FAILED_DRIVER],
timeout=events.JobSchedulerEvent.EVENT_INTERVAL_SECONDS),
 ],
f'sky down -y {name} && {smoke_tests_utils.down_cluster_for_cloud_cmd(name)}',
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast
@pytest.mark.gcp
deftest_gcp_stale_job_manual_restart():
name=smoke_tests_utils.get_cluster_name()
name_on_cloud=common_utils.make_cluster_name_on_cloud(
name, sky.GCP.max_cluster_name_length())
zone='us-central1-a'
query_cmd= (f'gcloud compute instances list --filter='
f'"(labels.ray-cluster-name={name_on_cloud})" '
f'--zones={zone} --format="value(name)"')
stop_cmd= (f'gcloud compute instances stop --zone={zone}'
f' --quiet $({query_cmd})')
test=smoke_tests_utils.Test(
'gcp_stale_job_manual_restart',
 [
smoke_tests_utils.launch_cluster_for_cloud_cmd('gcp', name),
f'sky launch -y -c {name} --cloud gcp --zone {zone}{smoke_tests_utils.LOW_RESOURCE_ARG} "echo hi"',
f'sky exec {name} -d "echo start; sleep 10000"',
# Stop the cluster manually.
smoke_tests_utils.run_cloud_cmd_on_cluster(name, cmd=stop_cmd),
'sleep 40',
f'sky launch -c {name} -y "echo hi"',
f'sky logs {name} 1 --status',
f'sky logs {name} 3 --status',
# Ensure the skylet updated the stale job status.
smoke_tests_utils.
get_cmd_wait_until_job_status_contains_without_matching_job(
cluster_name=name,
job_status=[sky.JobStatus.FAILED_DRIVER],
timeout=events.JobSchedulerEvent.EVENT_INTERVAL_SECONDS)
 ],
f'sky down -y {name} && {smoke_tests_utils.down_cluster_for_cloud_cmd(name)}',
 )
smoke_tests_utils.run_one_test(test)


# ---------- Check Sky's environment variables; workdir. ----------
@pytest.mark.no_fluidstack# Requires amazon S3
@pytest.mark.no_scp# SCP does not support num_nodes > 1 yet
@pytest.mark.no_vast# Vast does not support num_nodes > 1 yet
deftest_env_check(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
total_timeout_minutes=25ifgeneric_cloud=='azure'else15
test=smoke_tests_utils.Test(
'env_check',
 [
f'sky launch -y -c {name} --cloud {generic_cloud}{smoke_tests_utils.LOW_RESOURCE_ARG} examples/env_check.yaml',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
# Test with only setup.
f'sky launch -y -c {name} tests/test_yamls/test_only_setup.yaml',
f'sky logs {name} 2 --status',
f'sky logs {name} 2 | grep "hello world"',
 ],
f'sky down -y {name}',
timeout=total_timeout_minutes*60,
 )
smoke_tests_utils.run_one_test(test)


# ---------- CLI logs ----------
@pytest.mark.no_scp# SCP does not support num_nodes > 1 yet. Run test_scp_logs instead.
@pytest.mark.no_vast# Vast does not support num_nodes > 1 yet.
deftest_cli_logs(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
num_nodes=2
ifgeneric_cloud=='kubernetes':
# Kubernetes does not support multi-node
num_nodes=1
timestamp=time.time()
test=smoke_tests_utils.Test('cli_logs', [
f'sky launch -y -c {name} --cloud {generic_cloud} --num-nodes {num_nodes}{smoke_tests_utils.LOW_RESOURCE_ARG} "echo {timestamp} 1"',
f'sky exec {name} "echo {timestamp} 2"',
f'sky exec {name} "echo {timestamp} 3"',
f'sky exec {name} "echo {timestamp} 4"',
f'sky logs {name} 2 --status',
f'sky logs {name} 3 4 --sync-down',
f'sky logs {name} * --sync-down',
f'sky logs {name} 1 | grep "{timestamp} 1"',
f'sky logs {name} | grep "{timestamp} 4"',
 ], f'sky down -y {name}')
smoke_tests_utils.run_one_test(test)


@pytest.mark.scp
deftest_scp_logs():
name=smoke_tests_utils.get_cluster_name()
timestamp=time.time()
test=smoke_tests_utils.Test(
'SCP_cli_logs',
 [
f'sky launch -y -c {name}{smoke_tests_utils.SCP_TYPE} "echo {timestamp} 1"',
f'sky exec {name} "echo {timestamp} 2"',
f'sky exec {name} "echo {timestamp} 3"',
f'sky exec {name} "echo {timestamp} 4"',
f'sky logs {name} 2 --status',
f'sky logs {name} 3 4 --sync-down',
f'sky logs {name} * --sync-down',
f'sky logs {name} 1 | grep "{timestamp} 1"',
f'sky logs {name} | grep "{timestamp} 4"',
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


# ------- Testing the core API --------
# Most of the core APIs have been tested in the CLI tests.
# These tests are for testing the return value of the APIs not fully used in CLI.
deftest_core_api_sky_launch_exec(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
cloud=sky.CLOUD_REGISTRY.from_str(generic_cloud)
task=sky.Task(run="whoami")
task.set_resources(
sky.Resources(cloud=cloud, **smoke_tests_utils.LOW_RESOURCE_PARAM))
try:
job_id, handle=sky.get(sky.launch(task, cluster_name=name))
assertjob_id==1
asserthandleisnotNone
asserthandle.cluster_name==name
asserthandle.launched_resources.cloud.is_same_cloud(cloud)
job_id_exec, handle_exec=sky.get(sky.exec(task, cluster_name=name))
assertjob_id_exec==2
asserthandle_execisnotNone
asserthandle_exec.cluster_name==name
asserthandle_exec.launched_resources.cloud.is_same_cloud(cloud)
# For dummy task (i.e. task.run is None), the job won't be submitted.
dummy_task=sky.Task()
job_id_dummy, _=sky.get(sky.exec(dummy_task, cluster_name=name))
assertjob_id_dummyisNone
# Check the cluster status from the dashboard
cluster_exist=False
status_request_id= (
smoke_tests_utils.get_dashboard_cluster_status_request_id())
status_response= (
smoke_tests_utils.get_response_from_request_id(status_request_id))
forclusterinstatus_response:
ifcluster['name'] ==name:
cluster_exist=True
break
assertcluster_exist
finally:
sky.get(sky.down(name))


# The sky launch CLI has some additional checks to make sure the cluster is up/
# restarted. However, the core API doesn't have these; make sure it still works
@pytest.mark.no_kubernetes
deftest_core_api_sky_launch_fast(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
cloud=sky.CLOUD_REGISTRY.from_str(generic_cloud)
try:
task=sky.Task(run="whoami").set_resources(
sky.Resources(cloud=cloud, **smoke_tests_utils.LOW_RESOURCE_PARAM))
sky.launch(task,
cluster_name=name,
idle_minutes_to_autostop=1,
fast=True)
# Sleep to let the cluster autostop
smoke_tests_utils.get_cmd_wait_until_cluster_status_contains(
cluster_name=name,
cluster_status=[sky.ClusterStatus.STOPPED],
timeout=120)
# Run it again - should work with fast=True
sky.launch(task,
cluster_name=name,
idle_minutes_to_autostop=1,
fast=True)
finally:
sky.down(name)


deftest_jobs_launch_and_logs(generic_cloud: str):
# Use the context manager
withskypilot_config.override_skypilot_config(
smoke_tests_utils.LOW_CONTROLLER_RESOURCE_OVERRIDE_CONFIG):
name=smoke_tests_utils.get_cluster_name()
task=sky.Task(run="echo start job; sleep 30; echo end job")
cloud=sky.CLOUD_REGISTRY.from_str(generic_cloud)
task.set_resources(
sky.Resources(cloud=cloud, **smoke_tests_utils.LOW_RESOURCE_PARAM))
job_id, handle=sky.stream_and_get(sky.jobs.launch(task, name=name))
asserthandleisnotNone
# Check the job status from the dashboard
queue_request_id= (
smoke_tests_utils.get_dashboard_jobs_queue_request_id())
queue_response= (
smoke_tests_utils.get_response_from_request_id(queue_request_id))
job_exist=False
forjobinqueue_response:
ifjob['job_id'] ==job_id:
job_exist=True
break
assertjob_exist
try:
withtempfile.TemporaryFile(mode='w+', encoding='utf-8') asf:
sky.jobs.tail_logs(job_id=job_id, output_stream=f)
f.seek(0)
content=f.read()
assertcontent.count('start job') ==1
assertcontent.count('end job') ==1
finally:
sky.jobs.cancel(job_ids=[job_id])


# ---------- Testing YAML Specs ----------
# Our sky storage requires credentials to check the bucket existance when
# loading a task from the yaml file, so we cannot make it a unit test.
classTestYamlSpecs:
# TODO(zhwu): Add test for `to_yaml_config` for the Storage object.
# We should not use `examples/storage_demo.yaml` here, since it requires
# users to ensure bucket names to not exist and/or be unique.
_TEST_YAML_PATHS= [
'examples/minimal.yaml', 'examples/managed_job.yaml',
'examples/using_file_mounts.yaml', 'examples/resnet_app.yaml',
'examples/multi_hostname.yaml'
 ]

def_is_dict_subset(self, d1, d2):
"""Check if d1 is the subset of d2."""
fork, vind1.items():
ifknotind2:
ifisinstance(v, list) orisinstance(v, dict):
assertlen(v) ==0, (k, v)
else:
assertFalse, (k, v)
elifisinstance(v, dict):
assertisinstance(d2[k], dict), (k, v, d2)
self._is_dict_subset(v, d2[k])
elifisinstance(v, str):
ifk=='accelerators':
resources=sky.Resources()
resources._set_accelerators(v, None)
assertresources.accelerators==d2[k], (k, v, d2)
else:
assertv.lower() ==d2[k].lower(), (k, v, d2[k])
else:
assertv==d2[k], (k, v, d2[k])

def_check_equivalent(self, yaml_path):
"""Check if the yaml is equivalent after load and dump again."""
origin_task_config=common_utils.read_yaml(yaml_path)

task=sky.Task.from_yaml(yaml_path)
new_task_config=task.to_yaml_config()
# d1 <= d2
print(origin_task_config, new_task_config)
self._is_dict_subset(origin_task_config, new_task_config)

deftest_load_dump_yaml_config_equivalent(self):
"""Test if the yaml config is equivalent after load and dump again."""
pathlib.Path('~/datasets').expanduser().mkdir(exist_ok=True)
pathlib.Path('~/tmpfile').expanduser().touch()
pathlib.Path('~/.ssh').expanduser().mkdir(exist_ok=True)
pathlib.Path('~/.ssh/id_rsa.pub').expanduser().touch()
pathlib.Path('~/tmp-workdir').expanduser().mkdir(exist_ok=True)
pathlib.Path('~/Downloads/tpu').expanduser().mkdir(parents=True,
exist_ok=True)
foryaml_pathinself._TEST_YAML_PATHS:
self._check_equivalent(yaml_path)


# ---------- Testing Multiple Accelerators ----------
@pytest.mark.no_vast# Vast has low availability for K80 GPUs
@pytest.mark.no_fluidstack# Fluidstack does not support K80 gpus for now
@pytest.mark.no_paperspace# Paperspace does not support K80 gpus
@pytest.mark.no_nebius# Nebius does not support K80s
deftest_multiple_accelerators_ordered():
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'multiple-accelerators-ordered',
 [
f'sky launch -y -c {name} tests/test_yamls/test_multiple_accelerators_ordered.yaml | grep "Using user-specified accelerators list"',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
 ],
f'sky down -y {name}',
timeout=20*60,
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast# Vast has low availability for T4 GPUs
@pytest.mark.no_fluidstack# Fluidstack has low availability for T4 GPUs
@pytest.mark.no_paperspace# Paperspace does not support T4 GPUs
@pytest.mark.no_nebius# Nebius does not support T4 GPUs
deftest_multiple_accelerators_ordered_with_default():
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'multiple-accelerators-ordered',
 [
f'sky launch -y -c {name} tests/test_yamls/test_multiple_accelerators_ordered_with_default.yaml | grep "Using user-specified accelerators list"',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
f'sky status {name} | grep Spot',
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast# Vast has low availability for T4 GPUs
@pytest.mark.no_fluidstack# Fluidstack has low availability for T4 GPUs
@pytest.mark.no_paperspace# Paperspace does not support T4 GPUs
@pytest.mark.no_nebius# Nebius does not support T4 GPUs
deftest_multiple_accelerators_unordered():
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'multiple-accelerators-unordered',
 [
f'sky launch -y -c {name} tests/test_yamls/test_multiple_accelerators_unordered.yaml',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast# Vast has low availability for T4 GPUs
@pytest.mark.no_fluidstack# Fluidstack has low availability for T4 GPUs
@pytest.mark.no_paperspace# Paperspace does not support T4 GPUs
@pytest.mark.no_nebius# Nebius does not support T4 GPUs
deftest_multiple_accelerators_unordered_with_default():
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'multiple-accelerators-unordered-with-default',
 [
f'sky launch -y -c {name} tests/test_yamls/test_multiple_accelerators_unordered_with_default.yaml',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
f'sky status {name} | grep Spot',
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.no_vast# Requires other clouds to be enabled
@pytest.mark.no_fluidstack# Requires other clouds to be enabled
deftest_multiple_resources():
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'multiple-resources',
 [
f'sky launch -y -c {name} tests/test_yamls/test_multiple_resources.yaml',
f'sky logs {name} 1 --status', # Ensure the job succeeded.
 ],
f'sky down -y {name}',
 )
smoke_tests_utils.run_one_test(test)


# ---------- Sky Benchmark ----------
@pytest.mark.skip(reason='SkyBench is not supported in API server')
@pytest.mark.no_fluidstack# Requires other clouds to be enabled
@pytest.mark.no_vast# Requires other clouds to be enabled
@pytest.mark.no_paperspace# Requires other clouds to be enabled
@pytest.mark.no_kubernetes
@pytest.mark.aws# SkyBenchmark requires S3 access
deftest_sky_bench(generic_cloud: str):
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'sky-bench',
 [
f'sky bench launch -y -b {name} --cloud {generic_cloud} -i0 tests/test_yamls/minimal.yaml',
'sleep 120',
f'sky bench show {name} | grep sky-bench-{name} | grep FINISHED',
 ],
f'sky bench down {name} -y; sky bench delete {name} -y',
 )
smoke_tests_utils.run_one_test(test)


@pytest.fixture(scope='session')
defunreachable_context():
"""Setup the kubernetes context for the test.

 This fixture will copy the kubeconfig file and inject an unreachable context
 to it. So this must be session scoped that the kubeconfig is modified before
 the local API server starts.
 """
# Get kubeconfig path from environment variable or use default
kubeconfig_path=os.environ.get('KUBECONFIG',
os.path.expanduser('~/.kube/config'))
ifnotos.path.exists(kubeconfig_path):
return
importshutil

# Create a temp kubeconfig
temp_kubeconfig=tempfile.NamedTemporaryFile(delete=False, suffix='.yaml')
shutil.copy(kubeconfig_path, temp_kubeconfig.name)
original_kubeconfig=os.environ.get('KUBECONFIG')
os.environ['KUBECONFIG'] =temp_kubeconfig.name

free_port=common_utils.find_free_port(30000)
unreachable_name='_unreachable_context_'
subprocess.run(
'kubectl config set-cluster unreachable-cluster '
f'--server=https://127.0.0.1:{free_port} && '
'kubectl config set-credentials unreachable-user '
'--token="aQo=" && '
'kubectl config set-context '+unreachable_name+' '
'--cluster=unreachable-cluster --user=unreachable-user && '
# Restart the API server to pick up kubeconfig change
# TODO(aylei): There is a implicit API server restart before starting
# smoke tests in CI pipeline. We should move that to fixture to make
# the test coherent.
'sky api stop || true && sky api start',
shell=True,
check=True)

yieldunreachable_name

# Clean up
iforiginal_kubeconfig:
os.environ['KUBECONFIG'] =original_kubeconfig
else:
os.environ.pop('KUBECONFIG', None)
os.unlink(temp_kubeconfig.name)


@pytest.mark.kubernetes
deftest_kubernetes_context_failover(unreachable_context):
"""Test if the kubernetes context failover works.

 This test requires two kubernetes clusters:
 - kind-skypilot: the local cluster with mock labels for 8 H100 GPUs.
 - another accessible cluster: with enough CPUs
 To start the first cluster, run:
 sky local up
 # Add mock label for accelerator
 kubectl label node --overwrite skypilot-control-plane skypilot.co/accelerator=h100 --context kind-skypilot
 # Patch accelerator capacity
 kubectl patch node skypilot-control-plane --subresource=status -p '{"status": {"capacity": {"nvidia.com/gpu": "8"}}}' --context kind-skypilot
 # Add a new namespace to test the handling of namespaces
 kubectl create namespace test-namespace --context kind-skypilot
 # Set the namespace to test-namespace
 kubectl config set-context kind-skypilot --namespace=test-namespace --context kind-skypilot
 """
# Get context that is not kind-skypilot
contexts=subprocess.check_output('kubectl config get-contexts -o name',
shell=True).decode('utf-8').split('\n')
assertunreachable_contextincontexts, (
'unreachable_context should be initialized in the fixture')
context= [
contextforcontextincontexts
if (context!='kind-skypilot'andcontext!=unreachable_context)
 ][0]
# Test unreachable context and non-existing context do not break failover
config=textwrap.dedent(f"""\
 kubernetes:
 allowed_contexts:
 - {context}
 - {unreachable_context}
 - _nonexist_
 - kind-skypilot
 """)
withtempfile.NamedTemporaryFile(delete=True) asf:
f.write(config.encode('utf-8'))
f.flush()
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'kubernetes-context-failover',
 [
# Check if kind-skypilot is provisioned with H100 annotations already
'NODE_INFO=$(kubectl get nodes -o yaml --context kind-skypilot) && '
'echo "$NODE_INFO" | grep nvidia.com/gpu | grep 8 && '
'echo "$NODE_INFO" | grep skypilot.co/accelerator | grep h100 || '
'{ echo "kind-skypilot does not exist '
'or does not have mock labels for GPUs. Check the instructions in '
'tests/test_smoke.py::test_kubernetes_context_failover." && exit 1; }',
# Check namespace for kind-skypilot is test-namespace
'kubectl get namespaces --context kind-skypilot | grep test-namespace || '
'{ echo "Should set the namespace to test-namespace for kind-skypilot. Check the instructions in '
'tests/test_smoke.py::test_kubernetes_context_failover." && exit 1; }',
'sky show-gpus --cloud kubernetes --region kind-skypilot | grep H100 | grep "1, 2, 4, 8"',
# Get contexts and set current context to the other cluster that is not kind-skypilot
f'kubectl config use-context {context}',
# H100 should not in the current context
'! sky show-gpus --cloud kubernetes | grep H100',
f'sky launch -y -c {name}-1 --cpus 1 echo hi',
f'sky logs {name}-1 --status',
# It should be launched not on kind-skypilot
f'sky status -v {name}-1 | grep "{context}"',
# Test failure for launching H100 on other cluster
f'sky launch -y -c {name}-2 --gpus H100 --cpus 1 --cloud kubernetes --region {context} echo hi && exit 1 || true',
# Test failover
f'sky launch -y -c {name}-3 --gpus H100 --cpus 1 --cloud kubernetes echo hi',
f'sky logs {name}-3 --status',
# Test pods
f'kubectl get pods --context kind-skypilot | grep "{name}-3"',
# It should be launched on kind-skypilot
f'sky status -v {name}-3 | grep "kind-skypilot"',
# Should be 7 free GPUs
f'sky show-gpus --cloud kubernetes --region kind-skypilot | grep H100 | grep " 7"',
# Remove the line with "kind-skypilot"
f'sed -i "/kind-skypilot/d" {f.name}',
# Should still be able to exec and launch on existing cluster
f'sky exec {name}-3 "echo hi"',
f'sky logs {name}-3 --status',
f'sky status -r {name}-3 | grep UP',
f'sky launch -c {name}-3 --gpus h100 echo hi',
f'sky logs {name}-3 --status',
f'sky status -r {name}-3 | grep UP',
# Test failure for launching on unreachable context
f'kubectl config use-context {unreachable_context}',
f'sky launch -y -c {name}-4 --gpus H100 --cpus 1 --cloud kubernetes --region {unreachable_context} echo hi && exit 1 || true',
# Test failover from unreachable context
f'sky launch -y -c {name}-5 --cpus 1 echo hi',
 ],
f'sky down -y {name}-1 {name}-3 {name}-5',
env={
skypilot_config.ENV_VAR_SKYPILOT_CONFIG: f.name,
constants.SKY_API_SERVER_URL_ENV_VAR:
sky.server.common.get_server_url()
 },
 )
smoke_tests_utils.run_one_test(test)


deftest_launch_and_exec_async(generic_cloud: str):
"""Test if the launch and exec commands work correctly with --async."""
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'launch_and_exec_async',
 [
f'sky launch -c {name} -y --async',
# Async exec.
f'sky exec {name} echo --async',
# Async exec and cancel immediately.
 (f's=$(sky exec {name} echo --async) && '
'echo "$s" && '
'cancel_cmd=$(echo "$s" | grep "To cancel the request" | '
'sed -E "s/.*run: (sky api cancel .*).*/\\1/") && '
'echo "Extracted cancel command: $cancel_cmd" && '
'$cancel_cmd'),
# Sync exec must succeed after command end.
 (
f's=$(sky exec {name} echo) && echo "$s" && '
'echo "===check exec output===" && '
'job_id=$(echo "$s" | grep "Job submitted, ID:" | '
'sed -E "s/.*Job submitted, ID: ([0-9]+).*/\\1/") && '
f'sky logs {name} $job_id --status | grep "SUCCEEDED" && '
# If job_id is 1, async_job_id will be 2, and vice versa.
'async_job_id=$((3-job_id)) && '
f'echo "===check async job===" && echo "Job ID: $async_job_id" && '
# Wait async job to succeed.
f'{smoke_tests_utils.get_cmd_wait_until_job_status_succeeded(name, "$async_job_id")}'
 ),
# Cluster must be UP since the sync exec has been completed.
f'sky status {name} | grep "UP"',
# The cancelled job should not be scheduled, the job ID 3 is just
# not exist.
f'! sky logs {name} 3 --status | grep "SUCCEEDED"',
 ],
teardown=f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(generic_cloud))
smoke_tests_utils.run_one_test(test)


deftest_cancel_launch_and_exec_async(generic_cloud: str):
"""Test if async launch and exec commands work correctly when cluster is shutdown"""
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test('cancel_launch_and_exec_async', [
f'sky launch -c {name} -y --cloud {generic_cloud} --async',
 (f's=$(sky exec {name} echo --async) && '
'echo "$s" && '
'logs_cmd=$(echo "$s" | grep "Check logs with" | '
'sed -E "s/.*with: (sky api logs .*).*/\\1/") && '
'echo "Extracted logs command: $logs_cmd" && '
f'{smoke_tests_utils.get_cmd_wait_until_cluster_status_contains(name, [sky.ClusterStatus.INIT, sky.ClusterStatus.UP], 30)} && '
f'sky down -y {name} && '
'log_output=$(eval $logs_cmd || true) && '
'echo "===logs===" && echo "$log_output" && '
'echo "$log_output" | grep "cancelled"'),
 ],
teardown=f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(
generic_cloud))
smoke_tests_utils.run_one_test(test)


# ---------- Testing Exit Codes for CLI commands ----------
deftest_cli_exit_codes(generic_cloud: str):
"""Test that CLI commands properly return exit codes based on job success/failure."""
name=smoke_tests_utils.get_cluster_name()
test=smoke_tests_utils.Test(
'cli_exit_codes',
 [
# Test successful job exit code (0)
f'sky launch -y -c {name} --cloud {generic_cloud} "echo success" && echo "Exit code: $?"',
f'sky logs {name} 1 --status | grep SUCCEEDED',

# Test that sky logs with successful job returns 0
f'sky logs {name} 1 && echo "Exit code: $?"',

# Test failed job exit code (100)
f'sky exec {name} "exit 1" || echo "Command failed with code: $?" | grep "Command failed with code: 100"',
f'sky logs {name} 2 --status | grep FAILED',
f'sky logs {name} 2 || echo "Job logs exit code: $?" | grep "Job logs exit code: 100"',
 ],
f'sky down -y {name}',
timeout=smoke_tests_utils.get_timeout(generic_cloud),
 )
smoke_tests_utils.run_one_test(test)


@pytest.mark.lambda_cloud
deftest_lambda_cloud_open_ports():
"""Test Lambda Cloud open ports functionality.

 It tests the functionality by opening both a single port and a port range,
 verifying that both types of rules are created successfully. It also tests
 that consecutive individual ports are properly merged into a single range rule.
 """
# Test ports and port ranges
single_port='12345'
single_port_int=int(single_port)
port_range='5000-5010'
port_range_start=5000
port_range_end=5010

# Consecutive ports that should be merged
consecutive_ports= ['6000', '6001', '6002']
consecutive_start=6000
consecutive_end=6002

# Store initial rules to avoid modifying rules that existed before the test
initial_rules= []
lambda_client=None

fromsky.provision.lambda_cloudimportinstance
fromsky.provision.lambda_cloudimportlambda_utils

try:
# Initialize Lambda Cloud client
lambda_client=lambda_utils.LambdaCloudClient()

# Check if our test method exists - if not, test will be skipped
ifnothasattr(lambda_client, 'list_firewall_rules') ornothasattr(
lambda_client, 'create_firewall_rule'):
pytest.skip(
'LambdaCloudClient doesn\'t have required firewall rule methods'
 )

# Skip test for us-south-1 region where firewall rules are not supported
ifany('us-south-1'instr(rule)
forruleinlambda_client.list_catalog().values()):
# Check if our current region is us-south-1
instances=lambda_client.list_instances()
forinstininstances:
ifinst.get('region', {}).get('name') =='us-south-1':
pytest.skip(
'Firewall rules not supported in us-south-1 region')

# Get initial rules for debugging and tracking purposes
initial_rules=lambda_client.list_firewall_rules()
print(f'Initial firewall rules count: {len(initial_rules)}')

# Print example rule structure for debugging
ifinitial_rules:
print(f'Example rule structure: {initial_rules[0]}')

# 1. Test opening a single port
print(f'Opening single port {single_port}')
instance.open_ports('smoke-test-cluster', [single_port])
print(f'Successfully called open_ports for single port {single_port}')

# 2. Test opening a port range
print(f'Opening port range {port_range}')
instance.open_ports('smoke-test-cluster', [port_range])
print(f'Successfully called open_ports for port range {port_range}')

# 3. Test opening consecutive ports to verify merging
print(f'Opening consecutive ports {", ".join(consecutive_ports)}')
instance.open_ports('smoke-test-cluster', consecutive_ports)
print('Successfully called open_ports for consecutive ports '
f'{", ".join(consecutive_ports)}')

# Verify rules were created by getting current rules
current_rules=lambda_client.list_firewall_rules()
print(f'Rules after adding our test ports: {len(current_rules)}')

# Basic verification that rules were added
# (should have at least as many rules as before)
assertlen(current_rules) >=len(
initial_rules), 'No new rules were added'

# 4. Verify consecutive ports were merged into a range
merged_range_found=False
forruleincurrent_rules:
if (rule.get('protocol') =='tcp'andrule.get('port_range') and
len(rule.get('port_range')) ==2and
rule.get('port_range')[0] ==consecutive_startand
rule.get('port_range')[1] ==consecutive_end):

# Check that it's our auto-generated rule
description=rule.get('description', '')
if'SkyPilot auto-generated'indescription:
merged_range_found=True
print('Found merged port range rule: TCP '
f'{consecutive_start}-{consecutive_end}')
break

# Make sure port merging worked - now with a hard assertion
assertmerged_range_found, (
f'Ports {consecutive_ports} were not merged into a single range '
f'rule {consecutive_start}-{consecutive_end}. '
'Port rule merging is not working as expected.')

exceptExceptionase:
importtraceback
print(f'Error in test: {e}')
print(traceback.format_exc())
pytest.fail(f'Error testing Lambda Cloud open_ports: {str(e)}')

finally:
# Clean up the test ports we created, being careful to preserve
# pre-existing rules
iflambda_clientisNone:
print('Lambda client not initialized, skipping cleanup')
elifnotinitial_rules:
print('No initial rules were recorded, skipping cleanup for safety')
else:
try:
# We need to clean up manually since instance.cleanup_ports
# intentionally skips cleanup for Lambda Cloud (as firewall
# rules are global to the account)

# Get all current rules
current_rules=lambda_client.list_firewall_rules()

# Create a set of "fingerprints" for initial rules for faster
# comparison.