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
|
"""
Tests for Snakemake’s API
"""
from snakemake import snakemake
import asyncio
import sys
import tempfile
import os.path
from textwrap import dedent
import pytest
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)
@pytest.mark.skip(reason="This test is known to fail on Debian builds")
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
|
