test now finally works

services/autoscaling/tests/unit/test_modules_auto_scaling_dynamic.py

@@ -15,7 +15,6 @@
 from dataclasses import dataclass
 from typing import Any, cast
 from unittest import mock
-import unittest
 
 import aiodocker
 import arrow
@@ -194,7 +193,8 @@ def minimal_configuration(
     disable_dynamic_service_background_task: None,
     disable_buffers_pool_background_task: None,
     mocked_redis_server: None,
-) -> None: ...
+) -> None:
+    ...
 
 
 def _assert_rabbit_autoscaling_message_sent(
@@ -574,11 +574,11 @@ async def _assert_wait_for_ec2_instances_running() -> list[InstanceTypeDef]:
         available=with_drain_nodes_labelled,
     )
     # update our fake node
-    fake_attached_node.spec.labels[_OSPARC_SERVICES_READY_DATETIME_LABEL_KEY] = (
-        mock_docker_tag_node.call_args_list[0][1]["tags"][
-            _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
-        ]
-    )
+    fake_attached_node.spec.labels[
+        _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
+    ] = mock_docker_tag_node.call_args_list[0][1]["tags"][
+        _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
+    ]
     # check the activate time is later than attach time
     assert arrow.get(
         mock_docker_tag_node.call_args_list[1][1]["tags"][
@@ -607,11 +607,11 @@ async def _assert_wait_for_ec2_instances_running() -> list[InstanceTypeDef]:
         available=True,
     )
     # update our fake node
-    fake_attached_node.spec.labels[_OSPARC_SERVICES_READY_DATETIME_LABEL_KEY] = (
-        mock_docker_tag_node.call_args_list[1][1]["tags"][
-            _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
-        ]
-    )
+    fake_attached_node.spec.labels[
+        _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
+    ] = mock_docker_tag_node.call_args_list[1][1]["tags"][
+        _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
+    ]
     mock_docker_tag_node.reset_mock()
     mock_docker_set_node_availability.assert_not_called()
 
@@ -797,9 +797,9 @@ async def _assert_wait_for_ec2_instances_running() -> list[InstanceTypeDef]:
     if not with_drain_nodes_labelled:
         fake_attached_node.spec.availability = Availability.drain
     fake_attached_node.spec.labels[_OSPARC_SERVICE_READY_LABEL_KEY] = "false"
-    fake_attached_node.spec.labels[_OSPARC_SERVICES_READY_DATETIME_LABEL_KEY] = (
-        datetime.datetime.now(tz=datetime.UTC).isoformat()
-    )
+    fake_attached_node.spec.labels[
+        _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY
+    ] = datetime.datetime.now(tz=datetime.UTC).isoformat()
 
     # the node will not be terminated before the timeout triggers
     assert app_settings.AUTOSCALING_EC2_INSTANCES
@@ -901,7 +901,7 @@ async def _assert_wait_for_ec2_instances_terminated() -> None:
         await _assert_wait_for_ec2_instances_terminated()
 
 
-@pytest.mark.acceptance_test()
+@pytest.mark.acceptance_test
 @pytest.mark.parametrize(
     "scale_up_params",
     [
@@ -1245,7 +1245,9 @@ async def _(
 
 
 def _create_fake_association(
-    create_fake_node: Callable, drained_machine_id: str | None
+    create_fake_node: Callable,
+    drained_machine_id: str | None,
+    terminating_machine_id: str | None,
 ):
     fake_node_to_instance_map = {}
 
@@ -1266,6 +1268,10 @@ def _create_fake_node_with_labels(instance: EC2InstanceData) -> Node:
                         "true" if instance.id != drained_machine_id else "false"
                     ),
                 }
+                if instance.id == terminating_machine_id:
+                    fake_node.spec.labels |= {
+                        _OSPARC_NODE_TERMINATION_PROCESS_LABEL_KEY: arrow.utcnow().isoformat()
+                    }
                 fake_node_to_instance_map[instance] = fake_node
             return fake_node_to_instance_map[instance]
 
@@ -1279,7 +1285,6 @@ def _create_fake_node_with_labels(instance: EC2InstanceData) -> Node:
     return _fake_node_creator
 
 
-@pytest.mark.testit
 @pytest.mark.parametrize(
     "with_docker_join_drained", ["with_AUTOSCALING_DOCKER_JOIN_DRAINED"], indirect=True
 )
@@ -1333,6 +1338,7 @@ async def test_cluster_adapts_machines_on_the_fly(  # noqa: PLR0915
     mock_docker_tag_node: mock.Mock,
     mock_compute_node_used_resources: mock.Mock,
     spied_cluster_analysis: MockType,
+    mocker: MockerFixture,
 ):
     # pre-requisites
     assert app_settings.AUTOSCALING_EC2_INSTANCES
@@ -1374,7 +1380,7 @@ async def test_cluster_adapts_machines_on_the_fly(  # noqa: PLR0915
     mocked_associate_ec2_instances_with_nodes.assert_called_once_with([], [])
     mocked_associate_ec2_instances_with_nodes.reset_mock()
     mocked_associate_ec2_instances_with_nodes.side_effect = _create_fake_association(
-        create_fake_node, None
+        create_fake_node, None, None
     )
 
     #
@@ -1461,7 +1467,8 @@ async def test_cluster_adapts_machines_on_the_fly(  # noqa: PLR0915
         "simcore_service_autoscaling.modules.auto_scaling_core.utils_docker.get_node_empty_since",
         autospec=True,
         return_value=arrow.utcnow().datetime
-        - 2 * app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_TIME_BEFORE_DRAINING,
+        - 1.5
+        * app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_TIME_BEFORE_DRAINING,
     ) as mocked_get_node_empty_since:
         await auto_scale_cluster(
             app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
@@ -1475,43 +1482,86 @@ async def test_cluster_adapts_machines_on_the_fly(  # noqa: PLR0915
         -1
     ].ec2_instance.id
     mocked_associate_ec2_instances_with_nodes.side_effect = _create_fake_association(
-        create_fake_node, drained_machine_instance_id
+        create_fake_node, drained_machine_instance_id, None
     )
     await auto_scale_cluster(
         app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
     )
     assert spied_cluster_analysis.spy_return.active_nodes
     assert spied_cluster_analysis.spy_return.drained_nodes
-    # mock get_node_empty_since > app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_TIME_BEFORE_DRAINING
-    # auto_scale, this will drain the node
 
-    for _ in range(3):
+    # this will initiate termination now
+    with mock.patch(
+        "simcore_service_autoscaling.modules.auto_scaling_core.utils_docker.get_node_last_readyness_update",
+        autospec=True,
+        return_value=arrow.utcnow().datetime
+        - 1.5
+        * app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_TIME_BEFORE_TERMINATION,
+    ):
+        mock_docker_tag_node.reset_mock()
         await auto_scale_cluster(
             app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
         )
+        mock_docker_tag_node.assert_called_with(
+            mock.ANY,
+            spied_cluster_analysis.spy_return.drained_nodes[-1].node,
+            tags=mock.ANY,
+            available=False,
+        )
 
-        assert spied_cluster_analysis.spy_return.active_nodes
-        assert not spied_cluster_analysis.spy_return.drained_nodes
+    # and this should now recognize the node as terminating
+    mocked_associate_ec2_instances_with_nodes.side_effect = _create_fake_association(
+        create_fake_node, drained_machine_instance_id, drained_machine_instance_id
+    )
+    await auto_scale_cluster(
+        app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
+    )
+    assert spied_cluster_analysis.spy_return.active_nodes
+    assert not spied_cluster_analysis.spy_return.drained_nodes
+    assert spied_cluster_analysis.spy_return.terminating_nodes
 
-        all_instances = await ec2_client.describe_instances()
-        assert len(all_instances["Reservations"]) == 2, "there should be 2 Reservations"
-        reservation1 = all_instances["Reservations"][0]
-        assert "Instances" in reservation1
-        assert len(reservation1["Instances"]) == (
-            app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES - 1
-        ), f"expected {app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES-1} EC2 instances, found {len(reservation1['Instances'])}"
-        for instance in reservation1["Instances"]:
-            assert "InstanceType" in instance
-            assert instance["InstanceType"] == scale_up_params1.expected_instance_type
-
-        reservation2 = all_instances["Reservations"][0]
-        assert "Instances" in reservation2
-        assert len(reservation2["Instances"]) == (
-            app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES - 1
-        ), f"expected {app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES-1} EC2 instances, found {len(reservation2['Instances'])}"
-        for instance in reservation2["Instances"]:
-            assert "InstanceType" in instance
-            assert instance["InstanceType"] == scale_up_params2.expected_instance_type
+    # now this will terminate it and straight away start a new machine type
+    with mock.patch(
+        "simcore_service_autoscaling.modules.auto_scaling_core.utils_docker.get_node_termination_started_since",
+        autospec=True,
+        return_value=arrow.utcnow().datetime
+        - 1.5
+        * app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_TIME_BEFORE_TERMINATION,
+    ):
+        mocked_docker_remove_node = mocker.patch(
+            "simcore_service_autoscaling.modules.auto_scaling_core.utils_docker.remove_nodes",
+            return_value=None,
+            autospec=True,
+        )
+        await auto_scale_cluster(
+            app=initialized_app, auto_scaling_mode=DynamicAutoscaling()
+        )
+        mocked_docker_remove_node.assert_called_once()
+
+    # now let's check what we have
+    all_instances = await ec2_client.describe_instances()
+    assert len(all_instances["Reservations"]) == 2, "there should be 2 Reservations"
+    reservation1 = all_instances["Reservations"][0]
+    assert "Instances" in reservation1
+    assert len(reservation1["Instances"]) == (
+        app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES
+    ), f"expected {app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_MAX_INSTANCES} EC2 instances, found {len(reservation1['Instances'])}"
+    for instance in reservation1["Instances"]:
+        assert "InstanceType" in instance
+        assert instance["InstanceType"] == scale_up_params1.expected_instance_type
+        if instance["InstanceId"] == drained_machine_instance_id:
+            assert instance["State"]["Name"] == "terminated"
+        else:
+            assert instance["State"]["Name"] == "running"
+
+    reservation2 = all_instances["Reservations"][1]
+    assert "Instances" in reservation2
+    assert (
+        len(reservation2["Instances"]) == 1
+    ), f"expected 1 EC2 instances, found {len(reservation2['Instances'])}"
+    for instance in reservation2["Instances"]:
+        assert "InstanceType" in instance
+        assert instance["InstanceType"] == scale_up_params2.expected_instance_type
 
 
 @pytest.mark.parametrize(