From 832cf5746d2fec40b29f9962a14a9b88fde3b77a Mon Sep 17 00:00:00 2001 From: willGraham01 <1willgraham@gmail.com> Date: Mon, 22 Apr 2024 15:10:22 +0100 Subject: [PATCH 1/4] some sim sharing framework for stunting tests --- tests/conftest.py | 23 +++++-- tests/test_stunting.py | 138 ++++++++++++++++++++++++++++++++--------- 2 files changed, 125 insertions(+), 36 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 47d6c3fa16..35a5ab4763 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,11 @@ """Collection of shared fixtures""" +import os +from pathlib import Path + import pytest +from tlo import Date + DEFAULT_SEED = 83563095832589325021 @@ -10,13 +15,10 @@ def pytest_addoption(parser): type=int, nargs="*", default=[DEFAULT_SEED], - help="Seed(s) for simulation-level random number generator in tests" + help="Seed(s) for simulation-level random number generator in tests", ) parser.addoption( - "--skip-slow", - action="store_true", - default=False, - help="Skip slow tests" + "--skip-slow", action="store_true", default=False, help="Skip slow tests" ) @@ -34,4 +36,13 @@ def pytest_collection_modifyitems(config, items): def pytest_generate_tests(metafunc): if "seed" in metafunc.fixturenames: - metafunc.parametrize("seed", metafunc.config.getoption("seed")) + metafunc.parametrize("seed", metafunc.config.getoption("seed"), scope="session") + + +@pytest.fixture(scope="session") +def jan_1st_2010() -> Date: + return Date(2010, 1, 1) + +@pytest.fixture(scope="session") +def resource_filepath() -> Path: + return (Path(os.path.dirname(__file__)) / "../resources").resolve() diff --git a/tests/test_stunting.py b/tests/test_stunting.py index 4b11db1512..2b9713310f 100644 --- a/tests/test_stunting.py +++ b/tests/test_stunting.py @@ -221,59 +221,137 @@ def test_polling_event_progression(seed): assert (df.loc[df.is_alive & (df.age_years >= 5), 'un_HAZ_category'] == 'HAZ>=-2').all() -def test_routine_assessment_for_chronic_undernutrition_if_stunted_and_correctly_diagnosed(seed): - """Check that a call to `do_at_generic_first_appt` can lead to immediate recovery for a - stunted child (via an HSI), if there is checking and correct diagnosis.""" - popsize = 100 - sim = get_sim(seed) - sim.make_initial_population(n=popsize) +@pytest.fixture(scope="module") +def shared_stunting_sim(seed, jan_1st_2010, resource_filepath): + """""" + sim = Simulation(start_date=jan_1st_2010, seed=seed) + sim.register( + demography.Demography(resourcefilepath=resource_filepath), + enhanced_lifestyle.Lifestyle(resourcefilepath=resource_filepath), + healthsystem.HealthSystem( + resourcefilepath=resource_filepath, cons_availability="all" + ), + simplified_births.SimplifiedBirths(resourcefilepath=resource_filepath), + stunting.Stunting(resourcefilepath=resource_filepath), + stunting.StuntingPropertiesOfOtherModules(), + ) + sim.make_initial_population(n=100) sim.simulate(end_date=sim.start_date) - sim.modules['HealthSystem'].reset_queue() + return sim - # Make one person have non-severe stunting - df = sim.population.props - person_id = 0 - df.loc[person_id, 'age_years'] = 2 - df.loc[person_id, "un_HAZ_category"] = "-3<=HAZ<-2" - patient_details = sim.population.row_in_readonly_form(person_id) +@pytest.fixture(scope="module") +def patient_id() -> int: + return 0 + +@pytest.fixture(scope="function") +def reset_hsi_queue(shared_stunting_sim): + """ + Reset the HSI event queue before running this test. + """ + shared_stunting_sim.modules["HealthSystem"].reset_queue() + yield + +@pytest.fixture(scope="function") +def edit_sim_properties(shared_stunting_sim, param_overwrites): + """ + Temporarily edit the Stunting module parameters when running + this test. + + Property overwrites should be a dictionary whose key: value pairs + are the names of parameters in the stunting module, and the value + to overwrite that parameter with for the duration of the test. + """ + old_param_values = {} + params = shared_stunting_sim.modules["Stunting"].parameters + + # Edit properties for this test only + for param, old_value in param_overwrites.items(): + old_param_values[param] = old_value + params[param] = param_overwrites[param] + + yield + + # Reset properties to defaults + for param, old_value in old_param_values.items(): + params[param] = old_value + + +@pytest.fixture(scope="function") +def edit_patient_properties(shared_stunting_sim, patient_id, patient_overwrites): + """ + Edit the details of the patient with the given ID to match those given + in property_overwrites, reverting this change at the end of the test. + + Property overwrites should be a dictionary whose key: value pairs + are the names of parameters in the stunting module, and the value + to overwrite that parameter with for the duration of the test. + """ + old_patient_values = shared_stunting_sim.population.row_in_readonly_form(patient_id) + values_to_restore = {col: old_patient_values[col] for col in patient_overwrites.keys()} + + # Edit patient properties + shared_stunting_sim.population.props.loc[patient_id, patient_overwrites.keys()] = patient_overwrites.values() + + yield + + # Reset the patient to their pre-test state + shared_stunting_sim.population.props.loc[patient_id, values_to_restore.keys()] = ( + values_to_restore.values() + ) - # Make the probability of stunting checking/diagnosis as 1.0 - sim.modules['Stunting'].parameters['prob_stunting_diagnosed_at_generic_appt'] = 1.0 + +non_severe_stunting = {"age_years": 2, "un_HAZ_category": "-3<=HAZ<-2"} +non_severe_stunting_params = { + "prob_stunting_diagnosed_at_generic_appt": 1.0, + "effectiveness_of_complementary_feeding_education_in_stunting_reduction": 1.0, + "effectiveness_of_food_supplementation_in_stunting_reduction": 1.0, +} + + +@pytest.mark.parametrize( + "patient_overwrites, param_overwrites", + [ + pytest.param( + non_severe_stunting, non_severe_stunting_params, id="Non-severe stunting" + ) + ], +) +def test_routine_assessment_for_chronic_undernutrition_if_stunted_and_correctly_diagnosed( + shared_stunting_sim, patient_id, edit_patient_properties, edit_sim_properties, reset_hsi_queue +): + """Check that a call to `do_at_generic_first_appt` can lead to immediate recovery for a + stunted child (via an HSI), if there is checking and correct diagnosis.""" + patient_details = shared_stunting_sim.population.row_in_readonly_form(patient_id) + df = shared_stunting_sim.population.props # Subject the person to `do_at_generic_first_appt` - sim.modules["Stunting"].do_at_generic_first_appt( - patient_id=person_id, patient_details=patient_details + shared_stunting_sim.modules["Stunting"].do_at_generic_first_appt( + patient_id=patient_id, patient_details=patient_details ) # Check that there is an HSI scheduled for this person hsi_event_scheduled = [ ev - for ev in sim.modules["HealthSystem"].find_events_for_person(person_id) + for ev in shared_stunting_sim.modules["HealthSystem"].find_events_for_person(patient_id) if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) ] assert 1 == len(hsi_event_scheduled) - assert sim.date == hsi_event_scheduled[0][0] + assert shared_stunting_sim.date == hsi_event_scheduled[0][0] the_hsi_event = hsi_event_scheduled[0][1] - assert person_id == the_hsi_event.target - - # Make probability of treatment success is 1.0 (consumables are available through use of `ignore_cons_constraints`) - sim.modules['Stunting'].parameters[ - 'effectiveness_of_complementary_feeding_education_in_stunting_reduction'] = 1.0 - sim.modules['Stunting'].parameters[ - 'effectiveness_of_food_supplementation_in_stunting_reduction'] = 1.0 + assert patient_id == the_hsi_event.target # Run the HSI event the_hsi_event.run(squeeze_factor=0.0) # Check that the person is not longer stunted - assert df.at[person_id, 'un_HAZ_category'] == 'HAZ>=-2' + assert df.at[patient_id, 'un_HAZ_category'] == 'HAZ>=-2' # Check that there is a follow-up appointment scheduled hsi_event_scheduled_after_first_appt = [ - ev for ev in sim.modules['HealthSystem'].find_events_for_person(person_id) + ev for ev in shared_stunting_sim.modules['HealthSystem'].find_events_for_person(patient_id) if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) ] assert 2 == len(hsi_event_scheduled_after_first_appt) - assert (sim.date + pd.DateOffset(months=6)) == hsi_event_scheduled_after_first_appt[ + assert (shared_stunting_sim.date + pd.DateOffset(months=6)) == hsi_event_scheduled_after_first_appt[ 1 ][0] the_follow_up_hsi_event = hsi_event_scheduled_after_first_appt[1][1] @@ -283,7 +361,7 @@ def test_routine_assessment_for_chronic_undernutrition_if_stunted_and_correctly_ # Check that after running the following appointments there are no further appointments scheduled assert hsi_event_scheduled_after_first_appt == [ - ev for ev in sim.modules['HealthSystem'].find_events_for_person(person_id) + ev for ev in shared_stunting_sim.modules['HealthSystem'].find_events_for_person(patient_id) if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) ] From 8750cc4bb4496cd29f361dcd21f3943a23fdbbef Mon Sep 17 00:00:00 2001 From: willGraham01 <1willgraham@gmail.com> Date: Wed, 24 Apr 2024 10:35:31 +0100 Subject: [PATCH 2/4] Create actual infrastructure for sharing simulations --- tests/conftest.py | 165 +++++++++++++++++++- tests/test_stunting.py | 344 ++++++++++++++++------------------------- 2 files changed, 297 insertions(+), 212 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 35a5ab4763..f3fbbc1fd4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,14 +1,22 @@ """Collection of shared fixtures""" +from __future__ import annotations import os from pathlib import Path +from typing import TYPE_CHECKING, List import pytest -from tlo import Date +from tlo import Date, Module, Simulation +from tlo.methods import ( + demography, + enhanced_lifestyle, + healthsystem, + simplified_births, + stunting, +) DEFAULT_SEED = 83563095832589325021 - def pytest_addoption(parser): parser.addoption( "--seed", @@ -38,11 +46,164 @@ def pytest_generate_tests(metafunc): if "seed" in metafunc.fixturenames: metafunc.parametrize("seed", metafunc.config.getoption("seed"), scope="session") +## """Fixtures and classes that are to be utilised for sharing simulations across the test framework.""" + +if TYPE_CHECKING: + from pandas import DataFrame @pytest.fixture(scope="session") def jan_1st_2010() -> Date: return Date(2010, 1, 1) + @pytest.fixture(scope="session") def resource_filepath() -> Path: return (Path(os.path.dirname(__file__)) / "../resources").resolve() + + +@pytest.fixture(scope="session") +def small_shared_sim(seed, jan_1st_2010, resource_filepath): + """ + Shared simulation object that can be re-used across multiple tests. + + Note that to ensure the object can be shared between tests, it is + necessary that: + - All modules that are to make use of the shared simulation, + and their dependencies, are registered with the simulation. + - The initial population cannot be changed and must be sufficient + for all tests in which the object is to be used. + - The simulation cannot be further simulated into the future. + """ + sim = Simulation(start_date=jan_1st_2010, seed=seed) + sim.register( + demography.Demography(resourcefilepath=resource_filepath), + enhanced_lifestyle.Lifestyle(resourcefilepath=resource_filepath), + healthsystem.HealthSystem( + resourcefilepath=resource_filepath, cons_availability="all" + ), + simplified_births.SimplifiedBirths(resourcefilepath=resource_filepath), + stunting.Stunting(resourcefilepath=resource_filepath), + stunting.StuntingPropertiesOfOtherModules(), + ) + sim.make_initial_population(n=100) + sim.simulate(end_date=sim.start_date) + return sim + + +class _BaseSharedSim: + """ + Base class for creating tests that utilise a shared simulation. + Module-level tests that want to utilise the shared simulation + should inherit from this class, and set the "module" attribute + appropriately. + + This base class also defines a number of safe "setup/teardown" + fixtures to ensure that the state of the shared simulation is not + inadvertently altered between tests, creating a knock-on effect. + If a test needs to alter the state of the simulation; for example + - Clearing the HSI event queue / event queue + - Changing module parameters + - Changing patient details + then use the fixtures provided to ensure that the original state + of these objects is restored after the test runs. Then during the + test, you are free to edit these properties in the shared + simulation. + + This class also provides several properties for quick access to + properties of the shared simulation. + """ + + # Set this class-wide variable to be the name + # of the module that the class will be testing. + module: str + # This is how to access the shared simulation resource + # that tests will automatically hook into. + sim: Simulation + + @property + def this_module(self) -> Module: + """ + Points to the disease module being tested by this class, + within the shared simulation. + """ + return self.sim.modules[self.module] + + @property + def shared_sim_df(self) -> DataFrame: + """ + Points to the population DataFrame used by the + shared simulation. + + WARNING: Writes are persistent! + Use the setup fixture if you intend to make changes to + the shared DataFrame. + """ + return self.sim.population.props + + @property + def shared_sim_healthsystem(self) -> healthsystem.HealthSystem: + """ + Points to the HealthSystem module if in use by the shared + simulation. + """ + return self.sim.modules["HealthSystem"] + + @pytest.fixture(autouse=True, scope="function") + def _attach_to_shared_sim(self, small_shared_sim): + """ + Before each test in this suite, provide access to the shared + simulation fixture defined above. This ensures that every test + is run with the persistent simulation object in its context. + + WARNING: Writes to the shared simulation object will thus be + persistent between tests! If a test needs to modify module + parameters, use a fresh HSI queue, or similar, use the setup + fixtures also provided with the base class. + """ + self.sim = small_shared_sim + + @pytest.fixture(scope="function") + def clears_hsi_queue(self): + """ + Flags this test as needing to clear the HSI event queue + in the shared simulation. + + Using this fixture will cause pytest to cache the current + HSI queue of the shared simulation, clear the queue, run + the test, then restore the old queue during test teardown. + """ + cached_queue = list(self.shared_sim_healthsystem.HSI_EVENT_QUEUE) + self.shared_sim_healthsystem.reset_queue() + yield + self.shared_sim_healthsystem.HSI_EVENT_QUEUE = list(cached_queue) + + @pytest.fixture(scope="function") + def changes_module_properties(self): + """ + Flags this test as needing to change the properties of the + module being tested in the shared simulation. + + Using this fixture will cause pytest to cache the current + parameters used for the module in question prior to the test + commencing. Upon test teardown, module parameter values will + be restored to their pre-test state. + """ + old_param_values = dict(self.this_module.parameters) + yield + self.this_module.parameters = dict(old_param_values) + + @pytest.fixture(scope="function") + def changes_patient_properties(self, patient_id: int | List[int]): + """ + Flags this test as needing to manually change the properties + of the patient at the given index(es) in the shared simulation + DataFrame. + + Using this fixture will cause pytest to cache the corresponding + rows in the population DataFrame prior to the test commencing. + Upon test teardown, these DataFrame rows will be restored to + their pre-test state. + """ + cached_values = self.shared_sim_df.loc[patient_id].copy() + yield + self.shared_sim_df.loc[patient_id] = cached_values diff --git a/tests/test_stunting.py b/tests/test_stunting.py index 2b9713310f..da217747b3 100644 --- a/tests/test_stunting.py +++ b/tests/test_stunting.py @@ -14,6 +14,7 @@ from tlo.methods.stunting import HSI_Stunting_ComplementaryFeeding from tlo.util import random_date +from .conftest import _BaseSharedSim def get_sim(seed): """Return simulation objection with Stunting and other necessary modules registered.""" @@ -221,221 +222,144 @@ def test_polling_event_progression(seed): assert (df.loc[df.is_alive & (df.age_years >= 5), 'un_HAZ_category'] == 'HAZ>=-2').all() -@pytest.fixture(scope="module") -def shared_stunting_sim(seed, jan_1st_2010, resource_filepath): - """""" - sim = Simulation(start_date=jan_1st_2010, seed=seed) - sim.register( - demography.Demography(resourcefilepath=resource_filepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resource_filepath), - healthsystem.HealthSystem( - resourcefilepath=resource_filepath, cons_availability="all" - ), - simplified_births.SimplifiedBirths(resourcefilepath=resource_filepath), - stunting.Stunting(resourcefilepath=resource_filepath), - stunting.StuntingPropertiesOfOtherModules(), - ) - sim.make_initial_population(n=100) - sim.simulate(end_date=sim.start_date) - return sim - -@pytest.fixture(scope="module") -def patient_id() -> int: - return 0 - -@pytest.fixture(scope="function") -def reset_hsi_queue(shared_stunting_sim): - """ - Reset the HSI event queue before running this test. - """ - shared_stunting_sim.modules["HealthSystem"].reset_queue() - yield - -@pytest.fixture(scope="function") -def edit_sim_properties(shared_stunting_sim, param_overwrites): - """ - Temporarily edit the Stunting module parameters when running - this test. - - Property overwrites should be a dictionary whose key: value pairs - are the names of parameters in the stunting module, and the value - to overwrite that parameter with for the duration of the test. +class TestStunting_SharedSim(_BaseSharedSim): """ - old_param_values = {} - params = shared_stunting_sim.modules["Stunting"].parameters - - # Edit properties for this test only - for param, old_value in param_overwrites.items(): - old_param_values[param] = old_value - params[param] = param_overwrites[param] - - yield - - # Reset properties to defaults - for param, old_value in old_param_values.items(): - params[param] = old_value - - -@pytest.fixture(scope="function") -def edit_patient_properties(shared_stunting_sim, patient_id, patient_overwrites): + Tests for the Stunting module that make use of the shared simulation. """ - Edit the details of the patient with the given ID to match those given - in property_overwrites, reverting this change at the end of the test. - - Property overwrites should be a dictionary whose key: value pairs - are the names of parameters in the stunting module, and the value - to overwrite that parameter with for the duration of the test. - """ - old_patient_values = shared_stunting_sim.population.row_in_readonly_form(patient_id) - values_to_restore = {col: old_patient_values[col] for col in patient_overwrites.keys()} - - # Edit patient properties - shared_stunting_sim.population.props.loc[patient_id, patient_overwrites.keys()] = patient_overwrites.values() - - yield - # Reset the patient to their pre-test state - shared_stunting_sim.population.props.loc[patient_id, values_to_restore.keys()] = ( - values_to_restore.values() - ) - - -non_severe_stunting = {"age_years": 2, "un_HAZ_category": "-3<=HAZ<-2"} -non_severe_stunting_params = { - "prob_stunting_diagnosed_at_generic_appt": 1.0, - "effectiveness_of_complementary_feeding_education_in_stunting_reduction": 1.0, - "effectiveness_of_food_supplementation_in_stunting_reduction": 1.0, -} - - -@pytest.mark.parametrize( - "patient_overwrites, param_overwrites", - [ - pytest.param( - non_severe_stunting, non_severe_stunting_params, id="Non-severe stunting" + module: str = "Stunting" + + @pytest.fixture(scope="module") + def patient_id(self) -> int: + return 0 + + def test_routine_assessment_for_chronic_undernutrition_if_stunted_and_correctly_diagnosed( + self, + patient_id, + changes_patient_properties, + changes_module_properties, + clears_hsi_queue, + ): + """Check that a call to `do_at_generic_first_appt` can lead to immediate recovery for a + stunted child (via an HSI), if there is checking and correct diagnosis.""" + self.shared_sim_df.loc[patient_id, "age_years"] = 2 + self.shared_sim_df.loc[patient_id, "un_HAZ_category"] = "-3<=HAZ<-2" + + self.this_module.parameters["prob_stunting_diagnosed_at_generic_appt"] = 1.0 + self.this_module.parameters[ + "effectiveness_of_complementary_feeding_education_in_stunting_reduction" + ] = 1.0 + self.this_module.parameters[ + "effectiveness_of_food_supplementation_in_stunting_reduction" + ] = 1.0 + + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + # Subject the person to `do_at_generic_first_appt` + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, patient_details=patient_details ) - ], -) -def test_routine_assessment_for_chronic_undernutrition_if_stunted_and_correctly_diagnosed( - shared_stunting_sim, patient_id, edit_patient_properties, edit_sim_properties, reset_hsi_queue -): - """Check that a call to `do_at_generic_first_appt` can lead to immediate recovery for a - stunted child (via an HSI), if there is checking and correct diagnosis.""" - patient_details = shared_stunting_sim.population.row_in_readonly_form(patient_id) - df = shared_stunting_sim.population.props - - # Subject the person to `do_at_generic_first_appt` - shared_stunting_sim.modules["Stunting"].do_at_generic_first_appt( - patient_id=patient_id, patient_details=patient_details - ) - - # Check that there is an HSI scheduled for this person - hsi_event_scheduled = [ - ev - for ev in shared_stunting_sim.modules["HealthSystem"].find_events_for_person(patient_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - assert 1 == len(hsi_event_scheduled) - assert shared_stunting_sim.date == hsi_event_scheduled[0][0] - the_hsi_event = hsi_event_scheduled[0][1] - assert patient_id == the_hsi_event.target - - # Run the HSI event - the_hsi_event.run(squeeze_factor=0.0) - - # Check that the person is not longer stunted - assert df.at[patient_id, 'un_HAZ_category'] == 'HAZ>=-2' - # Check that there is a follow-up appointment scheduled - hsi_event_scheduled_after_first_appt = [ - ev for ev in shared_stunting_sim.modules['HealthSystem'].find_events_for_person(patient_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - assert 2 == len(hsi_event_scheduled_after_first_appt) - assert (shared_stunting_sim.date + pd.DateOffset(months=6)) == hsi_event_scheduled_after_first_appt[ - 1 - ][0] - the_follow_up_hsi_event = hsi_event_scheduled_after_first_appt[1][1] - - # Run the Follow-up HSI event - the_follow_up_hsi_event.run(squeeze_factor=0.0) - - # Check that after running the following appointments there are no further appointments scheduled - assert hsi_event_scheduled_after_first_appt == [ - ev for ev in shared_stunting_sim.modules['HealthSystem'].find_events_for_person(patient_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - - -def test_routine_assessment_for_chronic_undernutrition_if_stunted_but_no_checking(seed): - """Check that a call to `do_at_generic_first_appt` does not lead to an HSI for a stunted - child, if there is no checking/diagnosis.""" - popsize = 100 - sim = get_sim(seed) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=sim.start_date) - sim.modules['HealthSystem'].reset_queue() - # Make one person have severe stunting - df = sim.population.props - person_id = 0 - df.loc[person_id, 'age_years'] = 2 - df.loc[person_id, "un_HAZ_category"] = "HAZ<-3" - patient_details = sim.population.row_in_readonly_form(person_id) - - # Make the probability of stunting checking/diagnosis as 0.0 - sim.modules['Stunting'].parameters['prob_stunting_diagnosed_at_generic_appt'] = 0.0 - - # Subject the person to `do_at_generic_first_appt` - sim.modules["Stunting"].do_at_generic_first_appt(patient_id=person_id, patient_details=patient_details) - - # Check that there is no HSI scheduled for this person - hsi_event_scheduled = [ - ev[1] - for ev in sim.modules["HealthSystem"].find_events_for_person(person_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - assert 0 == len(hsi_event_scheduled) - - # Then make the probability of stunting checking/diagnosis 1.0 - # and check the HSI is scheduled for this person - sim.modules["Stunting"].parameters["prob_stunting_diagnosed_at_generic_appt"] = 1.0 - sim.modules["Stunting"].do_at_generic_first_appt( - patient_id=person_id, patient_details=patient_details - ) - hsi_event_scheduled = [ - ev[1] - for ev in sim.modules["HealthSystem"].find_events_for_person(person_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - assert 1 == len(hsi_event_scheduled) - - -def test_routine_assessment_for_chronic_undernutrition_if_not_stunted(seed): - """Check that a call to `do_at_generic_first_appt` does not lead to an HSI if there is no - stunting.""" - popsize = 100 - sim = get_sim(seed) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=sim.start_date) - sim.modules['HealthSystem'].reset_queue() - - # Make one person have no stunting - df = sim.population.props - person_id = 0 - df.loc[person_id, 'age_years'] = 2 - df.loc[person_id, 'un_HAZ_category'] = 'HAZ>=-2' - patient_details = sim.population.row_in_readonly_form(person_id) - - # Subject the person to `do_at_generic_first_appt` - sim.modules["Stunting"].do_at_generic_first_appt(patient_id=person_id, patient_details=patient_details) - - # Check that there is no HSI scheduled for this person - hsi_event_scheduled = [ - ev[1] - for ev in sim.modules["HealthSystem"].find_events_for_person(person_id) - if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) - ] - assert 0 == len(hsi_event_scheduled) + # Check that there is an HSI scheduled for this person + hsi_event_scheduled = [ + ev + for ev in self.shared_sim_healthsystem.find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + assert 1 == len(hsi_event_scheduled) + assert self.sim.date == hsi_event_scheduled[0][0] + the_hsi_event = hsi_event_scheduled[0][1] + assert patient_id == the_hsi_event.target + + # Run the HSI event + the_hsi_event.run(squeeze_factor=0.0) + + # Check that the person is not longer stunted + assert self.shared_sim_df.at[patient_id, 'un_HAZ_category'] == 'HAZ>=-2' + # Check that there is a follow-up appointment scheduled + hsi_event_scheduled_after_first_appt = [ + ev for ev in self.shared_sim_healthsystem.find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + assert 2 == len(hsi_event_scheduled_after_first_appt) + assert (self.sim.date + pd.DateOffset(months=6)) == hsi_event_scheduled_after_first_appt[ + 1 + ][0] + the_follow_up_hsi_event = hsi_event_scheduled_after_first_appt[1][1] + + # Run the Follow-up HSI event + the_follow_up_hsi_event.run(squeeze_factor=0.0) + + # Check that after running the following appointments there are no further appointments scheduled + assert hsi_event_scheduled_after_first_appt == [ + ev for ev in self.sim.modules['HealthSystem'].find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + + def test_routine_assessment_for_chronic_undernutrition_if_stunted_but_no_checking( + self, + patient_id, + changes_patient_properties, + changes_module_properties, + clears_hsi_queue, + ): + """Check that a call to `do_at_generic_first_appt` does not lead to an HSI for a stunted + child, if there is no checking/diagnosis.""" + self.shared_sim_df.loc[patient_id, "age_years"] = 2 + self.shared_sim_df.loc[patient_id, "un_HAZ_category"] = "HAZ<-3" + + self.this_module.parameters["prob_stunting_diagnosed_at_generic_appt"] = 0.0 + + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + # Subject the person to `do_at_generic_first_appt` + self.this_module.do_at_generic_first_appt(patient_id=patient_id, patient_details=patient_details) + + # Check that there is no HSI scheduled for this person + hsi_event_scheduled = [ + ev[1] + for ev in self.shared_sim_healthsystem.find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + assert 0 == len(hsi_event_scheduled) + + # Then make the probability of stunting checking/diagnosis 1.0 + # and check the HSI is scheduled for this person + self.this_module.parameters["prob_stunting_diagnosed_at_generic_appt"] = 1.0 + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, patient_details=patient_details + ) + hsi_event_scheduled = [ + ev[1] + for ev in self.shared_sim_healthsystem.find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + assert 1 == len(hsi_event_scheduled) + + def test_routine_assessment_for_chronic_undernutrition_if_not_stunted( + self, + patient_id, + changes_patient_properties, + changes_module_properties, + clears_hsi_queue, + ): + """Check that a call to `do_at_generic_first_appt` does not lead to an HSI if there is no + stunting.""" + # Make one person have no stunting + self.shared_sim_df.loc[patient_id, 'age_years'] = 2 + self.shared_sim_df.loc[patient_id, 'un_HAZ_category'] = 'HAZ>=-2' + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + # Subject the person to `do_at_generic_first_appt` + self.this_module.do_at_generic_first_appt(patient_id=patient_id, patient_details=patient_details) + + # Check that there is no HSI scheduled for this person + hsi_event_scheduled = [ + ev[1] + for ev in self.shared_sim_healthsystem.find_events_for_person(patient_id) + if isinstance(ev[1], HSI_Stunting_ComplementaryFeeding) + ] + assert 0 == len(hsi_event_scheduled) def test_math_of_incidence_calcs(seed): From 9b4673f78a8cba205c20a467ed41b3f37d6b1e6e Mon Sep 17 00:00:00 2001 From: willGraham01 <1willgraham@gmail.com> Date: Wed, 24 Apr 2024 12:36:44 +0100 Subject: [PATCH 3/4] refactor some of the diarrhoea tests as PoConcept --- tests/conftest.py | 67 +++- tests/test_diarrhoea.py | 838 ++++++++++++++++++---------------------- tests/test_stunting.py | 2 +- 3 files changed, 441 insertions(+), 466 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index f3fbbc1fd4..81e0b193f3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ """Collection of shared fixtures""" from __future__ import annotations +from copy import copy import os from pathlib import Path from typing import TYPE_CHECKING, List @@ -9,10 +10,14 @@ from tlo import Date, Module, Simulation from tlo.methods import ( demography, + diarrhoea, enhanced_lifestyle, + healthburden, + healthseekingbehaviour, healthsystem, simplified_births, stunting, + symptommanager ) DEFAULT_SEED = 83563095832589325021 @@ -77,13 +82,20 @@ def small_shared_sim(seed, jan_1st_2010, resource_filepath): sim = Simulation(start_date=jan_1st_2010, seed=seed) sim.register( demography.Demography(resourcefilepath=resource_filepath), + diarrhoea.Diarrhoea(resourcefilepath=resource_filepath, do_checks=True), + diarrhoea.DiarrhoeaPropertiesOfOtherModules(), enhanced_lifestyle.Lifestyle(resourcefilepath=resource_filepath), + healthburden.HealthBurden(resourcefilepath=resource_filepath), + healthseekingbehaviour.HealthSeekingBehaviour( + resourcefilepath=resource_filepath, + force_any_symptom_to_lead_to_healthcareseeking=True, + ), healthsystem.HealthSystem( resourcefilepath=resource_filepath, cons_availability="all" ), simplified_births.SimplifiedBirths(resourcefilepath=resource_filepath), stunting.Stunting(resourcefilepath=resource_filepath), - stunting.StuntingPropertiesOfOtherModules(), + symptommanager.SymptomManager(resourcefilepath=resource_filepath), ) sim.make_initial_population(n=100) sim.simulate(end_date=sim.start_date) @@ -155,6 +167,18 @@ def _attach_to_shared_sim(self, small_shared_sim): simulation fixture defined above. This ensures that every test is run with the persistent simulation object in its context. + NOTE: this is not strictly necessary in the current implementation + where we only have one simulation object to share; as we could + just pass the shared_small_sim explicitly to every test in the + (derived) class. + However, it does make accessing the simulation much more similar + to the main codebase (via self.sim.XXX rather than + shared_small_sim.XXX) and means we save on explicitly passing the + same fixture to a lot of tests since we do it automatically. + If we later define another simulation object that we want to share + between another set of tests, we can re-use this base class for + that purpose too, further saving on code repetition. + WARNING: Writes to the shared simulation object will thus be persistent between tests! If a test needs to modify module parameters, use a fresh HSI queue, or similar, use the setup @@ -168,15 +192,35 @@ def clears_hsi_queue(self): Flags this test as needing to clear the HSI event queue in the shared simulation. - Using this fixture will cause pytest to cache the current - HSI queue of the shared simulation, clear the queue, run - the test, then restore the old queue during test teardown. + Using this fixture will cause pytest to: + - Cache the current HSI queue of the shared simulation, + - Clear the queue, + - Run the test, + - Restore the old queue during test teardown. + The queue can safely be manually cleared again during the + test if this is necessary (EG if testing two calls to the + HSI scheduler). """ cached_queue = list(self.shared_sim_healthsystem.HSI_EVENT_QUEUE) self.shared_sim_healthsystem.reset_queue() yield self.shared_sim_healthsystem.HSI_EVENT_QUEUE = list(cached_queue) + @pytest.fixture(scope="function") + def changes_event_queue(self): + """ + Flags the test as needing to change the simulation + event queue, normally to check that certain events + have been scheduled based on treatment routines. + + Using this fixture will cause pytest to cache the + event queue prior to running the test, then restore + the event queue to this state at the end of the test. + """ + old_event_queue = copy(self.sim.event_queue) + yield + self.sim.event_queue = old_event_queue + @pytest.fixture(scope="function") def changes_module_properties(self): """ @@ -207,3 +251,18 @@ def changes_patient_properties(self, patient_id: int | List[int]): cached_values = self.shared_sim_df.loc[patient_id].copy() yield self.shared_sim_df.loc[patient_id] = cached_values + + @pytest.fixture(scope="function") + def changes_sim_date(self): + """ + Flags this test as needing to manually change the date of the + shared simulation; typically needed when testing cures or deaths + that are scheduled then occur. + + Using this fixture will cause pytest to cache the date of the + shared simulation prior to running the test, then restore the + simulation to that date during teardown. + """ + old_date = Date(self.sim.date) + yield + self.sim.date = Date(old_date) diff --git a/tests/test_diarrhoea.py b/tests/test_diarrhoea.py index 1e43550541..82a8010bdf 100644 --- a/tests/test_diarrhoea.py +++ b/tests/test_diarrhoea.py @@ -4,10 +4,11 @@ import os from itertools import product from pathlib import Path +from typing import Any, Dict, Literal, Optional import pandas as pd -import pytest from pandas import DateOffset +import pytest from tlo import Date, Module, Simulation, logging from tlo.analysis.utils import parse_log_file @@ -34,8 +35,9 @@ ) from tlo.methods.hsi_generic_first_appts import HSI_GenericNonEmergencyFirstAppt -resourcefilepath = Path(os.path.dirname(__file__)) / '../resources' +from .conftest import _BaseSharedSim +resourcefilepath = Path(os.path.dirname(__file__)) / "../resources" def check_dtypes(simulation): # check types of columns @@ -348,282 +350,388 @@ def run(spurious_symptoms): # run with spurious symptoms run(spurious_symptoms=True) +class TestDiarrhoea_SharedSim(_BaseSharedSim): + """ + Tests for the Diarrhoea module that make use of the shared simulation. + + Note: original tests used 200 initial pop size compared to the shared + simulation's 100, but the tests in this class appear agnostic to the + total population size (they directly edit a single patient). + """ + + module: str = "Diarrhoea" + + @pytest.fixture(scope="class") + def patient_id(self) -> int: + return 0 + + @pytest.fixture(scope="function") + def changes_module_properties(self): + """ + The Diarrhoea.models property takes a copy of the + module's parameters on initialisation, so is not + affected by our temporary parameter change from the + base class. + + Overwriting the base fixture to remedy this. + """ + old_param_values = dict(self.this_module.parameters) + yield + self.this_module.parameters = dict(old_param_values) + self.this_module.models.p = self.this_module.parameters + + def patient_with_symptoms( + self, + gi_type: Literal["bloody", "watery"] = "bloody", + dehydration: Literal["severe", "some"] = "severe", + scheduled_date_recovery: Date = pd.NaT, + scheduled_date_death: Optional[Date] = None, + ) -> Dict[str, Any]: + """ + Returns a dictionary of patient properties that correspond + to a child with diarrhoea symptoms. + + The extent and severity of the symptoms can be adjusted by + providing the relevant input values. The default set of + values is a child with bloody diarrhoea and severe + dehydration. + """ + if scheduled_date_death is None: + scheduled_date_death = self.sim.date + DateOffset(days=2) + return { + "age_years": 2, + "age_exact_years": 2.0, + "gi_has_diarrhoea": True, + "gi_pathogen": "shigella", + "gi_type": gi_type, + "gi_dehydration": dehydration, + "gi_duration_longer_than_13days": True, + "gi_date_of_onset": self.sim.date, + "gi_date_end_of_last_episode": self.sim.date + DateOffset(days=20), + "gi_scheduled_date_recovery": scheduled_date_recovery, + "gi_scheduled_date_death": scheduled_date_death, + "gi_treatment_date": pd.NaT, + } -def test_do_when_presentation_with_diarrhoea_severe_dehydration(seed): - """Check that when someone presents with diarrhoea and severe dehydration, the correct HSI is created""" - - start_date = Date(2010, 1, 1) - popsize = 200 # smallest population size that works - - sim = Simulation(start_date=start_date, seed=seed) - # Register the appropriate modules - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem( - resourcefilepath=resourcefilepath, - disable=False, - cons_availability='all' - ), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour( - resourcefilepath=resourcefilepath, - force_any_symptom_to_lead_to_healthcareseeking=True # every symptom leads to health-care seeking - ), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - - # Make DxTest for danger signs perfect: - sim.modules['Diarrhoea'].parameters['sensitivity_severe_dehydration_visual_inspection'] = 1.0 - sim.modules['Diarrhoea'].parameters['specificity_severe_dehydration_visual_inspection'] = 1.0 - - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) - df = sim.population.props - - # Set that person_id=0 is a child with bloody diarrhoea and severe dehydration - person_id = 0 - props_new = { - 'age_years': 2, - 'age_exact_years': 2.0, - 'gi_has_diarrhoea': True, - 'gi_pathogen': 'shigella', - 'gi_type': 'bloody', - 'gi_dehydration': 'severe', - 'gi_duration_longer_than_13days': True, - 'gi_date_of_onset': sim.date, - 'gi_date_end_of_last_episode': sim.date + DateOffset(days=20), - 'gi_scheduled_date_recovery': pd.NaT, - 'gi_scheduled_date_death': sim.date + DateOffset(days=2), - 'gi_treatment_date': pd.NaT, - } - df.loc[person_id, props_new.keys()] = props_new.values() - generic_hsi = HSI_GenericNonEmergencyFirstAppt( - module=sim.modules["HealthSeekingBehaviour"], person_id=person_id - ) - patient_details = sim.population.row_in_readonly_form(person_id) - - def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): - return generic_hsi.healthcare_system.dx_manager.run_dx_test( - tests, - hsi_event=generic_hsi, - use_dict_for_single=use_dict, - report_dxtest_tried=report_tried, + def test_do_when_presentation_with_diarrhoea_severe_dehydration( + self, + patient_id, + changes_module_properties, + changes_patient_properties, + clears_hsi_queue, + ): + """ + Check that when someone presents with diarrhoea and severe + dehydration, the correct HSI is created. + """ + # Make DxTest for danger signs perfect: + self.this_module.parameters['sensitivity_severe_dehydration_visual_inspection'] = 1.0 + self.this_module.parameters['specificity_severe_dehydration_visual_inspection'] = 1.0 + + # Set patient_id to be a child with bloody diarrhoea and severe dehydration + new_props = self.patient_with_symptoms() + self.shared_sim_df.loc[patient_id, new_props.keys()] = new_props.values() + generic_hsi = HSI_GenericNonEmergencyFirstAppt( + module=self.sim.modules["HealthSeekingBehaviour"], person_id=patient_id ) - - sim.modules['HealthSystem'].reset_queue() - sim.modules['Diarrhoea'].parameters['prob_hospitalization_on_danger_signs'] = 1.0 - sim.modules["Diarrhoea"].do_at_generic_first_appt( - patient_id=person_id, - patient_details=patient_details, - diagnosis_function=diagnosis_fn, - ) - evs = sim.modules['HealthSystem'].find_events_for_person(person_id) - - assert 1 == len(evs) - assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Inpatient) - - # 2) If DxTest of danger signs perfect but 0% chance of referral --> Inpatient HSI should not be created - sim.modules['HealthSystem'].reset_queue() - sim.modules['Diarrhoea'].parameters['prob_hospitalization_on_danger_signs'] = 0.0 - sim.modules["Diarrhoea"].do_at_generic_first_appt( - patient_id=person_id, - patient_details=patient_details, - diagnosis_function=diagnosis_fn, - ) - evs = sim.modules['HealthSystem'].find_events_for_person(person_id) - assert 1 == len(evs) - assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) - - -def test_do_when_presentation_with_diarrhoea_severe_dehydration_dxtest_notfunctional(seed): - """Check that when someone presents with diarrhoea and severe dehydration but the DxTest for danger signs - is not functional (0% sensitivity, 0% specificity) that an Outpatient appointment is created""" - - start_date = Date(2010, 1, 1) - popsize = 200 # smallest population size that works - - sim = Simulation(start_date=start_date, seed=seed) - # Register the appropriate modules - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem( - resourcefilepath=resourcefilepath, - disable=False, - cons_availability='all' - ), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour( - resourcefilepath=resourcefilepath, - force_any_symptom_to_lead_to_healthcareseeking=True # every symptom leads to health-care seeking - ), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - - # Make DxTest for danger signs not functional: - sim.modules['Diarrhoea'].parameters['sensitivity_severe_dehydration_visual_inspection'] = 0.0 - sim.modules['Diarrhoea'].parameters['specificity_severe_dehydration_visual_inspection'] = 0.0 - - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) - df = sim.population.props - - # Set that person_id=0 is a child with bloody diarrhoea and severe dehydration - person_id = 0 - props_new = { - 'age_years': 2, - 'age_exact_years': 2.0, - 'gi_has_diarrhoea': True, - 'gi_pathogen': 'shigella', - 'gi_type': 'bloody', - 'gi_dehydration': 'severe', - 'gi_duration_longer_than_13days': True, - 'gi_date_of_onset': sim.date, - 'gi_date_end_of_last_episode': sim.date + DateOffset(days=20), - 'gi_scheduled_date_recovery': pd.NaT, - 'gi_scheduled_date_death': sim.date + DateOffset(days=2), - 'gi_treatment_date': pd.NaT, - } - df.loc[person_id, props_new.keys()] = props_new.values() - generic_hsi = HSI_GenericNonEmergencyFirstAppt( - module=sim.modules['HealthSeekingBehaviour'], person_id=person_id) - patient_details = sim.population.row_in_readonly_form(person_id) - - def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): - return generic_hsi.healthcare_system.dx_manager.run_dx_test( - tests, - hsi_event=generic_hsi, - use_dict_for_single=use_dict, - report_dxtest_tried=report_tried, + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): + return generic_hsi.healthcare_system.dx_manager.run_dx_test( + tests, + hsi_event=generic_hsi, + use_dict_for_single=use_dict, + report_dxtest_tried=report_tried, + ) + + self.this_module.parameters['prob_hospitalization_on_danger_signs'] = 1.0 + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, + patient_details=patient_details, + diagnosis_function=diagnosis_fn, ) - - # Only an out-patient appointment should be created as the DxTest for danger signs is not functional. - sim.modules['Diarrhoea'].parameters['prob_hospitalization_on_danger_signs'] = 0.0 - sim.modules['HealthSystem'].reset_queue() - sim.modules["Diarrhoea"].do_at_generic_first_appt( - patient_id=person_id, - patient_details=patient_details, - diagnosis_function=diagnosis_fn, - ) - evs = sim.modules['HealthSystem'].find_events_for_person(person_id) - assert 1 == len(evs) - assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) - - -def test_do_when_presentation_with_diarrhoea_non_severe_dehydration(seed): - """Check that when someone presents with diarrhoea and non-severe dehydration, the out-patient HSI is created""" - - start_date = Date(2010, 1, 1) - popsize = 200 # smallest population size that works - - sim = Simulation(start_date=start_date, seed=seed) - # Register the appropriate modules - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem( - resourcefilepath=resourcefilepath, - disable=False, - cons_availability='all' - ), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour( - resourcefilepath=resourcefilepath, - force_any_symptom_to_lead_to_healthcareseeking=True # every symptom leads to health-care seeking - ), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - - # Make DxTest for danger signs perfect: - sim.modules['Diarrhoea'].parameters['sensitivity_severe_dehydration_visual_inspection'] = 1.0 - sim.modules['Diarrhoea'].parameters['specificity_severe_dehydration_visual_inspection'] = 1.0 - - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) - df = sim.population.props - - # Set that person_id=0 is a child with bloody diarrhoea and 'some' dehydration - person_id = 0 - props_new = { - 'age_years': 2, - 'age_exact_years': 2.0, - 'gi_has_diarrhoea': True, - 'gi_pathogen': 'shigella', - 'gi_type': 'bloody', - 'gi_dehydration': 'some', - 'gi_duration_longer_than_13days': True, - 'gi_date_of_onset': sim.date, - 'gi_date_end_of_last_episode': sim.date + DateOffset(days=20), - 'gi_scheduled_date_recovery': pd.NaT, - 'gi_scheduled_date_death': sim.date + DateOffset(days=2), - 'gi_treatment_date': pd.NaT, - } - df.loc[person_id, props_new.keys()] = props_new.values() - generic_hsi = HSI_GenericNonEmergencyFirstAppt( - module=sim.modules['HealthSeekingBehaviour'], person_id=person_id) - patient_details = sim.population.row_in_readonly_form(person_id) - - def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): - return generic_hsi.healthcare_system.dx_manager.run_dx_test( - tests, - hsi_event=generic_hsi, - use_dict_for_single=use_dict, - report_dxtest_tried=report_tried, + evs = self.shared_sim_healthsystem.find_events_for_person(patient_id) + + assert 1 == len(evs) + assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Inpatient) + + # 2) If DxTest of danger signs perfect but 0% chance of referral --> Inpatient HSI should not be created + self.shared_sim_healthsystem.reset_queue() + self.this_module.parameters['prob_hospitalization_on_danger_signs'] = 0.0 + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, + patient_details=patient_details, + diagnosis_function=diagnosis_fn, ) - # 1) Outpatient HSI should be created - sim.modules["HealthSystem"].reset_queue() - sim.modules["Diarrhoea"].do_at_generic_first_appt( - patient_id=person_id, patient_details=patient_details, diagnosis_function=diagnosis_fn - ) - evs = sim.modules["HealthSystem"].find_events_for_person(person_id) - - assert 1 == len(evs) - assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) - - -def test_run_each_of_the_HSI(seed): - """Check that HSI specified can be run correctly""" - start_date = Date(2010, 1, 1) - popsize = 200 # smallest population size that works + evs = self.shared_sim_healthsystem.find_events_for_person(patient_id) + assert 1 == len(evs) + assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) + + def test_do_when_presentation_with_diarrhoea_severe_dehydration_dxtest_notfunctional( + self, + patient_id, + changes_module_properties, + changes_patient_properties, + clears_hsi_queue, + ): + """ + Check that when someone presents with diarrhoea and severe dehydration but the DxTest for danger signs + is not functional (0% sensitivity, 0% specificity) that an Outpatient appointment is created + """ + # Make DxTest for danger signs not functional: + self.this_module.parameters['sensitivity_severe_dehydration_visual_inspection'] = 0.0 + self.this_module.parameters['specificity_severe_dehydration_visual_inspection'] = 0.0 + + # Set that patient_id is a child with bloody diarrhoea and severe dehydration + new_props = self.patient_with_symptoms() + self.shared_sim_df.loc[patient_id, new_props.keys()] = new_props.values() + generic_hsi = HSI_GenericNonEmergencyFirstAppt( + module=self.sim.modules['HealthSeekingBehaviour'], person_id=patient_id) + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): + return generic_hsi.healthcare_system.dx_manager.run_dx_test( + tests, + hsi_event=generic_hsi, + use_dict_for_single=use_dict, + report_dxtest_tried=report_tried, + ) + + # Only an out-patient appointment should be created as the DxTest for danger signs is not functional. + self.this_module.parameters['prob_hospitalization_on_danger_signs'] = 0.0 + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, + patient_details=patient_details, + diagnosis_function=diagnosis_fn, + ) + evs = self.shared_sim_healthsystem.find_events_for_person(patient_id) + assert 1 == len(evs) + assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) + + def test_do_when_presentation_with_diarrhoea_non_severe_dehydration( + self, + patient_id, + changes_module_properties, + changes_patient_properties, + clears_hsi_queue, + ): + """ + Check that when someone presents with diarrhoea and non-severe + dehydration, the out-patient HSI is created. + """ + # Make DxTest for danger signs perfect: + self.this_module.parameters['sensitivity_severe_dehydration_visual_inspection'] = 1.0 + self.this_module.parameters['specificity_severe_dehydration_visual_inspection'] = 1.0 + + new_props = self.patient_with_symptoms(dehydration="some") + self.shared_sim_df.loc[patient_id, new_props.keys()] = new_props.values() + generic_hsi = HSI_GenericNonEmergencyFirstAppt( + module=self.sim.modules['HealthSeekingBehaviour'], person_id=patient_id) + patient_details = self.sim.population.row_in_readonly_form(patient_id) + + def diagnosis_fn(tests, use_dict: bool = False, report_tried: bool = False): + return generic_hsi.healthcare_system.dx_manager.run_dx_test( + tests, + hsi_event=generic_hsi, + use_dict_for_single=use_dict, + report_dxtest_tried=report_tried, + ) + # 1) Outpatient HSI should be created + self.this_module.do_at_generic_first_appt( + patient_id=patient_id, patient_details=patient_details, diagnosis_function=diagnosis_fn + ) + evs = self.shared_sim_healthsystem.find_events_for_person(patient_id) - sim = Simulation(start_date=start_date, seed=seed) + assert 1 == len(evs) + assert isinstance(evs[0][1], HSI_Diarrhoea_Treatment_Outpatient) - # Register the appropriate modules - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem( - resourcefilepath=resourcefilepath, - disable=False, - cons_availability='all' - ), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour( - resourcefilepath=resourcefilepath, - force_any_symptom_to_lead_to_healthcareseeking=True # every symptom leads to health-care seeking - ), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) + def test_run_each_of_the_HSI(self, patient_id, changes_patient_properties): + """Check that HSI specified can be run correctly""" + # The Out-patient HSI + hsi_outpatient = HSI_Diarrhoea_Treatment_Outpatient( + person_id=patient_id, module=self.this_module + ) + hsi_outpatient.run(squeeze_factor=0) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) + # The In-patient HSI + hsi_outpatient = HSI_Diarrhoea_Treatment_Inpatient( + person_id=patient_id, module=self.this_module + ) + hsi_outpatient.run(squeeze_factor=0) + + def test_effect_of_vaccine(self, changes_module_properties): + """ + Check that if the vaccine is perfect, no one infected with + rotavirus and who has the vaccine gets severe dehydration. + + NOTE: disable_and_reject_all was previously set to "True" before + this test was refactored to use the shared simulation, however + this test is actually agnostic to this setting. + """ + # Get the method that determines dehydration + get_dehydration = self.this_module.models.get_dehydration + + # increase probability to ensure at least one case of severe dehydration when vaccine is imperfect + self.this_module.parameters["prob_dehydration_by_rotavirus"] = 1.0 + self.this_module.parameters["prob_dehydration_by_shigella"] = 1.0 + + # 1) Make effect of vaccine perfect + self.this_module.parameters[ + "rr_severe_dehydration_due_to_rotavirus_with_R1_under1yo" + ] = 0.0 + self.this_module.parameters[ + "rr_severe_dehydration_due_to_rotavirus_with_R1_over1yo" + ] = 0.0 + + # Check that if person has vaccine and is infected with rotavirus, there is never severe dehydration... + assert "severe" not in [ + get_dehydration(pathogen="rotavirus", va_rota_all_doses=True, age_years=1) + for _ in range(100) + ] + assert "severe" not in [ + get_dehydration(pathogen="rotavirus", va_rota_all_doses=True, age_years=4) + for _ in range(100) + ] + + # ... but if no vaccine or infected with another pathogen, then sometimes it is severe dehydration. + assert "severe" in [ + get_dehydration(pathogen="rotavirus", va_rota_all_doses=False, age_years=1) + for _ in range(100) + ] + assert "severe" in [ + get_dehydration(pathogen="shigella", va_rota_all_doses=True, age_years=1) + for _ in range(100) + ] + + # 2) Make effect of vaccine imperfect + self.this_module.parameters[ + "rr_severe_dehydration_due_to_rotavirus_with_R1_under1yo" + ] = 0.5 + self.this_module.parameters[ + "rr_severe_dehydration_due_to_rotavirus_with_R1_over1yo" + ] = 0.5 + + # Check that if the vaccine is imperfect and the person is infected with + # rotavirus, then there sometimes is severe dehydration. + assert "severe" in [ + get_dehydration(pathogen="rotavirus", va_rota_all_doses=True, age_years=1) + for _ in range(100) + ] + assert "severe" in [ + get_dehydration(pathogen="rotavirus", va_rota_all_doses=True, age_years=2) + for _ in range(100) + ] + + def test_check_perfect_treatment_leads_to_zero_risk_of_death( + self, + ): + """ + Check that for any permutation of condition, + if the treatment is successful, then it prevents death. + + NOTE: disable_and_reject_all was previously set to "True" before + this test was refactored to use the shared simulation, however + this test is actually agnostic to this setting. + """ + for ( + _pathogen, + _type, + _duration_longer_than_13days, + _dehydration, + _age_exact_years, + _ri_current_infection_status, + _untreated_hiv, + _un_clinical_acute_malnutrition, + ) in product( + self.this_module.pathogens, + ["watery", "bloody"], + [False, True], + ["none", "some", "severe"], + range(0, 4, 1), + [False, True], + [False, True], + ["MAM", "SAM", "well"], + ): + # Define the argument to `_get_probability_that_treatment_blocks_death` + # to represent successful treatment + _args = { + "pathogen": _pathogen, + "type": ( + _type, + "watery", + ), # <-- successful treatment removes blood in diarrhoea + "duration_longer_than_13days": _duration_longer_than_13days, + "dehydration": ( + _dehydration, + "none", + ), # <-- successful treatment removes dehydration + "age_exact_years": _age_exact_years, + "ri_current_infection_status": _ri_current_infection_status, + "untreated_hiv": _untreated_hiv, + "un_clinical_acute_malnutrition": _un_clinical_acute_malnutrition, + } + + assert ( + 1.0 + == self.this_module.models._get_probability_that_treatment_blocks_death( + **_args + ) + ), (f"Perfect treatment does not prevent death: {_args=}") + + def test_do_treatment_for_those_that_will_not_die( + self, + patient_id, + changes_patient_properties, + changes_sim_date, + changes_event_queue, + ): + """ + Check that when someone who will not die and is provided with + treatment and gets zinc, that the date of cure is brought + forward. + """ + scheduled_date_recovery = self.sim.date + DateOffset(days=10) + new_props = self.patient_with_symptoms( + gi_type="watery", + dehydration="some", + scheduled_date_recovery=scheduled_date_recovery, + scheduled_date_death=pd.NaT, + ) + self.shared_sim_df.loc[patient_id, new_props.keys()] = new_props.values() + self.sim.modules["SymptomManager"].change_symptom( + person_id=patient_id, + symptom_string="diarrhoea", + add_or_remove="+", + disease_module=self.this_module, + ) + # Run 'do_treatment' from an out-patient HSI. + in_patient_hsi = HSI_Diarrhoea_Treatment_Outpatient( + module=self.this_module, person_id=patient_id + ) + self.this_module.do_treatment(person_id=patient_id, hsi_event=in_patient_hsi) + + # check that a Cure Event is scheduled for earlier + evs = self.sim.find_events_for_person(patient_id) + assert 1 == len(evs) + assert isinstance(evs[0][1], DiarrhoeaCureEvent) + assert evs[0][0] == scheduled_date_recovery - pd.DateOffset( + days=self.this_module.parameters[ + "number_of_days_reduced_duration_with_zinc" + ] + ) - # The Out-patient HSI - hsi_outpatient = HSI_Diarrhoea_Treatment_Outpatient(person_id=0, module=sim.modules['Diarrhoea']) - hsi_outpatient.run(squeeze_factor=0) + # Run the Cure Event and check episode is ended. + self.sim.date = evs[0][0] + evs[0][1].apply(person_id=patient_id) + assert not self.shared_sim_df.at[patient_id, "gi_has_diarrhoea"] - # The In-patient HSI - hsi_outpatient = HSI_Diarrhoea_Treatment_Inpatient(person_id=0, module=sim.modules['Diarrhoea']) - hsi_outpatient.run(squeeze_factor=0) + # Check that a recovery event occurring later has no effect and does not error. + self.sim.date = scheduled_date_recovery + recovery_event = DiarrhoeaNaturalRecoveryEvent( + module=self.this_module, person_id=patient_id + ) + recovery_event.apply(person_id=patient_id) + assert not self.shared_sim_df.at[patient_id, "gi_has_diarrhoea"] def test_does_treatment_prevent_death(seed): @@ -847,198 +955,6 @@ def test_do_treatment_for_those_that_will_die_if_consumables_not_available(seed) assert pd.notnull(df.at[person_id, 'gi_treatment_date']) -def test_do_treatment_for_those_that_will_not_die(seed): - """Check that when someone who will not die and is provided with treatment and gets zinc, that the date of cure is - brought forward""" - - # ** If consumables are available **: - start_date = Date(2010, 1, 1) - popsize = 200 # smallest population size that works - sim = Simulation(start_date=start_date, seed=seed) - # Register the appropriate modules - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem( - resourcefilepath=resourcefilepath, - disable=False, - cons_availability='all' - ), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour( - resourcefilepath=resourcefilepath, - force_any_symptom_to_lead_to_healthcareseeking=True # every symptom leads to health-care seeking - ), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) - df = sim.population.props - - # Set that person_id=0 is a child with watery diarrhoea and 'some' dehydration: - person_id = 0 - scheduled_date_recovery = sim.date + DateOffset(days=10) - props_new = { - 'age_years': 2, - 'age_exact_years': 2.0, - 'gi_has_diarrhoea': True, - 'gi_pathogen': 'shigella', - 'gi_type': 'watery', - 'gi_dehydration': 'some', - 'gi_duration_longer_than_13days': True, - 'gi_date_of_onset': sim.date, - 'gi_date_end_of_last_episode': sim.date + DateOffset(days=20), - 'gi_scheduled_date_recovery': scheduled_date_recovery, - 'gi_scheduled_date_death': pd.NaT, - 'gi_treatment_date': pd.NaT, - } - df.loc[person_id, props_new.keys()] = props_new.values() - sim.modules['SymptomManager'].change_symptom( - person_id=0, - symptom_string='diarrhoea', - add_or_remove='+', - disease_module=sim.modules['Diarrhoea'] - ) - # Run 'do_treatment' from an out-patient HSI. - in_patient_hsi = HSI_Diarrhoea_Treatment_Outpatient( - module=sim.modules['Diarrhoea'], person_id=person_id) - sim.modules['Diarrhoea'].do_treatment(person_id=person_id, hsi_event=in_patient_hsi) - - # check that a Cure Event is scheduled for earlier - evs = sim.find_events_for_person(person_id) - assert 1 == len(evs) - assert isinstance(evs[0][1], DiarrhoeaCureEvent) - assert evs[0][0] == scheduled_date_recovery - \ - pd.DateOffset(days=sim.modules['Diarrhoea'].parameters['number_of_days_reduced_duration_with_zinc']) - - # Run the Cure Event and check episode is ended. - sim.date = evs[0][0] - evs[0][1].apply(person_id=person_id) - assert not df.at[person_id, 'gi_has_diarrhoea'] - - # Check that a recovery event occurring later has no effect and does not error. - sim.date = scheduled_date_recovery - recovery_event = DiarrhoeaNaturalRecoveryEvent(module=sim.modules['Diarrhoea'], person_id=person_id) - recovery_event.apply(person_id=person_id) - assert not df.at[person_id, 'gi_has_diarrhoea'] - - -def test_effect_of_vaccine(seed): - """Check that if the vaccine is perfect, no one infected with rotavirus and who has the vaccine gets severe - dehydration.""" - - # Create dummy simulation - start_date = Date(2010, 1, 1) - popsize = 200 - - sim = Simulation(start_date=start_date, seed=seed) - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem(resourcefilepath=resourcefilepath, disable_and_reject_all=True), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour(resourcefilepath=resourcefilepath), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=start_date) - - # Get the method that determines dehydration - get_dehydration = sim.modules['Diarrhoea'].models.get_dehydration - - # increase probability to ensure at least one case of severe dehydration when vaccine is imperfect - sim.modules['Diarrhoea'].parameters['prob_dehydration_by_rotavirus'] = 1.0 - sim.modules['Diarrhoea'].parameters['prob_dehydration_by_shigella'] = 1.0 - - # 1) Make effect of vaccine perfect - sim.modules['Diarrhoea'].parameters['rr_severe_dehydration_due_to_rotavirus_with_R1_under1yo'] = 0.0 - sim.modules['Diarrhoea'].parameters['rr_severe_dehydration_due_to_rotavirus_with_R1_over1yo'] = 0.0 - - # Check that if person has vaccine and is infected with rotavirus, there is never severe dehydration... - assert 'severe' not in [get_dehydration(pathogen='rotavirus', va_rota_all_doses=True, age_years=1) - for _ in range(100)] - assert 'severe' not in [get_dehydration(pathogen='rotavirus', va_rota_all_doses=True, age_years=4) - for _ in range(100)] - - # ... but if no vaccine or infected with another pathogen, then sometimes it is severe dehydration. - assert 'severe' in [get_dehydration(pathogen='rotavirus', va_rota_all_doses=False, age_years=1) - for _ in range(100)] - assert 'severe' in [get_dehydration(pathogen='shigella', va_rota_all_doses=True, age_years=1) - for _ in range(100)] - - # 2) Make effect of vaccine imperfect - sim.modules['Diarrhoea'].parameters['rr_severe_dehydration_due_to_rotavirus_with_R1_under1yo'] = 0.5 - sim.modules['Diarrhoea'].parameters['rr_severe_dehydration_due_to_rotavirus_with_R1_over1yo'] = 0.5 - - # Check that if the vaccine is imperfect and the person is infected with rotavirus, then there sometimes is severe - # dehydration. - assert 'severe' in [get_dehydration(pathogen='rotavirus', va_rota_all_doses=True, age_years=1) - for _ in range(100)] - assert 'severe' in [get_dehydration(pathogen='rotavirus', va_rota_all_doses=True, age_years=2) - for _ in range(100)] - - -def test_check_perfect_treatment_leads_to_zero_risk_of_death(seed): - """Check that for any permutation of condition, if the treatment is successful, then it prevents death""" - - start_date = Date(2010, 1, 1) - popsize = 200 - - sim = Simulation(start_date=start_date, seed=seed) - sim.register(demography.Demography(resourcefilepath=resourcefilepath), - simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath), - enhanced_lifestyle.Lifestyle(resourcefilepath=resourcefilepath), - healthsystem.HealthSystem(resourcefilepath=resourcefilepath, disable_and_reject_all=True), - symptommanager.SymptomManager(resourcefilepath=resourcefilepath), - healthseekingbehaviour.HealthSeekingBehaviour(resourcefilepath=resourcefilepath), - healthburden.HealthBurden(resourcefilepath=resourcefilepath), - diarrhoea.Diarrhoea(resourcefilepath=resourcefilepath, do_checks=True), - diarrhoea.DiarrhoeaPropertiesOfOtherModules(), - ) - sim.make_initial_population(n=popsize) - sim.simulate(end_date=sim.start_date) - - diarrhoea_module = sim.modules['Diarrhoea'] - - for ( - _pathogen, - _type, - _duration_longer_than_13days, - _dehydration, - _age_exact_years, - _ri_current_infection_status, - _untreated_hiv, - _un_clinical_acute_malnutrition - ) in product( - diarrhoea_module.pathogens, - ['watery', 'bloody'], - [False, True], - ['none', 'some', 'severe'], - range(0, 4, 1), - [False, True], - [False, True], - ['MAM', 'SAM', 'well'] - ): - # Define the argument to `_get_probability_that_treatment_blocks_death` to represent successful treatment - _args = { - 'pathogen': _pathogen, - 'type': (_type, 'watery'), # <-- successful treatment removes blood in diarrhoea - 'duration_longer_than_13days': _duration_longer_than_13days, - 'dehydration': (_dehydration, 'none'), # <-- successful treatment removes dehydration - 'age_exact_years': _age_exact_years, - 'ri_current_infection_status': _ri_current_infection_status, - 'untreated_hiv': _untreated_hiv, - 'un_clinical_acute_malnutrition': _un_clinical_acute_malnutrition - } - - assert 1.0 == diarrhoea_module.models._get_probability_that_treatment_blocks_death(**_args), \ - f"Perfect treatment does not prevent death: {_args=}" - - @pytest.mark.slow def test_zero_deaths_when_perfect_treatment(seed): """Check that there are no deaths when treatment is perfect and there is perfect healthcare seeking, and no diff --git a/tests/test_stunting.py b/tests/test_stunting.py index da217747b3..d975b465c2 100644 --- a/tests/test_stunting.py +++ b/tests/test_stunting.py @@ -229,7 +229,7 @@ class TestStunting_SharedSim(_BaseSharedSim): module: str = "Stunting" - @pytest.fixture(scope="module") + @pytest.fixture(scope="class") def patient_id(self) -> int: return 0 From 7230953f742dafd2d3913516603eac23b20182c4 Mon Sep 17 00:00:00 2001 From: willGraham01 <1willgraham@gmail.com> Date: Wed, 24 Apr 2024 15:12:15 +0100 Subject: [PATCH 4/4] Lint pass --- tests/conftest.py | 5 +++-- tests/test_diarrhoea.py | 2 +- tests/test_stunting.py | 1 + 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 81e0b193f3..4caf9f630a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,8 @@ """Collection of shared fixtures""" from __future__ import annotations -from copy import copy + import os +from copy import copy from pathlib import Path from typing import TYPE_CHECKING, List @@ -17,7 +18,7 @@ healthsystem, simplified_births, stunting, - symptommanager + symptommanager, ) DEFAULT_SEED = 83563095832589325021 diff --git a/tests/test_diarrhoea.py b/tests/test_diarrhoea.py index 82a8010bdf..9641658752 100644 --- a/tests/test_diarrhoea.py +++ b/tests/test_diarrhoea.py @@ -7,8 +7,8 @@ from typing import Any, Dict, Literal, Optional import pandas as pd -from pandas import DateOffset import pytest +from pandas import DateOffset from tlo import Date, Module, Simulation, logging from tlo.analysis.utils import parse_log_file diff --git a/tests/test_stunting.py b/tests/test_stunting.py index d975b465c2..2efe9e17a3 100644 --- a/tests/test_stunting.py +++ b/tests/test_stunting.py @@ -16,6 +16,7 @@ from .conftest import _BaseSharedSim + def get_sim(seed): """Return simulation objection with Stunting and other necessary modules registered."""