|
| 1 | +import io |
| 2 | +import unittest |
| 3 | +from collections import defaultdict |
| 4 | +from unittest import mock |
| 5 | + |
| 6 | +from gff3tool.lib.gff3_merge import merge |
| 7 | + |
| 8 | + |
| 9 | +class FakeGroups: |
| 10 | + def __init__(self, **_kwargs): |
| 11 | + self.mapName2ID = {} |
| 12 | + self.info = [] |
| 13 | + self.mapType2Log = { |
| 14 | + "other": "OTHER", |
| 15 | + "Delete": "DELETE", |
| 16 | + "simple": "SIMPLE", |
| 17 | + "multi-ref": "MULTI", |
| 18 | + } |
| 19 | + self.id2name = {} |
| 20 | + pgff = _kwargs.get("Pgff") |
| 21 | + if pgff is not None: |
| 22 | + for line in pgff.lines: |
| 23 | + line_id = line.get("attributes", {}).get("ID") |
| 24 | + if line_id: |
| 25 | + self.id2name[line_id] = line.get("attributes", {}).get("Name", line_id) |
| 26 | + |
| 27 | + def replacer(self, *_args, **_kwargs): |
| 28 | + root = _args[0] |
| 29 | + for child in root.get("children", []): |
| 30 | + child.setdefault("attributes", {})["replace_type"] = "other" |
| 31 | + return None |
| 32 | + |
| 33 | + def replacer_multi(self, *_args, **_kwargs): |
| 34 | + return "ok" |
| 35 | + |
| 36 | + def name2id(self, *_args, **_kwargs): |
| 37 | + return None |
| 38 | + |
| 39 | + |
| 40 | +class FakeGff: |
| 41 | + def __init__(self, lines): |
| 42 | + self.lines = lines |
| 43 | + self.features = defaultdict(list) |
| 44 | + for line in lines: |
| 45 | + line_id = line.get("attributes", {}).get("ID") |
| 46 | + if line_id: |
| 47 | + self.features[line_id].append(line) |
| 48 | + self.written_output = None |
| 49 | + |
| 50 | + def collect_roots(self, line): |
| 51 | + return [line] |
| 52 | + |
| 53 | + def collect_descendants(self, line): |
| 54 | + return line.get("children", []) |
| 55 | + |
| 56 | + def write(self, output_gff): |
| 57 | + self.written_output = output_gff |
| 58 | + |
| 59 | + |
| 60 | +class TestMergeMain(unittest.TestCase): |
| 61 | + def _make_root_with_child(self, root_id, child_status="active", child_replace=None): |
| 62 | + if child_replace is None: |
| 63 | + child_replace = ["NA"] |
| 64 | + |
| 65 | + child = { |
| 66 | + "line_type": "feature", |
| 67 | + "type": "mRNA", |
| 68 | + "line_status": "removed" if child_status == "removed" else "active", |
| 69 | + "line_raw": f"raw-{root_id}-child", |
| 70 | + "attributes": { |
| 71 | + "ID": f"{root_id}-RA", |
| 72 | + "replace": list(child_replace), |
| 73 | + }, |
| 74 | + "parents": [], |
| 75 | + "children": [], |
| 76 | + } |
| 77 | + if child_status == "removed": |
| 78 | + child["attributes"]["status"] = "Delete" |
| 79 | + |
| 80 | + root = { |
| 81 | + "line_type": "feature", |
| 82 | + "type": "gene", |
| 83 | + "attributes": {"ID": root_id}, |
| 84 | + "children": [child], |
| 85 | + } |
| 86 | + return root, child |
| 87 | + |
| 88 | + def test_delete_with_na_replace_raises_system_exit(self): |
| 89 | + wa_root, wa_child = self._make_root_with_child("gene1", child_status="active", child_replace=["NA"]) |
| 90 | + other_root, other_child = self._make_root_with_child("gene1", child_status="removed", child_replace=["NA"]) |
| 91 | + |
| 92 | + wa_gff = FakeGff([wa_root, wa_child]) |
| 93 | + other_gff = FakeGff([other_root, other_child]) |
| 94 | + |
| 95 | + def fake_gff_factory(gff_file=None, logger=None): |
| 96 | + if gff_file == "WA_sorted.gff": |
| 97 | + return wa_gff |
| 98 | + if gff_file == "other_sorted.gff": |
| 99 | + return other_gff |
| 100 | + raise AssertionError(f"Unexpected gff file: {gff_file}") |
| 101 | + |
| 102 | + with mock.patch.object(merge.gff3_sort, "main", autospec=True), \ |
| 103 | + mock.patch.object(merge.replace_OGS, "Groups", FakeGroups), \ |
| 104 | + mock.patch.object(merge, "Gff3", side_effect=fake_gff_factory), \ |
| 105 | + mock.patch.object(merge, "remove_files_from_list", autospec=True): |
| 106 | + with self.assertRaises(SystemExit) as cm: |
| 107 | + merge.main( |
| 108 | + gff_file1="wa.gff3", |
| 109 | + gff_file2="other.gff3", |
| 110 | + output_gff="out.gff3", |
| 111 | + report_fh=io.StringIO(), |
| 112 | + ) |
| 113 | + |
| 114 | + self.assertIn("replace tag for Delete replacement cannot be NA", str(cm.exception)) |
| 115 | + |
| 116 | + def test_main_writes_output_and_cleans_temp_files(self): |
| 117 | + wa_root, wa_child = self._make_root_with_child("geneX", child_status="active", child_replace=["NA"]) |
| 118 | + other_root, other_child = self._make_root_with_child("geneY", child_status="active", child_replace=["NA"]) |
| 119 | + |
| 120 | + wa_gff = FakeGff([wa_root, wa_child]) |
| 121 | + other_gff = FakeGff([other_root, other_child]) |
| 122 | + |
| 123 | + def fake_gff_factory(gff_file=None, logger=None): |
| 124 | + if gff_file == "WA_sorted.gff": |
| 125 | + return wa_gff |
| 126 | + if gff_file == "other_sorted.gff": |
| 127 | + return other_gff |
| 128 | + raise AssertionError(f"Unexpected gff file: {gff_file}") |
| 129 | + |
| 130 | + report = io.StringIO() |
| 131 | + |
| 132 | + with mock.patch.object(merge.gff3_sort, "main", autospec=True), \ |
| 133 | + mock.patch.object(merge.replace_OGS, "Groups", FakeGroups), \ |
| 134 | + mock.patch.object(merge, "Gff3", side_effect=fake_gff_factory), \ |
| 135 | + mock.patch.object(merge, "remove_files_from_list", autospec=True) as rm_files: |
| 136 | + merge.main( |
| 137 | + gff_file1="wa.gff3", |
| 138 | + gff_file2="other.gff3", |
| 139 | + output_gff="final.gff3", |
| 140 | + report_fh=report, |
| 141 | + ) |
| 142 | + |
| 143 | + self.assertEqual(other_gff.written_output, "final.gff3") |
| 144 | + rm_files.assert_called_once_with(["WA_sorted.gff", "other_sorted.gff"]) |
| 145 | + report_output = report.getvalue() |
| 146 | + self.assertIn("# Number of WA loci", report_output) |
| 147 | + self.assertIn("Change_log", report_output) |
| 148 | + |
| 149 | + |
| 150 | +if __name__ == "__main__": |
| 151 | + unittest.main() |
0 commit comments