diff --git a/juju/model/__init__.py b/juju/model/__init__.py index ef6680ff7..0063fbb6f 100644 --- a/juju/model/__init__.py +++ b/juju/model/__init__.py @@ -65,7 +65,7 @@ from ..tag import application as application_tag from ..url import URL, Schema from ..version import DEFAULT_ARCHITECTURE -from . import idle +from . import _idle if TYPE_CHECKING: from ..application import Application @@ -3301,7 +3301,7 @@ async def new_wait_for_idle( deadline = None if timeout is None else time.monotonic() + timeout async def status_on_demand(): - yield idle._check( + yield _idle.check( await self.get_status(), apps=apps, raise_on_error=raise_on_error, @@ -3309,7 +3309,7 @@ async def status_on_demand(): status=status, ) - async for done in idle._loop( + async for done in _idle.loop( status_on_demand(), apps=apps, wait_for_exact_units=wait_for_exact_units, diff --git a/juju/model/idle.py b/juju/model/_idle.py similarity index 96% rename from juju/model/idle.py rename to juju/model/_idle.py index fb1543e91..8f6b9a796 100644 --- a/juju/model/idle.py +++ b/juju/model/_idle.py @@ -21,8 +21,8 @@ @dataclass -class _CheckStatus: - """Return type for a single interaction, that is _check().""" +class CheckStatus: + """Return type check(), represents single loop iteration.""" units: set[str] """All units visible at this point.""" @@ -32,8 +32,8 @@ class _CheckStatus: """Units with stable agent status (FIXME details).""" -async def _loop( - foo: AsyncIterable[_CheckStatus | None], +async def loop( + foo: AsyncIterable[CheckStatus | None], *, apps: frozenset[str], wait_for_exact_units: int | None = None, @@ -99,15 +99,15 @@ async def _loop( yield rv -def _check( +def check( full_status: FullStatus, *, apps: frozenset[str], raise_on_error: bool, raise_on_blocked: bool, status: str | None, -) -> _CheckStatus | None: - """A single interaction of a wait_for_idle loop.""" +) -> CheckStatus | None: + """A single iteration of a wait_for_idle loop.""" for app_name in apps: if not full_status.applications.get(app_name): logger.info("Waiting for app %r", app_name) @@ -172,7 +172,7 @@ def _check( if app.status.status == "blocked" and raise_on_blocked: raise JujuAppError(f"{app_name!r} is blocked: {app.status.info!r}") - rv = _CheckStatus(set(), set(), set()) + rv = CheckStatus(set(), set(), set()) for app_name in apps: ready_units = [] diff --git a/tests/unit/test_idle_check.py b/tests/unit/test_idle_check.py index 679ac5ebe..8ed89e4f0 100644 --- a/tests/unit/test_idle_check.py +++ b/tests/unit/test_idle_check.py @@ -2,7 +2,6 @@ # Licensed under the Apache V2, see LICENCE file for details. from __future__ import annotations -# pyright: reportPrivateUsage=false import copy import json from typing import Any @@ -14,12 +13,12 @@ ) from juju.client.facade import _convert_response from juju.errors import JujuAgentError, JujuAppError, JujuMachineError, JujuUnitError -from juju.model.idle import _check, _CheckStatus +from juju.model._idle import CheckStatus, check def test_check_status(full_status: FullStatus, kwargs): - status = _check(full_status, **kwargs) - assert status == _CheckStatus( + status = check(full_status, **kwargs) + assert status == CheckStatus( { "grafana-agent-k8s/0", "hexanator/0", @@ -38,33 +37,33 @@ def test_check_status(full_status: FullStatus, kwargs): def test_check_status_missing_app(full_status: FullStatus, kwargs): kwargs["apps"] = ["missing", "hexanator"] - status = _check(full_status, **kwargs) + status = check(full_status, **kwargs) assert status is None def test_check_status_is_selective(full_status: FullStatus, kwargs): kwargs["apps"] = ["hexanator"] - status = _check(full_status, **kwargs) - assert status == _CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) + status = check(full_status, **kwargs) + assert status == CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) def test_no_apps(full_status: FullStatus, kwargs): kwargs["apps"] = [] - status = _check(full_status, **kwargs) - assert status == _CheckStatus(set(), set(), set()) + status = check(full_status, **kwargs) + assert status == CheckStatus(set(), set(), set()) def test_missing_app(full_status: FullStatus, kwargs): kwargs["apps"] = ["missing"] - status = _check(full_status, **kwargs) + status = check(full_status, **kwargs) assert status is None def test_no_units(response: dict[str, Any], kwargs): response["response"]["applications"]["hexanator"]["units"].clear() kwargs["apps"] = ["hexanator"] - status = _check(convert(response), **kwargs) - assert status == _CheckStatus(set(), set(), set()) + status = check(convert(response), **kwargs) + assert status == CheckStatus(set(), set(), set()) def test_app_error(response: dict[str, Any], kwargs): @@ -76,7 +75,7 @@ def test_app_error(response: dict[str, Any], kwargs): kwargs["raise_on_error"] = True with pytest.raises(JujuAppError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "big problem" in str(e) @@ -87,16 +86,16 @@ def test_exact_count(response: dict[str, Any], kwargs): kwargs["apps"] = ["hexanator"] - status = _check(convert(response), **kwargs) - assert status == _CheckStatus( + status = check(convert(response), **kwargs) + assert status == CheckStatus( {"hexanator/0", "hexanator/1"}, {"hexanator/0", "hexanator/1"}, set() ) def test_ready_units(full_status: FullStatus, kwargs): kwargs["apps"] = ["mysql-test-app"] - status = _check(full_status, **kwargs) - assert status == _CheckStatus( + status = check(full_status, **kwargs) + assert status == CheckStatus( {"mysql-test-app/0", "mysql-test-app/1"}, {"mysql-test-app/0", "mysql-test-app/1"}, set(), @@ -106,10 +105,8 @@ def test_ready_units(full_status: FullStatus, kwargs): def test_active_units(full_status: FullStatus, kwargs): kwargs["apps"] = ["mysql-test-app"] kwargs["status"] = "active" - status = _check(full_status, **kwargs) - assert status == _CheckStatus( - {"mysql-test-app/0", "mysql-test-app/1"}, set(), set() - ) + status = check(full_status, **kwargs) + assert status == CheckStatus({"mysql-test-app/0", "mysql-test-app/1"}, set(), set()) def test_ready_unit_requires_idle_agent(response: dict[str, Any], kwargs): @@ -120,7 +117,7 @@ def test_ready_unit_requires_idle_agent(response: dict[str, Any], kwargs): kwargs["apps"] = ["hexanator"] kwargs["status"] = "active" - status = _check(convert(response), **kwargs) + status = check(convert(response), **kwargs) assert status assert status.units == {"hexanator/0", "hexanator/1"} assert status.ready_units == {"hexanator/0"} @@ -134,10 +131,8 @@ def test_ready_unit_requires_workload_status(response: dict[str, Any], kwargs): kwargs["apps"] = ["hexanator"] kwargs["status"] = "active" - status = _check(convert(response), **kwargs) - assert status == _CheckStatus( - {"hexanator/0", "hexanator/1"}, {"hexanator/0"}, set() - ) + status = check(convert(response), **kwargs) + assert status == CheckStatus({"hexanator/0", "hexanator/1"}, {"hexanator/0"}, set()) def test_agent_error(response: dict[str, Any], kwargs): @@ -149,7 +144,7 @@ def test_agent_error(response: dict[str, Any], kwargs): kwargs["raise_on_error"] = True with pytest.raises(JujuAgentError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "hexanator/0" in str(e) assert "agent problem" in str(e) @@ -164,7 +159,7 @@ def test_workload_error(response: dict[str, Any], kwargs): kwargs["raise_on_error"] = True with pytest.raises(JujuUnitError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "hexanator/0" in str(e) assert "workload problem" in str(e) @@ -186,8 +181,8 @@ def test_machine_ok(response: dict[str, Any], kwargs): kwargs["apps"] = ["hexanator"] kwargs["raise_on_error"] = True - status = _check(convert(response), **kwargs) - assert status == _CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) + status = check(convert(response), **kwargs) + assert status == CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) def test_machine_error(response: dict[str, Any], kwargs): @@ -206,7 +201,7 @@ def test_machine_error(response: dict[str, Any], kwargs): kwargs["raise_on_error"] = True with pytest.raises(JujuMachineError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "potato" in str(e) @@ -220,7 +215,7 @@ def test_app_blocked(response: dict[str, Any], kwargs): kwargs["raise_on_blocked"] = True with pytest.raises(JujuAppError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "big problem" in str(e) @@ -234,7 +229,7 @@ def test_unit_blocked(response: dict[str, Any], kwargs): kwargs["raise_on_blocked"] = True with pytest.raises(JujuUnitError) as e: - _check(convert(response), **kwargs) + check(convert(response), **kwargs) assert "small problem" in str(e) @@ -248,8 +243,8 @@ def test_maintenance(response: dict[str, Any], kwargs): kwargs["apps"] = ["hexanator"] kwargs["status"] = "maintenance" - status = _check(convert(response), **kwargs) - assert status == _CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) + status = check(convert(response), **kwargs) + assert status == CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) @pytest.fixture diff --git a/tests/unit/test_idle_check_subordinate.py b/tests/unit/test_idle_check_subordinate.py index 97475341f..149e468ba 100644 --- a/tests/unit/test_idle_check_subordinate.py +++ b/tests/unit/test_idle_check_subordinate.py @@ -2,7 +2,6 @@ # Licensed under the Apache V2, see LICENCE file for details. from __future__ import annotations -# pyright: reportPrivateUsage=false import json from typing import Any @@ -12,12 +11,12 @@ FullStatus, ) from juju.client.facade import _convert_response -from juju.model.idle import _check, _CheckStatus +from juju.model._idle import CheckStatus, check def test_subordinate_apps(response: dict[str, Any], kwargs): - status = _check(convert(response), **kwargs) - assert status == _CheckStatus({"ntp/0", "ubuntu/0"}, {"ntp/0", "ubuntu/0"}, set()) + status = check(convert(response), **kwargs) + assert status == CheckStatus({"ntp/0", "ubuntu/0"}, {"ntp/0", "ubuntu/0"}, set()) def test_subordinate_is_selective(response, kwargs): @@ -25,8 +24,8 @@ def test_subordinate_is_selective(response, kwargs): "subordinates" ] subordinates["some-other/0"] = subordinates["ntp/0"] - status = _check(convert(response), **kwargs) - assert status == _CheckStatus({"ntp/0", "ubuntu/0"}, {"ntp/0", "ubuntu/0"}, set()) + status = check(convert(response), **kwargs) + assert status == CheckStatus({"ntp/0", "ubuntu/0"}, {"ntp/0", "ubuntu/0"}, set()) @pytest.fixture diff --git a/tests/unit/test_idle_loop.py b/tests/unit/test_idle_loop.py index d87e8c72d..34a8c6721 100644 --- a/tests/unit/test_idle_loop.py +++ b/tests/unit/test_idle_loop.py @@ -3,11 +3,9 @@ from __future__ import annotations import pytest - -# pyright: reportPrivateUsage=false from freezegun import freeze_time -from juju.model.idle import _CheckStatus, _loop +from juju.model._idle import CheckStatus, loop # Missing tests @@ -24,14 +22,14 @@ @pytest.mark.xfail(reason="FIXME I misunderstood what 'idle' means") async def test_at_least_units(): async def checks(): - yield _CheckStatus({"u/0", "u/1", "u/2"}, {"u/0"}, set()) - yield _CheckStatus({"u/0", "u/1", "u/2"}, {"u/0", "u/1"}, set()) - yield _CheckStatus({"u/0", "u/1", "u/2"}, {"u/0", "u/1", "u/2"}, set()) + yield CheckStatus({"u/0", "u/1", "u/2"}, {"u/0"}, set()) + yield CheckStatus({"u/0", "u/1", "u/2"}, {"u/0", "u/1"}, set()) + yield CheckStatus({"u/0", "u/1", "u/2"}, {"u/0", "u/1", "u/2"}, set()) with freeze_time(): assert [ v - async for v in _loop( + async for v in loop( checks(), apps=frozenset(["u"]), wait_for_units=2, @@ -41,8 +39,8 @@ async def checks(): async def test_ping_pong(): - good = _CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) - bad = _CheckStatus({"hexanator/0"}, set(), set()) + good = CheckStatus({"hexanator/0"}, {"hexanator/0"}, set()) + bad = CheckStatus({"hexanator/0"}, set(), set()) async def checks(): with freeze_time() as clock: @@ -54,7 +52,7 @@ async def checks(): assert [ v - async for v in _loop( + async for v in loop( checks(), apps=frozenset(["hexanator"]), wait_for_units=1,