File size: 7,495 Bytes
e69d4e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """Unit tests for the logic_consecutive_negative_responses function."""
import copd
import numpy as np
import pandas as pd
import pytest
@pytest.fixture
def exacerbation_event():
"""Dataframe index (27) of the exacerbation event of interest."""
return 27
@pytest.fixture
def first_pro_response():
"""Dataframe index (8) of the first weekly PRO response."""
return 8
@pytest.fixture
def second_pro_response(first_pro_response):
"""Dataframe index of the second weekly PRO response. Seven days after first."""
return first_pro_response + 7
@pytest.fixture
def third_pro_response(second_pro_response):
"""Dataframe index of the third weekly PRO response. Seven days after second."""
return second_pro_response + 7
@pytest.fixture
def input_df(exacerbation_event):
"""Sample input dataframe template - specific cases to be added in each test.
This initial dataframe has no PRO responses between the initial exacerbation at index
2 and the event of interest with DaysSinceLastExac=25 at index exacerbation_event (set
to 27). Interim PRO responses should be added in tests. Each row is a different day
(in chronological order). Add/subtract N from exacerbation_event to refer to N days
before or after the event by the dataframe index, e.g. exacerbation_event - 7 refers
to the day a week prior.
"""
df = pd.DataFrame({'PatientId': ['1'] * 31,
'DateOfEvent': pd.date_range('2022-01-01', '2022-01-31'),
'Q5Answered': [0] * 31,
'NegativeQ5': [np.nan] * 31,
'DaysSinceLastExac': [-1, -1, -1] + list(np.arange(1, 26)) +
list(np.arange(1, 4))})
# Add initial event to simulate DaysSinceLastExac restart from 1
df.loc[2, 'Q5Answered'] = 1
df.loc[2, 'NegativeQ5'] = 0
# Add event of interest (DaysSinceLastExac = 25)
df.loc[exacerbation_event, 'Q5Answered'] = 1
df.loc[exacerbation_event, 'NegativeQ5'] = 0
# Add a negative response 2 days after the event of interest (should not be counted)
df.loc[exacerbation_event + 2, 'Q5Answered'] = 1
df.loc[exacerbation_event + 2, 'NegativeQ5'] = 1
return df
def test_returns_one_when_no_responses(input_df, exacerbation_event):
"""Verify returns 1 (flag for removal) for no interim PRO responses."""
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 1
def test_returns_one_too_few_responses(input_df, exacerbation_event):
"""Verify returns 1 (flag for removal) for too few interim PRO responses."""
# Add a single negative response 7 days before the exacerbation event. Should fail PRO
# LOGIC because the negative response at index 29 is after the event of interest.
input_df.loc[exacerbation_event - 7, 'Q5Answered'] = 1
input_df.loc[exacerbation_event - 7, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 1
def test_returns_one_too_few_negative_responses(
input_df, exacerbation_event, second_pro_response, third_pro_response):
"""Verify returns 1 (flag for removal) for too few interim PRO responses."""
# Add a positive response and a single negative response. Should return one because
# the response at index 29 is after the period of interest.
input_df.loc[second_pro_response, 'Q5Answered'] = 1
input_df.loc[second_pro_response, 'NegativeQ5'] = 0
input_df.loc[third_pro_response, 'Q5Answered'] = 1
input_df.loc[third_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 1
def test_returns_one_too_few_consecutive_negative_responses_missing(
input_df, exacerbation_event, first_pro_response, second_pro_response,
third_pro_response):
"""Verify returns 1 (flag for removal) for too few consecutive -ve PRO responses.
Input has a missing response between the two negative responses.
"""
# Add negative responses at indices 8 and 22 (missing response at 15)
input_df.loc[first_pro_response, 'Q5Answered'] = 1
input_df.loc[first_pro_response, 'NegativeQ5'] = 1
input_df.loc[third_pro_response, 'Q5Answered'] = 1
input_df.loc[third_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 1
def test_returns_one_too_few_consecutive_negative_responses_positive(
input_df, exacerbation_event, first_pro_response, second_pro_response,
third_pro_response):
"""Verify returns 1 (flag for removal) for too few consecutive -ve PRO responses.
Input has a positive response between the two negative responses.
"""
# Add negative responses at indices 8 and 22, and a positive response at 15
input_df.loc[first_pro_response, 'Q5Answered'] = 1
input_df.loc[first_pro_response, 'NegativeQ5'] = 1
input_df.loc[second_pro_response, 'Q5Answered'] = 1
input_df.loc[second_pro_response, 'NegativeQ5'] = 0
input_df.loc[third_pro_response, 'Q5Answered'] = 1
input_df.loc[third_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 1
def test_returns_zero_enough_consecutive_negative_responses_default(
input_df, exacerbation_event, first_pro_response, second_pro_response):
"""Verify returns 0 (pass LOGIC criterion) for required consecutive -ve PRO responses.
Input has two consecutive negative responses. Should return 1 with default options.
"""
# Add negative responses at indices 8 and 15
input_df.loc[first_pro_response, 'Q5Answered'] = 1
input_df.loc[first_pro_response, 'NegativeQ5'] = 1
input_df.loc[second_pro_response, 'Q5Answered'] = 1
input_df.loc[second_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(input_df, exacerbation_event) == 0
def test_returns_one_too_few_consecutive_negative_responses_non_default(
input_df, exacerbation_event, first_pro_response, second_pro_response):
"""Verify returns 1 (flag for removal) for too few consecutive -ve PRO responses.
Input has two consecutive negative responses. Should return 0 with N=3.
"""
# Add negative responses at indices 8 and 15
input_df.loc[first_pro_response, 'Q5Answered'] = 1
input_df.loc[first_pro_response, 'NegativeQ5'] = 1
input_df.loc[second_pro_response, 'Q5Answered'] = 1
input_df.loc[second_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(
input_df, exacerbation_event, N=3) == 1
def test_returns_zero_too_few_consecutive_negative_responses_non_default(
input_df, exacerbation_event, first_pro_response, second_pro_response,
third_pro_response):
"""Verify returns 0 (pass LOGIC criterion) for required consecutive -ve PRO responses.
Input has three consecutive negative responses. Should return 0 with N=3
"""
# Add negative responses at indices 8, 15, and 22
input_df.loc[first_pro_response, 'Q5Answered'] = 1
input_df.loc[first_pro_response, 'NegativeQ5'] = 1
input_df.loc[second_pro_response, 'Q5Answered'] = 1
input_df.loc[second_pro_response, 'NegativeQ5'] = 1
input_df.loc[third_pro_response, 'Q5Answered'] = 1
input_df.loc[third_pro_response, 'NegativeQ5'] = 1
assert copd.logic_consecutive_negative_responses(
input_df, exacerbation_event, N=3) == 0
|