import sys import warnings import pytest from pydantic import ValidationError import ray from ray import cloudpickle, serve from ray._common.utils import import_attr from ray.serve._private.config import ( DeploymentConfig, ReplicaConfig, _proto_to_dict, ) from ray.serve._private.constants import ( DEFAULT_AUTOSCALING_POLICY_NAME, DEFAULT_GRPC_PORT, DEFAULT_ROLLING_UPDATE_PERCENTAGE, RAY_SERVE_ROUTER_RETRY_BACKOFF_MULTIPLIER, RAY_SERVE_ROUTER_RETRY_INITIAL_BACKOFF_S, RAY_SERVE_ROUTER_RETRY_MAX_BACKOFF_S, ) from ray.serve._private.request_router import PowerOfTwoChoicesRequestRouter from ray.serve._private.utils import DEFAULT from ray.serve.autoscaling_policy import default_autoscaling_policy from ray.serve.config import ( AutoscalingConfig, ControllerOptions, DeploymentActorConfig, GangPlacementStrategy, GangRuntimeFailurePolicy, GangSchedulingConfig, HTTPOptions, ProxyLocation, RequestRouterConfig, gRPCOptions, ) from ray.serve.generated.serve_pb2 import ( AutoscalingConfig as AutoscalingConfigProto, DeploymentConfig as DeploymentConfigProto, DeploymentLanguage, ) from ray.serve.generated.serve_pb2_grpc import add_UserDefinedServiceServicer_to_server from ray.serve.schema import ( DeploymentSchema, HTTPOptionsSchema, ServeApplicationSchema, ServeDeploySchema, ) fake_policy_return_value = 123 def fake_policy(): return fake_policy_return_value class FakeRequestRouter: ... @ray.remote class _TestDummyActor: """Used for deployment_actors import path test.""" pass @ray.remote class _TestRayActor: """Used for deployment_actors proto roundtrip test (needs __ray_actor_class__).""" def ping(self): """Dummy method to verify class is deserialized correctly.""" return "pong" def test_autoscaling_config_validation(): # Check validation over publicly exposed options with pytest.raises(ValidationError): # min_replicas must be nonnegative AutoscalingConfig(min_replicas=-1) with pytest.raises(ValidationError): # max_replicas must be positive AutoscalingConfig(max_replicas=0) # target_ongoing_requests must be nonnegative with pytest.raises(ValidationError): AutoscalingConfig(target_ongoing_requests=-1) # max_replicas must be greater than or equal to min_replicas # In Pydantic v2, ValueError in validators is wrapped in ValidationError with pytest.raises(ValidationError): AutoscalingConfig(min_replicas=100, max_replicas=1) AutoscalingConfig(min_replicas=1, max_replicas=100) AutoscalingConfig(min_replicas=10, max_replicas=10) # initial_replicas must be greater than or equal to min_replicas # In Pydantic v2, ValueError in validators is wrapped in ValidationError with pytest.raises(ValidationError): AutoscalingConfig(min_replicas=10, initial_replicas=1) with pytest.raises(ValidationError): AutoscalingConfig(min_replicas=10, initial_replicas=1, max_replicas=15) AutoscalingConfig(min_replicas=5, initial_replicas=10, max_replicas=15) AutoscalingConfig(min_replicas=5, initial_replicas=5, max_replicas=15) # initial_replicas must be less than or equal to max_replicas with pytest.raises(ValidationError): AutoscalingConfig(initial_replicas=10, max_replicas=8) with pytest.raises(ValidationError): AutoscalingConfig(min_replicas=1, initial_replicas=10, max_replicas=8) AutoscalingConfig(min_replicas=1, initial_replicas=4, max_replicas=5) AutoscalingConfig(min_replicas=1, initial_replicas=5, max_replicas=5) # Default values should not raise an error default_autoscaling_config = AutoscalingConfig() assert default_autoscaling_config.policy.is_default_policy_function() is True non_default_autoscaling_config = AutoscalingConfig( policy={"policy_function": "ray.serve.tests.unit.test_config:fake_policy"} ) assert non_default_autoscaling_config.policy.is_default_policy_function() is False # look_back_period_s must be greater than metrics_interval_s with pytest.warns(FutureWarning): AutoscalingConfig(look_back_period_s=5.0, metrics_interval_s=10.0) with pytest.warns(FutureWarning): AutoscalingConfig(look_back_period_s=10.0, metrics_interval_s=10.0) AutoscalingConfig(look_back_period_s=30.0, metrics_interval_s=10.0) AutoscalingConfig(look_back_period_s=20.0, metrics_interval_s=10.0) def test_autoscaling_config_metrics_interval_s_deprecation_warning() -> None: """Test that the metrics_interval_s deprecation warning is raised.""" # Warning is raised if we set metrics_interval_s to a non-default value with pytest.warns(DeprecationWarning): AutoscalingConfig(metrics_interval_s=5) # ... even if the AutoscalingConfig is instantiated implicitly via the @serve.deployment decorator with pytest.warns(DeprecationWarning): @serve.deployment(autoscaling_config={"metrics_interval_s": 5}) class Foo: ... # ... or if it is deserialized from proto as part of a DeploymentConfig (presumably in the Serve Controller) deployment_config_proto_bytes = DeploymentConfig( autoscaling_config=AutoscalingConfig(metrics_interval_s=5) ).to_proto_bytes() with pytest.warns(DeprecationWarning): DeploymentConfig.from_proto_bytes(deployment_config_proto_bytes) # Default settings should not raise a warning with warnings.catch_warnings(): warnings.simplefilter("error") AutoscalingConfig() class TestDeploymentConfig: def test_deployment_config_validation(self): # Test config ignoring unknown keys (required for forward-compatibility) DeploymentConfig(new_version_key=-1) # Test num_replicas validation. DeploymentConfig(num_replicas=1) # Pydantic v2 uses different error type names with pytest.raises(ValidationError, match="int_parsing"): DeploymentConfig(num_replicas="hello") with pytest.raises(ValidationError, match="greater_than_equal"): DeploymentConfig(num_replicas=-1) # Test dynamic default for max_ongoing_requests. assert DeploymentConfig().max_ongoing_requests == 5 def test_max_constructor_retry_count_validation(self): # Test max_constructor_retry_count validation. DeploymentConfig(max_constructor_retry_count=1) DeploymentConfig(max_constructor_retry_count=10) # Pydantic v2 uses different error type names with pytest.raises(ValidationError, match="int_parsing"): DeploymentConfig(max_constructor_retry_count="hello") with pytest.raises(ValidationError, match="greater_than"): DeploymentConfig(max_constructor_retry_count=-1) with pytest.raises(ValidationError, match="greater_than"): DeploymentConfig(max_constructor_retry_count=0) # Test default value assert DeploymentConfig().max_constructor_retry_count == 20 def test_deployment_config_update(self): b = DeploymentConfig(num_replicas=1, max_ongoing_requests=1) # Test updating a key works. b.num_replicas = 2 assert b.num_replicas == 2 # Check that not specifying a key doesn't update it. assert b.max_ongoing_requests == 1 # Check that input is validated. with pytest.raises(ValidationError): b.num_replicas = "Hello" with pytest.raises(ValidationError): b.num_replicas = -1 def test_from_default(self): """Check from_default() method behavior.""" # Valid parameters dc = DeploymentConfig.from_default(num_replicas=5, is_cross_language=True) assert dc.num_replicas == 5 assert dc.is_cross_language is True # Invalid parameters should raise TypeError with pytest.raises(TypeError): DeploymentConfig.from_default(num_replicas=5, is_xlang=True) # Validation should still be performed with pytest.raises(ValidationError): DeploymentConfig.from_default(num_replicas="hello world") def test_from_default_ignore_default(self): """Check that from_default() ignores DEFAULT.VALUE kwargs.""" # Valid parameter with DEFAULT.VALUE passed in should be ignored DeploymentConfig.from_default(num_replicas=DEFAULT.VALUE) def test_setting_and_getting_request_router_class(self): """Check that setting and getting request_router_class works.""" # The request_router_class path is derived from the class's __module__ attribute request_router_path = ( f"{FakeRequestRouter.__module__}.{FakeRequestRouter.__name__}" ) # Passing request_router_class as a class. deployment_config = DeploymentConfig.from_default( request_router_config=RequestRouterConfig( request_router_class=FakeRequestRouter ) ) assert ( deployment_config.request_router_config.request_router_class == request_router_path ) assert ( deployment_config.request_router_config.get_request_router_class() == FakeRequestRouter ) # Passing request_router_class as an import path. deployment_config = DeploymentConfig.from_default( request_router_config=RequestRouterConfig( request_router_class=request_router_path ) ) assert ( deployment_config.request_router_config.request_router_class == request_router_path ) assert ( deployment_config.request_router_config.get_request_router_class() == FakeRequestRouter ) # Not passing request_router_class should # default to `PowerOfTwoChoicesRequestRouter`. deployment_config = DeploymentConfig.from_default() assert ( deployment_config.request_router_config.request_router_class == "ray.serve._private.request_router:PowerOfTwoChoicesRequestRouter" ) assert ( deployment_config.request_router_config.get_request_router_class() == PowerOfTwoChoicesRequestRouter ) def test_backoff_params_imperative(self): """Check that custom backoff params are set via the imperative path.""" custom_initial = 0.1 custom_multiplier = 3.0 custom_max = 2.0 deployment_config = DeploymentConfig.from_default( request_router_config=RequestRouterConfig( initial_backoff_s=custom_initial, backoff_multiplier=custom_multiplier, max_backoff_s=custom_max, ) ) assert ( deployment_config.request_router_config.initial_backoff_s == custom_initial ) assert ( deployment_config.request_router_config.backoff_multiplier == custom_multiplier ) assert deployment_config.request_router_config.max_backoff_s == custom_max def test_backoff_params_defaults_imperative(self): """Check that backoff params use defaults when not specified.""" deployment_config = DeploymentConfig.from_default() assert ( deployment_config.request_router_config.initial_backoff_s == RAY_SERVE_ROUTER_RETRY_INITIAL_BACKOFF_S ) assert ( deployment_config.request_router_config.backoff_multiplier == RAY_SERVE_ROUTER_RETRY_BACKOFF_MULTIPLIER ) assert ( deployment_config.request_router_config.max_backoff_s == RAY_SERVE_ROUTER_RETRY_MAX_BACKOFF_S ) def test_backoff_params_declarative_schema(self): """Check that backoff params can be set via the declarative schema.""" schema = DeploymentSchema( name="test-deployment", request_router_config=RequestRouterConfig( initial_backoff_s=0.1, backoff_multiplier=3.0, max_backoff_s=2.0, ), ) assert schema.request_router_config.initial_backoff_s == 0.1 assert schema.request_router_config.backoff_multiplier == 3.0 assert schema.request_router_config.max_backoff_s == 2.0 def test_deployment_actors_config(self): """Test deployment_actors config and proto roundtrip.""" @ray.remote class DummyActor: pass actor_config = DeploymentActorConfig( name="prefix_tree", actor_class=DummyActor, init_kwargs={"max_depth": 100}, actor_options={"num_cpus": 0.1}, ) config = DeploymentConfig( num_replicas=1, deployment_actors=[actor_config], ) assert config.deployment_actors is not None assert len(config.deployment_actors) == 1 assert config.deployment_actors[0].name == "prefix_tree" assert isinstance(config.deployment_actors[0].actor_class, str) assert config.deployment_actors[0]._serialized_actor_class assert ( config.deployment_actors[0].get_actor_class().__ray_actor_class__.__name__ == "DummyActor" ) assert config.deployment_actors[0].init_kwargs == {"max_depth": 100} deserialized = DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) assert deserialized.deployment_actors is not None assert len(deserialized.deployment_actors) == 1 assert deserialized.deployment_actors[0].name == "prefix_tree" assert isinstance(deserialized.deployment_actors[0].actor_class, str) assert deserialized.deployment_actors[0]._serialized_actor_class assert ( deserialized.deployment_actors[0] .get_actor_class() .__ray_actor_class__.__name__ == "DummyActor" ) assert deserialized.deployment_actors[0].init_kwargs == {"max_depth": 100} def test_deployment_actors_config_duplicate_names_raise(self): """Test that duplicate deployment_actor names raise ValueError.""" with pytest.raises(ValueError, match="unique names"): DeploymentConfig( num_replicas=1, deployment_actors=[ DeploymentActorConfig( name="dup", actor_class=_TestDummyActor, init_kwargs={}, ), DeploymentActorConfig( name="dup", actor_class=_TestDummyActor, init_kwargs={}, ), ], ) def test_deployment_actors_config_import_path(self): """actor_class stays as string until _serialize_actor_class() is called (happens in the build task where user code is importable). """ actor_config_str = DeploymentActorConfig( name="actor_from_path", actor_class="ray.serve.tests.unit.test_config._TestDummyActor", init_kwargs={"max_depth": 50}, ) assert isinstance(actor_config_str.actor_class, str) assert not actor_config_str._serialized_actor_class # Simulate what build_serve_application does actor_config_str._serialize_actor_class() assert actor_config_str._serialized_actor_class config_str = DeploymentConfig( num_replicas=1, deployment_actors=[actor_config_str], ) proto = config_str.to_proto() assert len(proto.deployment_actors) == 1 assert proto.deployment_actors[0].name == "actor_from_path" assert proto.deployment_actors[0].actor_class_name != "" deserialized_str = DeploymentConfig.from_proto_bytes( config_str.to_proto_bytes() ) resolved_str = deserialized_str.deployment_actors[0].get_actor_class() assert ( resolved_str.__ray_actor_class__.__name__ == _TestDummyActor.__ray_actor_class__.__name__ ) def test_proto_roundtrip_preserves_actor_class(self): """DeploymentActorConfig survives proto serialization and can reconstruct the actor class via get_actor_class(). """ cfg = DeploymentActorConfig( name="counter", actor_class=_TestRayActor, init_kwargs={}, ) dc = DeploymentConfig(num_replicas=1, deployment_actors=[cfg]) deserialized = DeploymentConfig.from_proto_bytes(dc.to_proto_bytes()) actor_cfg = deserialized.deployment_actors[0] assert actor_cfg._serialized_actor_class assert isinstance(actor_cfg.actor_class, str) resolved = actor_cfg.get_actor_class() assert resolved.__ray_actor_class__.__name__ == "_TestRayActor" # Verify we can instantiate and invoke methods (class serialized properly) underlying = resolved.__ray_actor_class__ instance = underlying() assert instance.ping() == "pong" class TestReplicaConfig: def test_basic_validation(self): class Class: pass def function(_): pass ReplicaConfig.create(Class) ReplicaConfig.create(function) with pytest.raises(TypeError): ReplicaConfig.create(Class()) def test_ray_actor_options_validation(self): class Class: pass ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={ "num_cpus": 1.0, "num_gpus": 10, "resources": {"abc": 1.0}, "memory": 1000000.0, }, ) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options=1.0) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options=False) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options={"num_cpus": "hello"}) with pytest.raises(ValueError): ReplicaConfig.create(Class, ray_actor_options={"num_cpus": -1}) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options={"num_gpus": "hello"}) with pytest.raises(ValueError): ReplicaConfig.create(Class, ray_actor_options={"num_gpus": -1}) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options={"memory": "hello"}) with pytest.raises(ValueError): ReplicaConfig.create(Class, ray_actor_options={"memory": -1}) with pytest.raises(TypeError): ReplicaConfig.create(Class, ray_actor_options={"resources": []}) disallowed_ray_actor_options = { "max_concurrency", "max_restarts", "max_task_retries", "name", "namespace", "lifetime", "placement_group", "placement_group_bundle_index", "placement_group_capture_child_tasks", "max_pending_calls", "scheduling_strategy", "get_if_exists", "_metadata", } for option in disallowed_ray_actor_options: with pytest.raises(ValueError): ReplicaConfig.create(Class, ray_actor_options={option: None}) def test_max_replicas_per_node_validation(self): class Class: pass ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=5, ) # Invalid type with pytest.raises(TypeError, match="Get invalid type"): ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node="1", ) # Invalid: not in the range of [1, 100] with pytest.raises( ValueError, match=r"Valid values are None or an integer in the range of \[1, 100\]", ): ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=0, ) with pytest.raises( ValueError, match=r"Valid values are None or an integer in the range of \[1, 100\]", ): ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=110, ) with pytest.raises( ValueError, match=r"Valid values are None or an integer in the range of \[1, 100\]", ): ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=-1, ) def test_placement_group_options_validation(self): class Class: pass # Specify placement_group_bundles without num_cpus or placement_group_strategy. ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1.0}], ) # Specify placement_group_bundles with integer value. ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}], ) # Specify placement_group_bundles and placement_group_strategy. ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1.0}], placement_group_strategy="STRICT_PACK", ) # Specify placement_group_bundles and placement_group_strategy and num_cpus. ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={"num_cpus": 1}, placement_group_bundles=[{"CPU": 1.0}], placement_group_strategy="STRICT_PACK", ) # Invalid: placement_group_strategy without placement_group_bundles. with pytest.raises( ValueError, match="`placement_group_bundles` must also be provided" ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_strategy="PACK", ) # Invalid: unsupported placement_group_strategy. with pytest.raises( ValueError, match="Invalid placement group strategy FAKE_NEWS" ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1.0}], placement_group_strategy="FAKE_NEWS", ) # Invalid: malformed placement_group_bundles. with pytest.raises( ValueError, match=("Bundles must be a non-empty list of resource dictionaries."), ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": "1.0"}], ) # Invalid: invalid placement_group_bundles. with pytest.raises( ValueError, match="cannot be an empty dictionary or resources with only 0", ): ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={"num_cpus": 0, "num_gpus": 0}, placement_group_bundles=[{"CPU": 0, "GPU": 0}], ) # Invalid: replica actor does not fit in the first bundle (CPU). with pytest.raises( ValueError, match=( "the resource requirements for the actor must be a " "subset of the first bundle." ), ): ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={"num_cpus": 1}, placement_group_bundles=[{"CPU": 0.1}], ) # Invalid: replica actor does not fit in the first bundle (CPU). with pytest.raises( ValueError, match=( "the resource requirements for the actor must be a " "subset of the first bundle." ), ): ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={"num_gpus": 1}, placement_group_bundles=[{"CPU": 1.0}], ) # Invalid: replica actor does not fit in the first bundle (custom resource). with pytest.raises( ValueError, match=( "the resource requirements for the actor must be a " "subset of the first bundle." ), ): ReplicaConfig.create( Class, tuple(), dict(), ray_actor_options={"resources": {"custom": 1}}, placement_group_bundles=[{"CPU": 1}], ) def test_mutually_exclusive_max_replicas_per_node_and_placement_group_bundles(self): class Class: pass ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=5, ) ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1.0}], ) with pytest.raises( ValueError, match=( "Setting max_replicas_per_node is not allowed when " "placement_group_bundles is provided." ), ): ReplicaConfig.create( Class, tuple(), dict(), max_replicas_per_node=5, placement_group_bundles=[{"CPU": 1.0}], ) with pytest.raises( ValueError, match=( "Setting max_replicas_per_node is not allowed when " "placement_group_bundles is provided." ), ): config = ReplicaConfig.create(Class, tuple(), dict()) config.update( ray_actor_options={}, max_replicas_per_node=5, placement_group_bundles=[{"CPU": 1.0}], ) def test_replica_config_lazy_deserialization(self): def f(): return "Check this out!" f_serialized = cloudpickle.dumps(f) config = ReplicaConfig( "f", f_serialized, cloudpickle.dumps(()), cloudpickle.dumps({}), {} ) assert config.serialized_deployment_def == f_serialized assert config._deployment_def is None assert config.serialized_init_args == cloudpickle.dumps(tuple()) assert config._init_args is None assert config.serialized_init_kwargs == cloudpickle.dumps(dict()) assert config._init_kwargs is None assert isinstance(config.ray_actor_options, dict) assert isinstance(config.resource_dict, dict) assert config.deployment_def() == "Check this out!" assert config.init_args == tuple() assert config.init_kwargs == dict() def test_placement_group_bundle_label_selector_validation(self): class Class: pass # Label selector provided without bundles with pytest.raises( ValueError, match="If `placement_group_bundle_label_selector` is provided, `placement_group_bundles` must also be provided.", ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundle_label_selector=[{"gpu": "T4"}], ) # bundle_label_selector list does not match bundles list length with pytest.raises( ValueError, match="The length of `bundle_label_selector` should equal the length of `bundles`", ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}, {"CPU": 1}, {"CPU": 1}], placement_group_bundle_label_selector=[{"gpu": "T4"}, {"gpu": "L4"}], ) # Valid config - multiple bundles provided for one bundle_label_selector. config = ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}, {"CPU": 1}, {"CPU": 1}], placement_group_bundle_label_selector=[{"gpu": "T4"}], ) assert config.placement_group_bundle_label_selector == [ {"gpu": "T4"}, {"gpu": "T4"}, {"gpu": "T4"}, ] # Valid config - multiple bundles and an equal number of bundle label selectors. config = ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}, {"CPU": 1}], placement_group_bundle_label_selector=[{"gpu": "T4"}, {"gpu": "L4"}], ) assert config.placement_group_bundle_label_selector == [ {"gpu": "T4"}, {"gpu": "L4"}, ] def test_placement_group_fallback_strategy_validation(self): class Class: pass # Validate that fallback strategy provided without bundles raises error. with pytest.raises( ValueError, match="If `placement_group_fallback_strategy` is provided, `placement_group_bundles` must also be provided.", ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_fallback_strategy=[{"bundles": [{"CPU": 1}]}], ) # Validate that fallback strategy is a list with pytest.raises( TypeError, match="placement_group_fallback_strategy must be a list of dictionaries.", ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}], placement_group_fallback_strategy="not_a_list", ) # Fallback strategy list contains non-dict items with pytest.raises( TypeError, match="placement_group_fallback_strategy entry at index 1 must be a dictionary.", ): ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}], placement_group_fallback_strategy=[ {"bundles": [{"CPU": 1}]}, "invalid_entry", ], ) # Valid config config = ReplicaConfig.create( Class, tuple(), dict(), placement_group_bundles=[{"CPU": 1}], placement_group_fallback_strategy=[{"bundles": [{"CPU": 1}]}], ) assert config.placement_group_fallback_strategy == [{"bundles": [{"CPU": 1}]}] class TestAutoscalingConfig: def test_target_ongoing_requests(self): autoscaling_config = AutoscalingConfig() assert autoscaling_config.get_target_ongoing_requests() == 2 autoscaling_config = AutoscalingConfig(target_ongoing_requests=7) assert autoscaling_config.get_target_ongoing_requests() == 7 def test_scaling_factor(self): autoscaling_config = AutoscalingConfig() assert autoscaling_config.get_upscaling_factor() == 1 assert autoscaling_config.get_downscaling_factor() == 1 autoscaling_config = AutoscalingConfig(smoothing_factor=0.4) assert autoscaling_config.get_upscaling_factor() == 0.4 assert autoscaling_config.get_downscaling_factor() == 0.4 autoscaling_config = AutoscalingConfig(upscale_smoothing_factor=0.4) assert autoscaling_config.get_upscaling_factor() == 0.4 assert autoscaling_config.get_downscaling_factor() == 1 autoscaling_config = AutoscalingConfig(downscale_smoothing_factor=0.4) assert autoscaling_config.get_upscaling_factor() == 1 assert autoscaling_config.get_downscaling_factor() == 0.4 autoscaling_config = AutoscalingConfig( smoothing_factor=0.4, upscale_smoothing_factor=0.1, downscale_smoothing_factor=0.01, ) assert autoscaling_config.get_upscaling_factor() == 0.1 assert autoscaling_config.get_downscaling_factor() == 0.01 autoscaling_config = AutoscalingConfig( smoothing_factor=0.4, upscaling_factor=0.5, downscaling_factor=0.6, ) assert autoscaling_config.get_upscaling_factor() == 0.5 assert autoscaling_config.get_downscaling_factor() == 0.6 autoscaling_config = AutoscalingConfig( smoothing_factor=0.4, upscale_smoothing_factor=0.1, downscale_smoothing_factor=0.01, upscaling_factor=0.5, downscaling_factor=0.6, ) assert autoscaling_config.get_upscaling_factor() == 0.5 assert autoscaling_config.get_downscaling_factor() == 0.6 class TestGangSchedulingConfig: def test_gang_scheduling_config_validation(self): """Test GangSchedulingConfig field validation.""" with pytest.raises(ValidationError): GangSchedulingConfig() # gang_size must be >= 1 with pytest.raises(ValidationError): GangSchedulingConfig(gang_size=0) with pytest.raises(ValidationError): GangSchedulingConfig(gang_size=-1) config = GangSchedulingConfig(gang_size=1) assert config.gang_size == 1 config = GangSchedulingConfig(gang_size=4) assert config.gang_size == 4 def test_gang_scheduling_config_defaults(self): """Test GangSchedulingConfig default values.""" config = GangSchedulingConfig(gang_size=4) assert config.gang_placement_strategy == GangPlacementStrategy.PACK assert config.runtime_failure_policy == GangRuntimeFailurePolicy.RESTART_GANG def test_gang_scheduling_config_custom_values(self): """Test GangSchedulingConfig with custom values.""" config = GangSchedulingConfig( gang_size=8, gang_placement_strategy=GangPlacementStrategy.SPREAD, ) assert config.gang_size == 8 assert config.gang_placement_strategy == GangPlacementStrategy.SPREAD assert config.runtime_failure_policy == GangRuntimeFailurePolicy.RESTART_GANG def test_gang_placement_strategy_options(self): """Test all GangPlacementStrategy options are valid.""" for strategy in GangPlacementStrategy: config = GangSchedulingConfig(gang_size=4, gang_placement_strategy=strategy) assert config.gang_placement_strategy == strategy def test_gang_runtime_failure_policy_options(self): """Test all GangRuntimeFailurePolicy options are valid.""" # RESTART_GANG should work. config = GangSchedulingConfig( gang_size=4, runtime_failure_policy=GangRuntimeFailurePolicy.RESTART_GANG, ) assert config.runtime_failure_policy == GangRuntimeFailurePolicy.RESTART_GANG # RESTART_REPLICA is not yet implemented. with pytest.raises(NotImplementedError): GangSchedulingConfig( gang_size=4, runtime_failure_policy=GangRuntimeFailurePolicy.RESTART_REPLICA, ) def test_gang_scheduling_config_via_decorator_error(self): """Test that gang_scheduling_config validation errors are raised.""" with pytest.raises( ValueError, match="num_replicas.*must be a multiple of gang_size" ): @serve.deployment(gang_scheduling_config=GangSchedulingConfig(gang_size=4)) def f(): return "test" def test_gang_scheduling_config_scale_to_zero_rejected(self): """Test that min_replicas=0 is rejected with gang_scheduling_config.""" with pytest.raises( ValueError, match="Scale to zero isn't supported for gang-scheduled deployments", ): @serve.deployment( num_replicas="auto", gang_scheduling_config=GangSchedulingConfig(gang_size=3), autoscaling_config={"min_replicas": 0, "max_replicas": 9}, ) def f(): return "test" def test_gang_scheduling_config_auto_num_replicas(self): """Test that num_replicas='auto' is allowed with gang_scheduling_config.""" @serve.deployment( num_replicas="auto", gang_scheduling_config=GangSchedulingConfig(gang_size=4), autoscaling_config={"min_replicas": 4, "max_replicas": 8}, ) def f(): return "test" assert f._deployment_config.autoscaling_config is not None assert f._deployment_config.gang_scheduling_config.gang_size == 4 assert f._deployment_config.autoscaling_config.min_replicas == 4 assert f._deployment_config.autoscaling_config.max_replicas == 8 def test_gang_scheduling_config_auto_num_replicas_via_options(self): """Test that num_replicas='auto' works via .options() with gang config.""" @serve.deployment( num_replicas=4, gang_scheduling_config=GangSchedulingConfig(gang_size=4), ) def f(): return "test" f2 = f.options( num_replicas="auto", autoscaling_config={"min_replicas": 4, "max_replicas": 8}, ) assert f2._deployment_config.autoscaling_config is not None assert f._deployment_config.gang_scheduling_config.gang_size == 4 assert f2._deployment_config.autoscaling_config.min_replicas == 4 assert f2._deployment_config.autoscaling_config.max_replicas == 8 def test_gang_scheduling_config_proto_roundtrip(self): """Test roundtrip serialization of GangSchedulingConfig through protobuf.""" # Test with gang_scheduling_config config = DeploymentConfig( num_replicas=8, gang_scheduling_config=GangSchedulingConfig( gang_size=4, gang_placement_strategy=GangPlacementStrategy.SPREAD, runtime_failure_policy=GangRuntimeFailurePolicy.RESTART_GANG, ), ) deserialized = DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) assert deserialized.gang_scheduling_config is not None assert deserialized.gang_scheduling_config.gang_size == 4 assert ( deserialized.gang_scheduling_config.gang_placement_strategy == GangPlacementStrategy.SPREAD ) assert ( deserialized.gang_scheduling_config.runtime_failure_policy == GangRuntimeFailurePolicy.RESTART_GANG ) # Test without gang_scheduling_config config = DeploymentConfig(num_replicas=2) deserialized = DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) assert deserialized.gang_scheduling_config is None def test_gang_scheduling_config_via_decorator(self): """Test that gang_scheduling_config can be passed via @serve.deployment decorator.""" @serve.deployment( num_replicas=8, gang_scheduling_config=GangSchedulingConfig(gang_size=4) ) def f(): return "test" # Verify the config is properly set assert f._deployment_config.gang_scheduling_config is not None assert f._deployment_config.gang_scheduling_config.gang_size == 4 def test_gang_scheduling_config_invalid_num_replicas_via_options(self): @serve.deployment( num_replicas=4, gang_scheduling_config=GangSchedulingConfig(gang_size=2) ) def f(): pass with pytest.raises(ValueError, match="must be a multiple of gang_size"): f.options(num_replicas=5) with pytest.raises(ValueError, match="must be a multiple of gang_size"): f.options(num_replicas=3) d = f.options(num_replicas=6) assert d.num_replicas == 6 def test_gang_scheduling_config_invalid_gang_size_via_options(self): @serve.deployment( num_replicas=4, gang_scheduling_config=GangSchedulingConfig(gang_size=2) ) def f(): pass with pytest.raises(ValueError, match="must be a multiple of gang_size"): f.options(gang_scheduling_config=GangSchedulingConfig(gang_size=3)) d = f.options(gang_scheduling_config=GangSchedulingConfig(gang_size=4)) assert d._deployment_config.gang_scheduling_config.gang_size == 4 def test_config_schemas_forward_compatible(): # Test configs ignoring unknown keys (required for forward-compatibility) ServeDeploySchema( http_options=HTTPOptionsSchema( new_version_config_key="this config is from newer version of Ray" ), applications=[ ServeApplicationSchema( import_path="module.app", deployments=[ DeploymentSchema( name="deployment", new_version_config_key="this config is from newer version" " of Ray", ) ], new_version_config_key="this config is from newer version of Ray", ), ], new_version_config_key="this config is from newer version of Ray", ) def test_http_options(): HTTPOptions() # `middlewares` is removed: a non-empty list raises, but an empty list is # a no-op (matches the prior warn-on-non-empty behavior) so internal # normalization via model_copy/defaults does not break. HTTPOptions(middlewares=[]) with pytest.raises(ValueError, match="`middlewares` in HTTPOptions"): HTTPOptions(host="8.8.8.8", middlewares=[object()]) # `num_cpus` is removed: a non-zero value raises; 0 (the old default) is # a no-op so default construction is unaffected. HTTPOptions(num_cpus=0) with pytest.raises(ValueError, match="`num_cpus` in HTTPOptions"): HTTPOptions(num_cpus=2) # Test configs ignoring unknown keys (required for forward-compatibility) HTTPOptions(new_version_config_key="this config is from newer version of Ray") # `location` is deprecated; it defaults to None (proxy_location is the # authority). host=None still disables. Setting a non-None location warns; # location=None is a no-op (internal model_dump roundtrips pass it) and must # NOT warn. assert HTTPOptions().location is None assert HTTPOptions(host=None).location == ProxyLocation.Disabled with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") assert HTTPOptions(location=None).location is None assert not any("`location` in HTTPOptions" in str(w.message) for w in caught) with pytest.warns(DeprecationWarning, match="`location` in HTTPOptions"): assert HTTPOptions(location=ProxyLocation.EveryNode).location == "EveryNode" def test_with_proto(): # Test roundtrip config = DeploymentConfig(num_replicas=100, max_ongoing_requests=16) assert config == DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) # Test user_config object config = DeploymentConfig(user_config={"python": ("native", ["objects"])}) assert config == DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) def test_rolling_update_percentage_proto_roundtrip(): """Ensure `rolling_update_percentage` survives to_proto/from_proto. Because the proto field is declared `optional double`, an explicit value must round-trip losslessly, and an absent field (simulating an older controller during a rolling upgrade) must fall back to the Python-level default instead of a proto3 zero default. """ # Explicit non-default value survives the round-trip. config = DeploymentConfig(rolling_update_percentage=0.5) roundtripped = DeploymentConfig.from_proto_bytes(config.to_proto_bytes()) assert roundtripped.rolling_update_percentage == 0.5 # Simulate an older controller that didn't carry this field: start from a # valid proto (so the other non-optional fields satisfy Pydantic's # validators on deserialization) and clear just `rolling_update_percentage`. # Clearing the optional field makes HasField() return False, mimicking a # proto serialized before this field existed. proto = DeploymentConfig().to_proto() proto.ClearField("rolling_update_percentage") assert not proto.HasField("rolling_update_percentage") deserialized = DeploymentConfig.from_proto(proto) assert deserialized.rolling_update_percentage == DEFAULT_ROLLING_UPDATE_PERCENTAGE @pytest.mark.parametrize("use_deprecated_smoothing_factor", [True, False]) def test_zero_default_proto(use_deprecated_smoothing_factor): # Test that options set to zero (protobuf default value) still retain their # original value after being serialized and deserialized. autoscaling_config = { "min_replicas": 1, "max_replicas": 2, "downscale_delay_s": 0, } if use_deprecated_smoothing_factor: autoscaling_config["smoothing_factor"] = 0.123 else: autoscaling_config["upscaling_factor"] = 0.123 autoscaling_config["downscaling_factor"] = 0.123 config = DeploymentConfig(autoscaling_config=autoscaling_config) serialized_config = config.to_proto_bytes() deserialized_config = DeploymentConfig.from_proto_bytes(serialized_config) new_delay_s = deserialized_config.autoscaling_config.downscale_delay_s assert new_delay_s == 0 # Check that this test is not spuriously passing. default_downscale_delay_s = AutoscalingConfig().downscale_delay_s assert new_delay_s != default_downscale_delay_s def test_grpc_options(): """Test gRPCOptions. When the gRPCOptions object is created, the default values are set correctly. When the gRPCOptions object is created with user-specified values, the values are set correctly. Also, if the user provided an invalid grpc_servicer_function, it raises errors. """ default_grpc_options = gRPCOptions() assert default_grpc_options.port == DEFAULT_GRPC_PORT assert default_grpc_options.grpc_servicer_functions == [] assert default_grpc_options.grpc_servicer_func_callable == [] assert default_grpc_options.request_timeout_s is None port = 9001 grpc_servicer_functions = [ "ray.serve.generated.serve_pb2_grpc.add_UserDefinedServiceServicer_to_server", ] request_timeout_s = 1 grpc_options = gRPCOptions( port=port, grpc_servicer_functions=grpc_servicer_functions, request_timeout_s=request_timeout_s, ) assert grpc_options.port == port assert grpc_options.grpc_servicer_functions == grpc_servicer_functions assert grpc_options.grpc_servicer_func_callable == [ add_UserDefinedServiceServicer_to_server ] assert grpc_options.request_timeout_s == request_timeout_s # Import not found should raise ModuleNotFoundError. grpc_servicer_functions = ["fake.service.that.does.not.exist"] with pytest.raises(ModuleNotFoundError) as exception: grpc_options = gRPCOptions(grpc_servicer_functions=grpc_servicer_functions) _ = grpc_options.grpc_servicer_func_callable assert "can't be imported!" in str(exception) # Not callable should raise ValueError. grpc_servicer_functions = ["ray.serve._private.constants.DEFAULT_HTTP_PORT"] with pytest.raises(ValueError) as exception: grpc_options = gRPCOptions(grpc_servicer_functions=grpc_servicer_functions) _ = grpc_options.grpc_servicer_func_callable assert "is not a callable function!" in str(exception) class TestControllerOptions: """Validator + parsing coverage for ControllerOptions. The runtime validator is intentionally strict: v0 only accepts ``runtime_env.env_vars``. Other ``runtime_env`` keys (pip, working_dir, ...) would mutate the long-lived detached controller actor's dependencies and are rejected with a message pointing at deployment-level runtime_env. """ def test_default_is_empty(self): opts = ControllerOptions() assert opts.runtime_env is None def test_accepts_dict_from_model_validate(self): opts = ControllerOptions.model_validate( {"runtime_env": {"env_vars": {"X": "y"}}} ) assert opts.runtime_env == {"env_vars": {"X": "y"}} def test_accepts_env_vars_with_str_str(self): opts = ControllerOptions( runtime_env={ "env_vars": { "RAY_SERVE_HAPROXY_TCP_NODELAY": "1", "RAY_SERVE_HAPROXY_NBTHREAD": "16", } } ) assert opts.runtime_env["env_vars"]["RAY_SERVE_HAPROXY_TCP_NODELAY"] == "1" def test_accepts_empty_env_vars(self): opts = ControllerOptions(runtime_env={"env_vars": {}}) assert opts.runtime_env == {"env_vars": {}} def test_accepts_explicit_none(self): opts = ControllerOptions(runtime_env=None) assert opts.runtime_env is None @pytest.mark.parametrize( "disallowed_key, value", [ ("pip", ["numpy"]), ("working_dir", "/tmp"), ("py_modules", []), ("conda", "env.yaml"), ("container", {}), # A future runtime_env key we'd want to land via an explicit # API broadening, not by silently accepting it. ("nsight", "default"), ], ) def test_rejects_non_env_vars_runtime_env_keys(self, disallowed_key, value): with pytest.raises(ValidationError) as exc: ControllerOptions(runtime_env={disallowed_key: value}) msg = str(exc.value) assert "only supports ['env_vars']" in msg assert disallowed_key in msg def test_rejects_runtime_env_not_a_dict(self): with pytest.raises(ValidationError): ControllerOptions(runtime_env="env_vars=FOO") def test_rejects_env_vars_not_a_dict(self): with pytest.raises(ValidationError) as exc: ControllerOptions(runtime_env={"env_vars": ["FOO=bar"]}) assert "env_vars must be a dict" in str(exc.value) def test_rejects_env_vars_explicit_none(self): # Explicit ``env_vars: null`` (e.g., from YAML) is distinct from the # key being absent; reject it so a bad config fails locally instead of # surfacing later from the Ray runtime_env layer. with pytest.raises(ValidationError) as exc: ControllerOptions(runtime_env={"env_vars": None}) assert "env_vars must be a dict" in str(exc.value) @pytest.mark.parametrize( "bad_value", [1, 3.14, True, None, ["1"], {"nested": "value"}], ) def test_rejects_non_str_env_var_values(self, bad_value): with pytest.raises(ValidationError) as exc: ControllerOptions(runtime_env={"env_vars": {"X": bad_value}}) assert "must be str" in str(exc.value) @pytest.mark.parametrize("bad_key", ["", 1, None]) def test_rejects_bad_env_var_keys(self, bad_key): with pytest.raises(ValidationError): ControllerOptions(runtime_env={"env_vars": {bad_key: "v"}}) def test_rejects_extra_top_level_fields(self): # extra="forbid" guards against typos and against silently accepting # future fields without a validator update. with pytest.raises(ValidationError): ControllerOptions(runtimeenv={"env_vars": {}}) # missing underscore def test_rejects_mixed_allowed_and_disallowed_runtime_env_keys(self): with pytest.raises(ValidationError) as exc: ControllerOptions(runtime_env={"env_vars": {"X": "y"}, "pip": ["numpy"]}) assert "pip" in str(exc.value) def test_proxy_location_normalize(): assert ProxyLocation._normalize(None) is None assert ProxyLocation._normalize(ProxyLocation.Disabled) == ProxyLocation.Disabled assert ProxyLocation._normalize(ProxyLocation.HeadOnly) == ProxyLocation.HeadOnly assert ProxyLocation._normalize(ProxyLocation.EveryNode) == ProxyLocation.EveryNode assert ProxyLocation._normalize("Disabled") == ProxyLocation.Disabled assert ProxyLocation._normalize("HeadOnly") == ProxyLocation.HeadOnly assert ProxyLocation._normalize("EveryNode") == ProxyLocation.EveryNode assert ProxyLocation._normalize("NoServer") == ProxyLocation.Disabled with pytest.raises(ValueError): ProxyLocation._normalize("Unknown") with pytest.raises(TypeError): ProxyLocation._normalize({"some_other_obj"}) @pytest.mark.parametrize( "policy", [ None, {"policy_function": "ray.serve.tests.unit.test_config:fake_policy"}, {"policy_function": fake_policy}, ], ) def test_autoscaling_policy_serializations(policy): """Test that autoscaling policy can be serialized and deserialized. This test checks that the autoscaling policy can be serialized and deserialized for when the policy is a function, a string, or None (default). """ autoscaling_config = AutoscalingConfig() if policy: autoscaling_config = AutoscalingConfig(policy=policy) config = DeploymentConfig.from_default(autoscaling_config=autoscaling_config) deserialized_autoscaling_policy = DeploymentConfig.from_proto_bytes( config.to_proto_bytes() ).autoscaling_config.policy.get_policy() if policy is None: # Compare function attributes instead of function objects since # cloudpickle.register_pickle_by_value() causes deserialization to # create a new function object rather than returning the same object assert ( deserialized_autoscaling_policy.__name__ == default_autoscaling_policy.__name__ ) assert ( deserialized_autoscaling_policy.__module__ == default_autoscaling_policy.__module__ ) else: # Compare function behavior instead of function objects # since serialization/deserialization creates new function objects assert deserialized_autoscaling_policy() == fake_policy() def test_autoscaling_policy_import_fails_for_non_existing_policy(): """Test that autoscaling policy will error out for non-existing policy. This test will ensure non-existing policy will be caught. It can happen when we moved the default policy or when user pass in a non-existing policy. """ # Right now we don't allow modifying the autoscaling policy, so this will not fail policy = "i.dont.exist:fake_policy" with pytest.raises(ModuleNotFoundError): AutoscalingConfig(policy={"policy_function": policy}) def test_default_autoscaling_policy_import_path(): """Test that default autoscaling policy can be imported.""" policy = import_attr(DEFAULT_AUTOSCALING_POLICY_NAME) assert policy == default_autoscaling_policy class TestGetControllerImpl: """White-box checks that ``get_controller_impl`` wires ``ControllerOptions`` into the controller actor class's default options. These cover the path without starting a Ray cluster -- the live end-to-end env-propagation test lives in ``tests/test_standalone.py::test_serve_start_controller_options``. """ def _default_options(self, controller_options=None): from ray.serve._private.default_impl import get_controller_impl return get_controller_impl( controller_options=controller_options )._default_options def test_no_runtime_env_key_when_options_is_none(self): opts = self._default_options(controller_options=None) # Hardcoded fields stay; no runtime_env unless explicitly requested. assert opts["name"] == "SERVE_CONTROLLER_ACTOR" assert opts["namespace"] == "serve" assert "runtime_env" not in opts def test_no_runtime_env_key_when_runtime_env_is_none(self): opts = self._default_options(ControllerOptions(runtime_env=None)) assert "runtime_env" not in opts def test_env_vars_passthrough(self): opts = self._default_options( ControllerOptions( runtime_env={ "env_vars": { "RAY_SERVE_HAPROXY_TCP_NODELAY": "1", "RAY_SERVE_HAPROXY_NBTHREAD": "16", } } ) ) assert opts["runtime_env"]["env_vars"] == { "RAY_SERVE_HAPROXY_TCP_NODELAY": "1", "RAY_SERVE_HAPROXY_NBTHREAD": "16", } class TestProtoToDict: def test_empty_fields(self): """Test _proto_to_dict() to deserialize protobuf with empty fields""" proto = DeploymentConfigProto() result = _proto_to_dict(proto) # Defaults are filled. assert result["num_replicas"] == 0 assert result["max_ongoing_requests"] == 0 assert result["user_config"] == b"" assert result["user_configured_option_names"] == [] # Nested profobufs don't exist. assert "autoscaling_config" not in result def test_non_empty_fields(self): """Test _proto_to_dict() to deserialize protobuf with non-empty fields""" num_replicas = 111 max_ongoing_requests = 222 proto = DeploymentConfigProto( num_replicas=num_replicas, max_ongoing_requests=max_ongoing_requests, ) result = _proto_to_dict(proto) # Fields with non-empty values are filled correctly. assert result["num_replicas"] == num_replicas assert result["max_ongoing_requests"] == max_ongoing_requests # Empty fields are continue to be filled with default values. assert result["user_config"] == b"" def test_nested_protobufs(self): """Test _proto_to_dict() to deserialize protobuf with nested protobufs""" num_replicas = 111 max_ongoing_requests = 222 min_replicas = 333 proto = DeploymentConfigProto( num_replicas=num_replicas, max_ongoing_requests=max_ongoing_requests, autoscaling_config=AutoscalingConfigProto( min_replicas=min_replicas, ), ) result = _proto_to_dict(proto) # Non-empty field is filled correctly. assert result["num_replicas"] == num_replicas assert result["max_ongoing_requests"] == max_ongoing_requests # Nested protobuf is filled correctly. assert result["autoscaling_config"]["min_replicas"] == min_replicas def test_repeated_field(self): """Test _proto_to_dict() to deserialize protobuf with repeated field""" user_configured_option_names = ["foo", "bar"] config = DeploymentConfig.from_default( user_configured_option_names=user_configured_option_names, ) proto_bytes = config.to_proto_bytes() proto = DeploymentConfigProto.FromString(proto_bytes) result = _proto_to_dict(proto) # Repeated field is filled correctly as list. assert set(result["user_configured_option_names"]) == set( user_configured_option_names ) assert isinstance(result["user_configured_option_names"], list) def test_enum_field(self): """Test _proto_to_dict() to deserialize protobuf with enum field""" proto = DeploymentConfigProto( deployment_language=DeploymentLanguage.JAVA, ) result = _proto_to_dict(proto) # Enum field is filled correctly. assert result["deployment_language"] == DeploymentLanguage.JAVA def test_optional_field(self): """Test _proto_to_dict() to deserialize protobuf with optional field""" min_replicas = 1 proto = AutoscalingConfigProto( min_replicas=min_replicas, ) result = _proto_to_dict(proto) # Non-empty field is filled correctly. assert result["min_replicas"] == 1 # Empty field is filled correctly. assert result["max_replicas"] == 0 # Optional field should not be filled. assert "initial_replicas" not in result if __name__ == "__main__": import sys sys.exit(pytest.main(["-v", "-s", __file__]))