Select Git revision
blending.py
Forked from
mhby1g21 / 360monodepth
Source project has a limited visibility.
test_acmc.py 9.35 KiB
import pytest
import argparse
import sys
import time
import shutil
import logging
from pathlib import Path
from acmc import trud, omop, main, phen, 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.yml"), # config.yml test case
("config2.yml"), # config.yml test case
("config3.yml"), # config.yml test case
],
)
def test_phen_workflow(tmp_dir, monkeypatch, caplog, config_file):
print(f"Temporary directory: {tmp_dir}") # Prints path for debugging
# init phenotype in temp directory
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
# validate phenotype and copy examples across
with caplog.at_level(logging.DEBUG):
shutil.rmtree(phen_path / phen.CONCEPTS_DIR)
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.yml")
monkeypatch.setattr(
sys, "argv", ["main.py", "phen", "validate", "-d", str(phen_path.resolve())]
)
main.main()
assert "Phenotype validated successfully" in caplog.text
# map phenotypes at read2 and read3
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 version message
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"publish",
"-d",
str(phen_path.resolve()),
"-m",
"version message",
],
)
main.main()
assert "Phenotype published successfully" in caplog.text
# diff phenotype
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-od",
str(phen_path.resolve()),
"-ov",
"0.0.1",
],
)
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 / phen.CONCEPTS_DIR)
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.yml", phen_path / "config.yml")
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
# map phenotype example 2
shutil.copy(phen_path / "config2.yml", phen_path / "config.yml")
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 0.0.1
with caplog.at_level(logging.DEBUG):
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-ov",
"0.0.1",
],
)
main.main()
assert "Phenotypes diff'd successfully" in caplog.text
# check changes
with open(phen_path / "latest_0.0.1_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.yml", phen_path / "config.yml")
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 0.0.1
with caplog.at_level(logging.DEBUG):
old_path = tmp_dir / "0.0.1"
monkeypatch.setattr(
sys,
"argv",
[
"main.py",
"phen",
"diff",
"-d",
str(phen_path.resolve()),
"-ov",
"0.0.1",
],
)
main.main()
assert "Phenotypes diff'd successfully" in caplog.text
with open(phen_path / "latest_0.0.1_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
# TEST REPO NEEDS TO BE AUTOMATED
# Create remote repo acmc-test1 (https://git.soton.ac.uk/mjbonifa/acmc-test1.git) and acmc-test2 (https://git.soton.ac.uk/mjbonifa/acmc-test2.git)
# Init repo from the remote acmc-test1
# Copy example and run map
# Publish creating a version on the remote repo
# Fork repo from acmc-test1 with remote acmc-test2
# Publish repo creating a version on the new repo