diff --git a/tests/conftest.py b/tests/conftest.py
index 47d6c3fa16..4caf9f630a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,8 +1,27 @@
"""Collection of shared fixtures"""
+from __future__ import annotations
+
+import os
+from copy import copy
+from pathlib import Path
+from typing import TYPE_CHECKING, List
+
import pytest
-DEFAULT_SEED = 83563095832589325021
+from tlo import Date, Module, Simulation
+from tlo.methods import (
+ demography,
+ diarrhoea,
+ enhanced_lifestyle,
+ healthburden,
+ healthseekingbehaviour,
+ healthsystem,
+ simplified_births,
+ stunting,
+ symptommanager,
+)
+DEFAULT_SEED = 83563095832589325021
def pytest_addoption(parser):
parser.addoption(
@@ -10,13 +29,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 +50,220 @@ 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")
+
+## """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),
+ 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),
+ symptommanager.SymptomManager(resourcefilepath=resource_filepath),
+ )
+ 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.
+
+ 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
+ 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,
+ - 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):
+ """
+ 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
+
+ @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..9641658752 100644
--- a/tests/test_diarrhoea.py
+++ b/tests/test_diarrhoea.py
@@ -4,6 +4,7 @@
import os
from itertools import product
from pathlib import Path
+from typing import Any, Dict, Literal, Optional
import pandas as pd
import pytest
@@ -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 4b11db1512..2efe9e17a3 100644
--- a/tests/test_stunting.py
+++ b/tests/test_stunting.py
@@ -14,6 +14,8 @@
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,143 +223,144 @@ 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)
- sim.simulate(end_date=sim.start_date)
- sim.modules['HealthSystem'].reset_queue()
-
- # 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)
-
- # Make the probability of stunting checking/diagnosis as 1.0
- sim.modules['Stunting'].parameters['prob_stunting_diagnosed_at_generic_appt'] = 1.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 an HSI scheduled for this person
- hsi_event_scheduled = [
- ev
- 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)
- assert 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
-
- # 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'
- # 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)
- 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[
- 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 sim.modules['HealthSystem'].find_events_for_person(person_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)
+class TestStunting_SharedSim(_BaseSharedSim):
+ """
+ Tests for the Stunting module that make use of the shared simulation.
+ """
+
+ module: str = "Stunting"
+
+ @pytest.fixture(scope="class")
+ 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
+ )
+
+ # 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):