27516931519

Committed 15 Jun 2026 12:24AM UTC coverage: 98.378% (+0.1%) from 98.261%

Build # 27516931519

Build Type

push

github

Committed by

Vini2

Commit Message

DOC: Update docs

Coverage Stats

182 of 185 relevant lines covered (98.38%)

7.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.05

/tests/test_cli.py

import pathlib
import pickle
from types import SimpleNamespace

import pytest

from click.testing import CliRunner

from metacoag import metacoag_pipeline, metacoag_runner
from metacoag.cli import main
from metacoag.metacoag_utils.bidirectionalmap import BidirectionalMap


__author__ = "Vijini Mallawaarachchi"
__credits__ = ["Vijini Mallawaarachchi"]


DATADIR = pathlib.Path(__file__).parent / "data"


@pytest.fixture(scope="session")
def tmp_dir(tmpdir_factory):
    return tmpdir_factory.mktemp("tmp")


@pytest.fixture(autouse=True)
def workingdir(tmp_dir, monkeypatch):
    """set the working directory for all tests"""
    monkeypatch.chdir(tmp_dir)


@pytest.fixture(scope="session")
def runner():
    """exportrc works correctly."""
    return CliRunner()


def test_metacoag_spades_run(runner, tmp_dir):
    outpath = tmp_dir
    dir_name = DATADIR / "5G_metaspades"
    graph = dir_name / "assembly_graph_with_scaffolds.gfa"
    contigs = dir_name / "contigs.fasta"
    paths = dir_name / "contigs.paths"
    abundance = dir_name / "coverm_mean_coverage.tsv"
    args = f"--assembler spades --graph {graph} --contigs {contigs} --paths {paths} --abundance {abundance} --output {outpath}".split()
    r = runner.invoke(main, args, catch_exceptions=False)
    assert r.exit_code == 0, r.output


def test_metacoag_megahit_run(runner, tmp_dir):
    outpath = tmp_dir
    dir_name = DATADIR / "5G_MEGAHIT"
    graph = dir_name / "final.gfa"
    contigs = dir_name / "final.contigs.fa"
    abundance = dir_name / "abundance.tsv"
    args = f"--assembler megahit --graph {graph} --contigs {contigs} --abundance {abundance} --output {outpath}".split()
    r = runner.invoke(main, args, catch_exceptions=False)
    assert r.exit_code == 0, r.output


def test_continue_option_is_available(runner):
    result = runner.invoke(main, ["--help"])

    assert result.exit_code == 0
    assert "--continue" in result.output


def test_bidirectional_map_can_be_checkpointed(tmp_path):
    mapping = BidirectionalMap()
    mapping[1] = "contig"
    checkpoint_path = tmp_path / "mapping.pickle"

    with open(checkpoint_path, "wb") as handle:
        pickle.dump(mapping, handle)

    with open(checkpoint_path, "rb") as handle:
        restored = pickle.load(handle)

    assert restored[1] == "contig"
    assert restored.inverse["contig"] == 1


def test_continue_resumes_after_last_completed_stage(tmp_path, monkeypatch):
    input_paths = {}
    for name in ("graph", "contigs", "abundance", "hmm"):
        path = tmp_path / name
        path.write_text(name)
        input_paths[name] = str(path)

    args = SimpleNamespace(
        assembler="custom",
        graph=input_paths["graph"],
        contigs=input_paths["contigs"],
        abundance=input_paths["abundance"],
        paths=None,
        output=str(tmp_path / "output"),
        hmm=input_paths["hmm"],
        prefix="resume",
        min_length=1000,
        p_intra=0.1,
        p_inter=0.01,
        d_limit=20,
        depth=10,
        n_mg=108,
        no_cut_tc=False,
        mg_threshold=0.5,
        bin_mg_threshold=0.33333,
        min_bin_size=200000,
        delimiter=",",
        nthreads=2,
        continue_run=False,
    )
    calls = []

    monkeypatch.setattr(metacoag_pipeline, "log_inputs", lambda *unused: None)
    monkeypatch.setattr(
        metacoag_pipeline,
        "load_graph",
        lambda *unused: (calls.append("graph") or "graph-data", "isolated"),
    )
    monkeypatch.setattr(
        metacoag_pipeline,
        "extract_features",
        lambda *unused: calls.append("features") or "feature-data",
    )
    monkeypatch.setattr(
        metacoag_pipeline,
        "parse_markers",
        lambda *unused: calls.append("markers") or "marker-data",
    )
    monkeypatch.setattr(
        metacoag_pipeline,
        "match_seed_bins",
        lambda *unused: calls.append("seed_bins") or {"seed": True},
    )

    def interrupt_propagation(*unused):
        calls.append("propagated_bins")
        raise RuntimeError("interrupted")

    monkeypatch.setattr(
        metacoag_pipeline,
        "propagate_bins",
        interrupt_propagation,
    )

    with pytest.raises(RuntimeError, match="interrupted"):
        metacoag_runner.run(args)

    config = metacoag_pipeline.PipelineConfig.from_args(args)
    checkpoint_path = metacoag_runner._checkpoint_path(config)
    assert checkpoint_path.is_file()
    assert calls == [
        "graph",
        "features",
        "markers",
        "seed_bins",
        "propagated_bins",
    ]

    calls.clear()
    args.continue_run = True

    def completed_stage_called(*unused):
        raise AssertionError("a completed stage was rerun")

    monkeypatch.setattr(metacoag_pipeline, "load_graph", completed_stage_called)
    monkeypatch.setattr(metacoag_pipeline, "extract_features", completed_stage_called)
    monkeypatch.setattr(metacoag_pipeline, "parse_markers", completed_stage_called)
    monkeypatch.setattr(metacoag_pipeline, "match_seed_bins", completed_stage_called)
    monkeypatch.setattr(
        metacoag_pipeline,
        "propagate_bins",
        lambda *unused: calls.append("propagated_bins") or {"propagated": True},
    )
    monkeypatch.setattr(
        metacoag_pipeline,
        "merge_bins",
        lambda *unused: calls.append("merge_plan") or "merge-plan",
    )
    monkeypatch.setattr(
        metacoag_pipeline,
        "write_output",
        lambda *unused: calls.append("output"),
    )

    metacoag_runner.run(args)

    assert calls == ["propagated_bins", "merge_plan", "output"]
    assert not checkpoint_path.exists()


def test_continue_rejects_changed_options(tmp_path):
    input_paths = {}
    for name in ("graph", "contigs", "abundance", "hmm"):
        path = tmp_path / name
        path.write_text(name)
        input_paths[name] = str(path)

    args = SimpleNamespace(
        assembler="custom",
        graph=input_paths["graph"],
        contigs=input_paths["contigs"],
        abundance=input_paths["abundance"],
        paths=None,
        output=str(tmp_path / "output"),
        hmm=input_paths["hmm"],
        prefix="",
        min_length=1000,
        p_intra=0.1,
        p_inter=0.01,
        d_limit=20,
        depth=10,
        n_mg=108,
        no_cut_tc=False,
        mg_threshold=0.5,
        bin_mg_threshold=0.33333,
        min_bin_size=200000,
        delimiter=",",
        nthreads=2,
        continue_run=False,
    )
    config = metacoag_pipeline.PipelineConfig.from_args(args)
    metacoag_pipeline.validate_inputs(config)
    metacoag_runner._save_checkpoint(config, "graph", {"graph_data": "graph"})

    args.continue_run = True
    args.min_length = 2000
    changed_config = metacoag_pipeline.PipelineConfig.from_args(args)

    with pytest.raises(ValueError, match="does not match"):
        metacoag_runner._load_checkpoint(changed_config)

1	import pathlib	8✔
2	import pickle	8✔
3	from types import SimpleNamespace	8✔
4
5	import pytest	8✔
6
7	from click.testing import CliRunner	8✔
8
9	from metacoag import metacoag_pipeline, metacoag_runner	8✔
10	from metacoag.cli import main	8✔
11	from metacoag.metacoag_utils.bidirectionalmap import BidirectionalMap	8✔
12
13
14	__author__ = "Vijini Mallawaarachchi"	8✔
15	__credits__ = ["Vijini Mallawaarachchi"]	8✔
16
17
18	DATADIR = pathlib.Path(__file__).parent / "data"	8✔
19
20
21	@pytest.fixture(scope="session")	8✔
22	def tmp_dir(tmpdir_factory):	8✔
23	return tmpdir_factory.mktemp("tmp")	8✔
24
25
26	@pytest.fixture(autouse=True)	8✔
27	def workingdir(tmp_dir, monkeypatch):	8✔
28	"""set the working directory for all tests"""
29	monkeypatch.chdir(tmp_dir)	8✔
30
31
32	@pytest.fixture(scope="session")	8✔
33	def runner():	8✔
34	"""exportrc works correctly."""
35	return CliRunner()	8✔
36
37
38	def test_metacoag_spades_run(runner, tmp_dir):	8✔
39	outpath = tmp_dir	8✔
40	dir_name = DATADIR / "5G_metaspades"	8✔
41	graph = dir_name / "assembly_graph_with_scaffolds.gfa"	8✔
42	contigs = dir_name / "contigs.fasta"	8✔
43	paths = dir_name / "contigs.paths"	8✔
44	abundance = dir_name / "coverm_mean_coverage.tsv"	8✔
45	args = f"--assembler spades --graph {graph} --contigs {contigs} --paths {paths} --abundance {abundance} --output {outpath}".split()	8✔
46	r = runner.invoke(main, args, catch_exceptions=False)	8✔
47	assert r.exit_code == 0, r.output	8✔
48
49
50	def test_metacoag_megahit_run(runner, tmp_dir):	8✔
51	outpath = tmp_dir	8✔
52	dir_name = DATADIR / "5G_MEGAHIT"	8✔
53	graph = dir_name / "final.gfa"	8✔
54	contigs = dir_name / "final.contigs.fa"	8✔
55	abundance = dir_name / "abundance.tsv"	8✔
56	args = f"--assembler megahit --graph {graph} --contigs {contigs} --abundance {abundance} --output {outpath}".split()	8✔
57	r = runner.invoke(main, args, catch_exceptions=False)	8✔
58	assert r.exit_code == 0, r.output	8✔
59
60
61	def test_continue_option_is_available(runner):	8✔
62	result = runner.invoke(main, ["--help"])	8✔
63
64	assert result.exit_code == 0	8✔
65	assert "--continue" in result.output	8✔
66
67
68	def test_bidirectional_map_can_be_checkpointed(tmp_path):	8✔
69	mapping = BidirectionalMap()	8✔
70	mapping[1] = "contig"	8✔
71	checkpoint_path = tmp_path / "mapping.pickle"	8✔
72
73	with open(checkpoint_path, "wb") as handle:	8✔
74	pickle.dump(mapping, handle)	8✔
75
76	with open(checkpoint_path, "rb") as handle:	8✔
77	restored = pickle.load(handle)	8✔
78
79	assert restored[1] == "contig"	8✔
80	assert restored.inverse["contig"] == 1	8✔
81
82
83	def test_continue_resumes_after_last_completed_stage(tmp_path, monkeypatch):	8✔
84	input_paths = {}	8✔
85	for name in ("graph", "contigs", "abundance", "hmm"):	8✔
86	path = tmp_path / name	8✔
87	path.write_text(name)	8✔
88	input_paths[name] = str(path)	8✔
89
90	args = SimpleNamespace(	8✔
91	assembler="custom",
92	graph=input_paths["graph"],
93	contigs=input_paths["contigs"],
94	abundance=input_paths["abundance"],
95	paths=None,
96	output=str(tmp_path / "output"),
97	hmm=input_paths["hmm"],
98	prefix="resume",
99	min_length=1000,
100	p_intra=0.1,
101	p_inter=0.01,
102	d_limit=20,
103	depth=10,
104	n_mg=108,
105	no_cut_tc=False,
106	mg_threshold=0.5,
107	bin_mg_threshold=0.33333,
108	min_bin_size=200000,
109	delimiter=",",
110	nthreads=2,
111	continue_run=False,
112	)
113	calls = []	8✔
114
115	monkeypatch.setattr(metacoag_pipeline, "log_inputs", lambda *unused: None)	8✔
116	monkeypatch.setattr(	8✔
117	metacoag_pipeline,
118	"load_graph",
119	lambda *unused: (calls.append("graph") or "graph-data", "isolated"),
120	)
121	monkeypatch.setattr(	8✔
122	metacoag_pipeline,
123	"extract_features",
124	lambda *unused: calls.append("features") or "feature-data",
125	)
126	monkeypatch.setattr(	8✔
127	metacoag_pipeline,
128	"parse_markers",
129	lambda *unused: calls.append("markers") or "marker-data",
130	)
131	monkeypatch.setattr(	8✔
132	metacoag_pipeline,
133	"match_seed_bins",
134	lambda *unused: calls.append("seed_bins") or {"seed": True},
135	)
136
137	def interrupt_propagation(*unused):	8✔
138	calls.append("propagated_bins")	8✔
139	raise RuntimeError("interrupted")	8✔
140
141	monkeypatch.setattr(	8✔
142	metacoag_pipeline,
143	"propagate_bins",
144	interrupt_propagation,
145	)
146
147	with pytest.raises(RuntimeError, match="interrupted"):	8✔
148	metacoag_runner.run(args)	8✔
149
150	config = metacoag_pipeline.PipelineConfig.from_args(args)	8✔
151	checkpoint_path = metacoag_runner._checkpoint_path(config)	8✔
152	assert checkpoint_path.is_file()	8✔
153	assert calls == [	8✔
154	"graph",
155	"features",
156	"markers",
157	"seed_bins",
158	"propagated_bins",
159	]
160
161	calls.clear()	8✔
162	args.continue_run = True	8✔
163
164	def completed_stage_called(*unused):	8✔
165	raise AssertionError("a completed stage was rerun")	×
166
167	monkeypatch.setattr(metacoag_pipeline, "load_graph", completed_stage_called)	8✔
168	monkeypatch.setattr(metacoag_pipeline, "extract_features", completed_stage_called)	8✔
169	monkeypatch.setattr(metacoag_pipeline, "parse_markers", completed_stage_called)	8✔
170	monkeypatch.setattr(metacoag_pipeline, "match_seed_bins", completed_stage_called)	8✔
171	monkeypatch.setattr(	8✔
172	metacoag_pipeline,
173	"propagate_bins",
174	lambda *unused: calls.append("propagated_bins") or {"propagated": True},
175	)
176	monkeypatch.setattr(	8✔
177	metacoag_pipeline,
178	"merge_bins",
179	lambda *unused: calls.append("merge_plan") or "merge-plan",
180	)
181	monkeypatch.setattr(	8✔
182	metacoag_pipeline,
183	"write_output",
184	lambda *unused: calls.append("output"),
185	)
186
187	metacoag_runner.run(args)	8✔
188
189	assert calls == ["propagated_bins", "merge_plan", "output"]	8✔
190	assert not checkpoint_path.exists()	8✔
191
192
193	def test_continue_rejects_changed_options(tmp_path):	8✔
194	input_paths = {}	8✔
195	for name in ("graph", "contigs", "abundance", "hmm"):	8✔
196	path = tmp_path / name	8✔
197	path.write_text(name)	8✔
198	input_paths[name] = str(path)	8✔
199
200	args = SimpleNamespace(	8✔
201	assembler="custom",
202	graph=input_paths["graph"],
203	contigs=input_paths["contigs"],
204	abundance=input_paths["abundance"],
205	paths=None,
206	output=str(tmp_path / "output"),
207	hmm=input_paths["hmm"],
208	prefix="",
209	min_length=1000,
210	p_intra=0.1,
211	p_inter=0.01,
212	d_limit=20,
213	depth=10,
214	n_mg=108,
215	no_cut_tc=False,
216	mg_threshold=0.5,
217	bin_mg_threshold=0.33333,
218	min_bin_size=200000,
219	delimiter=",",
220	nthreads=2,
221	continue_run=False,
222	)
223	config = metacoag_pipeline.PipelineConfig.from_args(args)	8✔
224	metacoag_pipeline.validate_inputs(config)	8✔
225	metacoag_runner._save_checkpoint(config, "graph", {"graph_data": "graph"})	8✔
226
227	args.continue_run = True	8✔
228	args.min_length = 2000	8✔
229	changed_config = metacoag_pipeline.PipelineConfig.from_args(args)	8✔
230
231	with pytest.raises(ValueError, match="does not match"):	8✔
232	metacoag_runner._load_checkpoint(changed_config)	8✔

metagentools / MetaCoAG / 27516931519

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous