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
|
"""testing the default import"""
import os
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import TestCase
import pytest
from cogent3 import app_help, available_apps, get_app, open_data_store
from cogent3.app.composable import LOADER, WRITER, is_app
from cogent3.util.table import Table
def _get_all_composables(tmp_dir_name):
tmp_dir_name = Path(tmp_dir_name)
test_model1 = get_app("model", "HKY85")
test_model2 = get_app("model", "GN")
test_hyp = get_app("hypothesis", test_model1, test_model2)
test_num_reps = 100
return [
get_app(
"align_to_ref",
),
get_app("progressive_align", model="GY94"),
get_app("fast_slow_dist", moltype="dna", fast_calc="hamming"),
get_app("ancestral_states"),
get_app("bootstrap", hyp=test_hyp, num_reps=test_num_reps),
get_app("hypothesis", test_model1, test_model2),
get_app("model", "GN"),
get_app("tabulate_stats"),
get_app("fixed_length", 100),
get_app("sample.min_length", 100),
get_app("write_db", open_data_store(tmp_dir_name / "delme.sqlitedb", mode="w")),
get_app(
"write_json",
open_data_store(tmp_dir_name / "json", suffix="json", mode="w"),
),
get_app(
"write_seqs",
open_data_store(tmp_dir_name / "fasta", suffix="fasta", mode="w"),
),
get_app("omit_bad_seqs"),
get_app("omit_degenerates"),
get_app("omit_duplicated"),
get_app("take_codon_positions", 1),
get_app("take_named_seqs"),
get_app("take_n_seqs", 2),
get_app("trim_stop_codons", gc=1),
get_app("select_translatable"),
get_app("quick_tree"),
get_app("scale_branches"),
get_app("uniformize_tree"),
]
class TestAvailableApps(TestCase):
def test_available_apps(self):
"""available_apps returns a table"""
from cogent3.util.table import Table
apps = available_apps()
self.assertIsInstance(apps, Table)
self.assertTrue(apps.shape[0] > 10)
def test_composable_pairwise_applications(self):
"""Properly compose two composable applications"""
with TemporaryDirectory(dir=".") as dirname:
applications = _get_all_composables(os.path.join(dirname, "delme"))
for app in applications:
self.assertTrue(is_app(app), msg=app)
composable_application_tuples = [
(app1, app2)
for app1 in applications
for app2 in applications
if app1 != app2
and (
app1._return_types & app2._data_types
or app1._return_types & {"SerialisableType", "IdentifierType"}
)
and app1.app_type is not WRITER
and app2.app_type is not LOADER
]
for app_a, app_b in composable_application_tuples:
app_a.disconnect()
app_b.disconnect()
# Compose two composable applications, there should not be exceptions.
app_a + app_b
def test_incompatible_pairwise_applications(self):
"""Properly identify two incompatible applications"""
with TemporaryDirectory(dir=".") as dirname:
applications = _get_all_composables(os.path.join(dirname, "delme"))
for app in applications:
self.assertTrue(is_app(app))
incompatible_application_tuples = [
(app1, app2)
for app1 in applications
for app2 in applications
if app1.app_type is WRITER
or app2.app_type is LOADER
and app1 != app2
and not app1._return_types & app2._data_types
and not app1._return_types & {"SerialisableType", "IdentifierType"}
]
for app_a, app_b in incompatible_application_tuples:
err_type = ValueError if app_a is app_b else TypeError
app_a.disconnect()
app_b.disconnect()
# Compose two incompatible applications, there should be exceptions.
with self.assertRaises(err_type):
app_a + app_b
@pytest.mark.parametrize("name", ("sample.min_length", "min_length"))
def test_get_app(name):
app = get_app(name, 500)
assert app.__class__.__name__.endswith(name.split(".")[-1])
def test_get_app_kwargs():
# when an app has a name kwarg
# we should still be able to use get_app!
_ = get_app("model", "F81", name="F81-model")
def test_app_help(capsys):
app_help("concat")
got = capsys.readouterr().out
assert "Options" in got
assert got.count("SerialisableType") == 1 # output type
@pytest.mark.parametrize(
"app_name", ("bootstrap", "from_primitive", "load_db", "take_named_seqs")[:1]
)
def test_app_help_signature(capsys, app_name):
from cogent3.app import _get_app_matching_name, _make_signature
with pytest.raises(ValueError, match="app cannot be None"):
_make_signature(None)
got = _make_signature(_get_app_matching_name(app_name))
# app name is in quotes
assert f'"{app_name}"' in got
# check split across multiple lines if long signature
if len(got) > 70:
assert got.count("\n") > 1
def test_available_apps_filter():
"""available apps can be filtered by name"""
app_name_filter: str = "load"
filtered_apps = available_apps(app_name_filter)
assert isinstance(filtered_apps, Table)
assert len(filtered_apps) > 0
# check every returned table row 'name' has filter in it
assert sum(app_name_filter in n for n in filtered_apps.columns["name"]) == len(
filtered_apps
)
|
