forked from snakemake/snakemake
/
testapi.py
119 lines (101 loc) · 3.23 KB
/
testapi.py
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
"""
Tests for Snakemake’s API
"""
from snakemake import snakemake
import asyncio
import sys
import tempfile
import os.path
from textwrap import dedent
def test_keep_logger():
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "Snakefile")
with open(path, "w") as f:
print("rule:\n output: 'result.txt'\n shell: 'touch {output}'", file=f)
snakemake(path, workdir=tmpdir, keep_logger=True)
def test_workflow_calling():
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "Snakefile")
with open(path, "w") as f:
print(
dedent(
"""
rule:
output: 'result.txt'
run:
with open(output[0], 'w') as f:
print("hello", file=f)
"""
),
file=f,
)
workflow = Workflow(
snakefile=snakefile,
overwrite_workdir=tmpdir,
)
def test_run_script_directive():
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "Snakefile")
with open(path, "w") as f:
print(
dedent(
"""
rule:
output: 'result.txt'
run:
with open(output[0], 'w') as f:
print("hello", file=f)
"""
),
file=f,
)
snakemake(path, workdir=tmpdir)
def test_run_script_directive_async():
"""Tests :func`snakemake.common.async_run`. The test ensures the ability to
execute Snakemake API even if an asyncio event loop is already running.
"""
import tracemalloc
from snakemake.common import async_run
tracemalloc.start()
async def dummy_task():
await asyncio.sleep(0.00001)
async def main():
async_run(dummy_task())
test_run_script_directive()
asyncio.run(main())
def test_dicts_in_config():
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "Snakefile")
with open(path, "w") as f:
print(
dedent(
"""
rule:
output: 'result.txt'
run:
with open(output[0], 'w') as f:
print("hello, this option " + config["this_option"] + "; this test dictionary " + config["test"]["this_dict"], file=f)
"""
),
file=f,
)
snakemake(
path,
workdir=tmpdir,
config={
"this_option": "does_not_break",
"test": {"this_dict": "shoult_not_either"},
},
)
def test_lockexception():
from snakemake.persistence import Persistence
from snakemake.exceptions import LockException
persistence = Persistence()
persistence.all_inputfiles = lambda: ["A.txt"]
persistence.all_outputfiles = lambda: ["B.txt"]
persistence.lock()
try:
persistence.lock()
except LockException as e:
return True
assert False