From 3321e089ac44ddd7b3b5f2a036db192756219601 Mon Sep 17 00:00:00 2001 From: Dima Tisnek Date: Thu, 3 Oct 2024 17:51:09 +0900 Subject: [PATCH] chore: refactor test setup --- tests/unit/test_wait_for_idle.py | 136 +++++++++++++++++++++---------- 1 file changed, 92 insertions(+), 44 deletions(-) diff --git a/tests/unit/test_wait_for_idle.py b/tests/unit/test_wait_for_idle.py index a43c722c9..1b31c1b36 100644 --- a/tests/unit/test_wait_for_idle.py +++ b/tests/unit/test_wait_for_idle.py @@ -20,29 +20,46 @@ def full_status_response(pytestconfig: pytest.Config) -> dict: return json.loads(((pytestconfig.rootpath / "fullstatus.json")).read_text()) -def test_foo(full_status_response: dict): - # tweak the response here - fs = _convert_response(full_status_response, cls=FullStatus) +def model_fake(resp: dict) -> ModelFake: + m = ModelFake() + m._response = resp + fs = _convert_response(resp, cls=FullStatus) assert fs.applications - assert fs.applications["hexanator"] - hexanator = fs.applications["hexanator"] - assert hexanator.status # DetailedStatus - assert hexanator.status.status == "active" - assert hexanator.status.info == "" + for name in fs.applications: + app = m._applications[name] = ApplicationFake(name, m) + + fsapp = fs.applications[name] + assert fsapp + assert fsapp.status # DetailedStatus + assert isinstance(fsapp.status.status, str) + app._status = fsapp.status.status + + assert isinstance(fsapp.status.info, str) + app._status_message = fsapp.status.info + + for uname in fsapp.units: + app._units.append(unit := UnitFake(uname, m)) - assert fs.applications["grafana-agent-k8s"] - grafana = fs.applications["grafana-agent-k8s"] # ApplicationStatus + fsunit = fsapp.units[uname] + assert fsunit - assert grafana.units["grafana-agent-k8s/0"] - grafana0 = grafana.units["grafana-agent-k8s/0"] + assert fsunit.agent_status # DetailedStatus + assert isinstance(fsunit.agent_status.status, str) + unit._agent_status = fsunit.agent_status.status - assert grafana0.workload_status # DetailedStatus - assert isinstance(grafana0.workload_status.info, str) + assert isinstance(fsunit.agent_status.info, str) + unit._agent_status_message = fsunit.agent_status.info - assert grafana0.workload_status.info.startswith("Missing") + assert fsunit.workload_status # DetailedStatus + assert isinstance(fsunit.workload_status.status, str) + unit._workload_status = fsunit.workload_status.status + assert isinstance(fsunit.workload_status.info, str) + unit._workload_status_message = fsunit.workload_status.info + + return m class ModelFake(Model): _applications: Dict[str, Application] @@ -126,36 +143,67 @@ def kwargs() -> Dict[str, Any]: ) +async def test_model_fake(full_status_response): + """Validate parity between the FullStatus response and data from the library API""" + m = model_fake(full_status_response) + + app = m._applications["hexanator"] + assert len(app._units) == 1 + + u = app._units[0] + assert u._agent_status == "idle" + assert not u._agent_status_message + assert u._workload_status == "active" + assert not u._workload_status_message + + app = m._applications["grafana-agent-k8s"] + assert len(app._units) == 1 + + u = app._units[0] + assert u._agent_status == "idle" + assert not u._agent_status_message + assert u._workload_status == "blocked" + assert u._workload_status_message.startswith("Missing incoming") + + + app = m._applications["mysql-test-app"] + assert len(app._units) == 2 + + u = [u for u in app._units if u.name.endswith("/0")][0] + assert u._agent_status == "idle" + assert not u._agent_status_message + assert u._workload_status == "waiting" + assert not u._workload_status_message + + u = [u for u in app._units if u.name.endswith("/1")][0] + assert u._agent_status == "idle" + 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 = ModelFake() - m._response = full_status_response - - # FIXME fill these in from response JSON, assert equivalence - app = m._applications["hexanator"] = ApplicationFake("hexanator", m) - app._units.append(u := UnitFake("hexanator/0", m)) - u._agent_status = "FIXME" - u._agent_status_message = "FIXME" - u._workload_status = "FIXME" - u._workload_status_message = "FIXME" - - app = m._applications["grafana-agent-k8s"] = ApplicationFake("grafana-agent-k8s", m) - app._units.append(u := UnitFake("grafana-agent-k8s/0", m)) - u._agent_status = "FIXME" - u._agent_status_message = "FIXME" - u._workload_status = "FIXME" - u._workload_status_message = "FIXME" - - app = m._applications["mysql-test-app"] = ApplicationFake("mysql-test-app", m) - app._units.append(u := UnitFake("mysql-test-app/0", m)) - u._agent_status = "FIXME" - u._agent_status_message = "FIXME" - u._workload_status = "FIXME" - u._workload_status_message = "FIXME" - app._units.append(u := UnitFake("mysql-test-app/1", m)) - u._agent_status = "FIXME" - u._agent_status_message = "FIXME" - u._workload_status = "FIXME" - u._workload_status_message = "FIXME" + 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)