chore: more tests

juju/model.py

@@ -33,6 +33,7 @@
     TYPE_CHECKING,
     # Union,
 )
+from typing import reveal_type as reveal_type
 
 import yaml
 import websockets
@@ -62,6 +63,7 @@
 from ._sync import SyncCacheLine, ThreadedAsyncRunner
 
 if TYPE_CHECKING:
+    from .client._definitions import FullStatus
     from .application import Application
     from .machine import Machine
     from .relation import Relation
@@ -2545,7 +2547,7 @@ async def get_action_status(self, uuid_or_prefix=None, name=None):
             results[tag.untag('action-', a.action.tag)] = a.status
         return results
 
-    async def get_status(self, filters=None, utc=False):
+    async def get_status(self, filters=None, utc=False) -> FullStatus:
         """Return the status of the model.
 
         :param str filters: Optional list of applications, units, or machines
@@ -3049,7 +3051,28 @@ async def _check_idle(
         units_ready: Set[str] = set(),  # The units that are in the desired state
         last_log_time: List[Optional[datetime]] = [None],
         start_time: datetime = datetime.now(),
-    ):
+    ) -> bool:
+        now = datetime.now()
+        tmp = await self.get_status()
+        reveal_type(tmp.applications)
+        for app_name in apps:
+            app = tmp.applications.get(app_name)
+            if not app:
+                logging.info("Waiting for app %r", app_name)
+                return False
+            reveal_type(app)
+            # FIXME what if app has zero units?
+            # what if the caller is waiting for units to go down to zero?
+            reveal_type(app.units)
+            if len(app.units) < _wait_for_units:
+                logging.info("Waitinf for app %r units %s/%s",
+                             app_name, len(app.units), _wait_for_units)
+                return False
+
+            # TODO: refactor to simplify later
+            # datetime -> float; check vs outer loop
+            idle_times.setdefault(app_name, now)
+            # TODO continue here...
         return True
 
     async def _legacy_check_idle(
@@ -3069,6 +3092,7 @@ async def _legacy_check_idle(
         last_log_time: List[Optional[datetime]] = [None],
         start_time: datetime = datetime.now(),
     ):
+        #__import__("pdb").set_trace()
         _timeout = timedelta(seconds=timeout) if timeout is not None else None
         _idle_period = timedelta(seconds=idle_period)
         log_interval = timedelta(seconds=30)
@@ -3079,48 +3103,26 @@ async def _legacy_check_idle(
         errors: Dict[Literal["Machine", "Agent", "App", "Unit"], List[Any]] = {}
         blocks: Dict[Literal["Machine", "Agent", "App", "Unit"], List[Any]] = {}
 
-        # FIXME call FullStatus without filters,
-        # restrict the result to "expected apps" if that's set
         for app_name in apps:
             if app_name not in self.applications:
                 busy.append(app_name + " (missing)")
                 return False
-            # TODO
             app = self.applications[app_name]
-            # FIXME: Funny that Application.get_status()
-            # calls ClientFacade.FullStatus() without patterns
-            # (that is for all applications and units)
-            #
-            # FIXME FullStatus already contains app status
-            # app.get_status():
-            # - derive_status(
-            #   - app.status AND
-            #     - app.safe_data["status"]["current"] OR
-            #     - app.units[0..N].workload_status
-            #   - FullStatus().applications.get(app.name)
             app_status = await app.get_status()
             if raise_on_error and app_status == "error":
-                # FIXME app_name
                 errors.setdefault("App", []).append(app.name)
             if raise_on_blocked and app_status == "blocked":
-                # FIXME app_name
                 blocks.setdefault("App", []).append(app.name)
 
             # Check if wait_for_exact_units flag is used
             if wait_for_exact_units is not None:
-                # FIXME units are listed in FullStatus
                 if len(app.units) != wait_for_exact_units:
-                    # FIXME app_name
                     busy.append(app.name + " (waiting for exactly %s units, current : %s)" %
-                                # FIXME units in FullStatus
                                 (wait_for_exact_units, len(app.units)))
                     return False
             # If we have less # of units then required, then wait a bit more
-            # FIXME units in FullStatus
             elif len(app.units) < _wait_for_units:
-                # FIXME app_name
                 busy.append(app.name + " (not enough units yet - %s/%s)" %
-                            # FIXME units in FullStatus
                             (len(app.units), _wait_for_units))
                 return False
             # User is waiting for at least a certain # of units, and we have enough
@@ -3130,37 +3132,25 @@ async def _legacy_check_idle(
                 # exit the loop. Don't just return here, though, we might still have some
                 # errors to raise at the end
                 return True
-            # FIXME units in FullStatus
             for unit in app.units:
-                # FIXME ...
                 if raise_on_error and unit.machine is not None and unit.machine.status == "error":
-                    # FIXME ...
                     errors.setdefault("Machine", []).append(unit.machine.id)
                     return False
-                # FIXME ...
                 if raise_on_error and unit.agent_status == "error":
-                    # FIXME name is the units key
                     errors.setdefault("Agent", []).append(unit.name)
                     return False
-                # FIXME ...
                 if raise_on_error and unit.workload_status == "error":
-                    # FIXME name is the units key
                     errors.setdefault("Unit", []).append(unit.name)
                     return False
-                # FIXME ...
                 if raise_on_blocked and unit.workload_status == "blocked":
-                    # FIXME name is the units key
                     blocks.setdefault("Unit", []).append(unit.name)
                     return False
                 # TODO (cderici): we need two versions of wait_for_idle, one for waiting on
                 #  individual units, another one for waiting for an application.
                 #  The convoluted logic below is the result of trying to do both at the same
                 #  time
-                #
-                # FIXME ...
                 need_to_wait_more_for_a_particular_status = status and (unit.workload_status != status)
                 app_is_in_desired_status = (not status) or (app_status == status)
-                # FIXME ...
                 if not need_to_wait_more_for_a_particular_status and \
                         unit.agent_status == "idle" and \
                         (wait_for_at_least_units or app_is_in_desired_status):
@@ -3175,11 +3165,8 @@ async def _legacy_check_idle(
 
                     # Either way, the unit is ready, start measuring the time period that
                     # it needs to stay in that state (i.e. idle_period)
-                    #
-                    # FIXME key in units dict
                     units_ready.add(unit.name)
                     now = datetime.now()
-                    # FIXME key in units dict
                     idle_start = idle_times.setdefault(unit.name, now)
 
                     if now - idle_start < _idle_period:
@@ -3188,9 +3175,7 @@ async def _legacy_check_idle(
                                                             unit.workload_status,
                                                             unit.workload_status_message))
                 else:
-                    # FIXME key in dict
                     idle_times.pop(unit.name, None)
-                    # FIXME bits in FullStatus
                     busy.append("{} [{}] {}: {}".format(unit.name,
                                                         unit.agent_status,
                                                         unit.workload_status,

tests/unit/test_wait_for_idle.py

@@ -4,7 +4,8 @@
 
 import json
 import pytest
-from typing import Any, Dict, List
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Tuple, Union
 from typing import reveal_type as reveal_type
 from unittest.mock import Mock
 
@@ -15,6 +16,71 @@
 from juju.unit import Unit
 
 
+async def test_no_apps(full_status_response: dict, kwargs: Dict[str, Any]):
+    kwargs["apps"] = []
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert idle and legacy
+
+
+async def test_missing_app(full_status_response: dict, kwargs: Dict[str, Any]):
+    kwargs["apps"] = ["missing"]
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert not idle and not legacy
+
+
+async def test_no_units(full_status_response: dict, kwargs: Dict[str, Any]):
+    full_status_response["response"]["applications"]["hexanator"]["units"].clear()
+    kwargs["apps"] = ["hexanator"]
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert idle == legacy
+    assert not idle and not legacy
+
+
+async def test_idle_app(full_status_response: dict, kwargs: Dict[str, Any]):
+    kwargs["apps"] = ["hexanator"]
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert idle and legacy
+
+
+async def test_idle_period(full_status_response: dict, kwargs: Dict[str, Any]):
+    kwargs["apps"] = ["hexanator"]
+    kwargs["idle_period"] = 1
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert not idle and not legacy
+
+
+async def test_after_idle_period(full_status_response: dict, kwargs: Dict[str, Any]):
+    kwargs["apps"] = ["hexanator"]
+    kwargs["idle_period"] = 1
+    kwargs["idle_times"] = {"hexanator": datetime.now() - timedelta(seconds=2)}
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert idle and legacy
+
+
+async def test_something_useful(full_status_response: dict, kwargs: Dict[str, Any]):
+    # tweak the state
+    full_status_response["response"]["applications"]["hexanator"]["status"]["status"] = "BROKEN"
+    kwargs["wait_for_exact_units"] = 2
+
+    idle, legacy = await model_fake(full_status_response).check_both(**kwargs)
+    assert idle == legacy
+
+
+@pytest.fixture
+def kwargs() -> Dict[str, Any]:
+    return dict(
+        apps=["hexanator", "grafana-agent-k8s", "mysql-test-app"],
+        raise_on_error=False,
+        raise_on_blocked=False,
+        status=None,
+        wait_for_at_least_units=None,
+        wait_for_exact_units=None,
+        timeout=100,
+        idle_period=0,
+        _wait_for_units=1,
+    )
+
+
 @pytest.fixture
 def full_status_response(pytestconfig: pytest.Config) -> dict:
     return json.loads(((pytestconfig.rootpath / "fullstatus.json")).read_text())
@@ -65,6 +131,20 @@ class ModelFake(Model):
     _applications: Dict[str, Application]
     _response: Dict
 
+    async def check_both(self, **kwargs) -> Tuple[Union[bool, Exception], Union[bool, Exception]]:
+        try:
+            idle = await self._check_idle(**kwargs)
+        except Exception as e:
+            idle = e
+
+        try:
+            legacy = await self._legacy_check_idle(**kwargs)
+        except Exception as e:
+            raise
+            legacy = e
+
+        return idle, legacy
+
     @property
     def applications(self) -> Dict[str, Application]:
         return self._applications
@@ -128,20 +208,6 @@ def workload_status_message(self) -> str:
         return self._workload_status_message
 
 
-@pytest.fixture
-def kwargs() -> Dict[str, Any]:
-    return dict(
-        apps=["hexanator", "grafana-agent-k8s"],
-        raise_on_error=False,
-        raise_on_blocked=False,
-        status=None,
-        wait_for_at_least_units=None,
-        wait_for_exact_units=None,
-        timeout=100,
-        idle_period=100,
-        _wait_for_units=1,
-    )
-
 
 async def test_model_fake(full_status_response):
     """Validate parity between the FullStatus response and data from the library API"""
@@ -180,40 +246,3 @@ async def test_model_fake(full_status_response):
     assert not u._agent_status_message
     assert u._workload_status == "waiting"
     assert not u._workload_status_message
-
-
-async def test_something(full_status_response, kwargs: Dict[str, Any]):
-    m = model_fake(full_status_response)
-
-    try:
-        idle = await m._check_idle(**kwargs)
-    except Exception as e:
-        idle = e
-
-    try:
-        legacy = await m._legacy_check_idle(**kwargs)
-    except Exception as e:
-        raise
-        legacy = e
-
-    assert idle == legacy
-
-
-async def test_something_useful(full_status_response: dict, kwargs: Dict[str, Any]):
-    # tweak the state
-    full_status_response["response"]["applications"]["hexanator"]["status"]["status"] = "BROKEN"
-    kwargs["wait_for_exact_units"] = 2
-    m = model_fake(full_status_response)
-
-    try:
-        idle = await m._check_idle(**kwargs)
-    except Exception as e:
-        idle = e
-
-    try:
-        legacy = await m._legacy_check_idle(**kwargs)
-    except Exception as e:
-        raise
-        legacy = e
-
-    assert idle == legacy