Something went wrong on our end
Select Git revision
test_acmc.py
-
mjbonifa authored
fix: added map definition to the config.yaml so that we know which maps are expected for the phenotype. It is still possible to run one of them using the -t option but they must be specified the config file. This means a user can run acmc phen map and all the required codes are created. It also reduces the chance of inconsistency between the map files generated between versions. It does not remove it entirely because it is still possible for a user to only run with a subset of the coding types but that should be discouraged. We retain the option for phenottype development because you might not want to run everything all of the time due the time it takes. Closes #40.
mjbonifa authoredfix: added map definition to the config.yaml so that we know which maps are expected for the phenotype. It is still possible to run one of them using the -t option but they must be specified the config file. This means a user can run acmc phen map and all the required codes are created. It also reduces the chance of inconsistency between the map files generated between versions. It does not remove it entirely because it is still possible for a user to only run with a subset of the coding types but that should be discouraged. We retain the option for phenottype development because you might not want to run everything all of the time due the time it takes. Closes #40.
test_acmc.py 9.72 KiB
import pytest
import argparse
import sys
import shutil
import logging
from pathlib import Path
from acmc import trud, omop, main, logging_config as lc
# setup logging
lc.setup_logger()
@pytest.fixture
def tmp_dir():
# Setup tmp directory
temp_dir = Path("./tests/tmp")
temp_dir.mkdir(parents=True, exist_ok=True)
# Yield the directory path to the test function
yield temp_dir
# Remove the directory after the test finishes
shutil.rmtree(temp_dir)
@pytest.fixture
def logger():
logger = logging.getLogger("acmc_logger")
logger.setLevel(logging.DEBUG)
stream_handler = logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
def test_phen_init_local_specified(tmp_dir, monkeypatch, caplog):
with caplog.at_level(logging.DEBUG):
phen_path = tmp_dir / "phen"
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "init", "-d", str(phen_path.resolve())]
)
# Mock input() to return "yes" to the question about reinitialising the directory
monkeypatch.setattr("builtins.input", lambda _: "y")
main.main()
assert "Phenotype initialised successfully" in caplog.text
# TODO: This test will need to be refactored so that the expected outputs match the config files
# right now it just tests that it runs successfully and does not check the contents of the output
@pytest.mark.parametrize(
"config_file",
[
("config1.yaml"), # config.yaml test case
("config2.yaml"), # config.yaml test case
("config3.yaml"), # config.yaml test case
],
)
def test_phen_workflow(tmp_dir, monkeypatch, caplog, config_file):
print(f"Temporary directory: {tmp_dir}") # Prints path for debugging
with caplog.at_level(logging.DEBUG):
phen_path = tmp_dir / "phen"
phen_path = phen_path.resolve()
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "init", "-d", str(phen_path.resolve())]
)
# Mock input() to return "yes" to the question about reinitialising the directory
monkeypatch.setattr("builtins.input", lambda _: "y")
main.main()
assert "Phenotype initialised successfully" in caplog.text
with caplog.at_level(logging.DEBUG):
# validate phenotype
# copy examples across
shutil.rmtree(phen_path / "codes")
ex_path = Path("./examples").resolve()
for item in ex_path.iterdir():
source = ex_path / item.name
destination = phen_path / item.name
if source.is_dir():
shutil.copytree(source, destination)
else:
shutil.copy(source, destination)
# copy the test file to configuration
shutil.copy(phen_path / config_file, phen_path / "config.yaml")
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "validate", "-d", str(phen_path.resolve())]
)
main.main()
assert "Phenotype validated successfully" in caplog.text
# map phenotype
for code_type in ["read2", "read3"]:
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"map",
"-d",
str(phen_path.resolve()),
"-t",
code_type,
],
)
main.main()
assert "Phenotype processed successfully" in caplog.text
# publish phenotype
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "publish", "-d", str(phen_path.resolve())]
)
main.main()
assert "Phenotype published successfully" in caplog.text
# copy phenotype'
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"copy",
"-d",
str(phen_path.resolve()),
"-td",
str(tmp_dir.resolve()),
"-v",
"v1.0.3",
],
)
main.main()
assert "Phenotype copied successfully" in caplog.text
# diff phenotype
with caplog.at_level(logging.DEBUG):
old_path = tmp_dir / "v1.0.3"
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-old",
str(old_path.resolve()),
],
)
main.main()
assert "Phenotypes diff'd successfully" in caplog.text
# TODO: Refactor this test so it is shortened, there's lots of duplicate codes
def test_diff(tmp_dir, monkeypatch, caplog):
print(f"Temporary directory: {tmp_dir}") # Prints path for debugging
# init phenotype
phen_path = tmp_dir / "phen"
phen_path = phen_path.resolve()
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "init", "-d", str(phen_path.resolve())]
)
# Mock input() to return "yes" to the question about reinitialising the directory
monkeypatch.setattr("builtins.input", lambda _: "y")
main.main()
# copy example codes
shutil.rmtree(phen_path / "codes")
ex_path = Path("./examples").resolve()
for item in ex_path.iterdir():
source = ex_path / item.name
destination = phen_path / item.name
if source.is_dir():
shutil.copytree(source, destination)
else:
shutil.copy(source, destination)
# map phenotype for config 1
shutil.copy(phen_path / "config1.yaml", phen_path / "config.yaml")
for code_type in ["read2"]:
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"map",
"-d",
str(phen_path.resolve()),
"-t",
code_type,
],
)
main.main()
assert "Phenotype processed successfully" in caplog.text
# publish phenotype
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "publish", "-d", str(phen_path.resolve())]
)
main.main()
assert "Phenotype published successfully" in caplog.text
# copy phenotype'
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"copy",
"-d",
str(phen_path.resolve()),
"-td",
str(tmp_dir.resolve()),
"-v",
"v1.0.3",
],
)
main.main()
assert "Phenotype copied successfully" in caplog.text
# map phenotype example 2
shutil.copy(phen_path / "config2.yaml", phen_path / "config.yaml")
for code_type in ["read2", "read3"]:
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"map",
"-d",
str(phen_path.resolve()),
"-t",
code_type,
],
)
main.main()
assert "Phenotype processed successfully" in caplog.text
# diff phenotype with v1.0.3
with caplog.at_level(logging.DEBUG):
old_path = tmp_dir / "v1.0.3"
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-old",
str(old_path.resolve()),
],
)
main.main()
assert "Phenotypes diff'd successfully" in caplog.text
# check changes
with open(phen_path / "v1.0.3_diff.md", "r") as file:
content = file.read()
assert "Removed concepts ['ABDO_PAIN']" in content
assert "Added concepts ['DID_NOT_ATTEND']" in content
assert "Added outputs: ['read3.csv']" in content
# map phenotype with example 3
shutil.copy(phen_path / "config3.yaml", phen_path / "config.yaml")
for code_type in ["read2", "read3", "snomed"]:
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"map",
"-d",
str(phen_path.resolve()),
"-t",
code_type,
],
)
main.main()
assert "Phenotype processed successfully" in caplog.text
# diff phenotype with v1.0.3
with caplog.at_level(logging.DEBUG):
old_path = tmp_dir / "v1.0.3"
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-old",
str(old_path.resolve()),
],
)
main.main()
assert "Phenotypes diff'd successfully" in caplog.text
with open(phen_path / "v1.0.3_diff.md", "r") as file:
content = file.read()
assert "Removed concepts ['ABDO_PAIN']" in content
assert "Added concepts ['DEPRESSION', 'DID_NOT_ATTEND', 'HYPERTENSION']" in content
assert "Added outputs: ['read3.csv', 'snomed.csv']" in content