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
|
import subprocess
import os
import errno
import shutil
import time
import signal
__test_files__ = ["sample_data/albacore_1d_original.fast5", "sample_data/metrichor_2d_original.fast5"]
__test_runs__ = ["lossless", "deep-lossless"]
__prefix__ = "picopore.test"
__raw_runs__ = [["--fastq","--summary"],["--summary","--no-fastq"],["--fastq","--no-summary"],["--no-fastq","--no-summary"], ["--manual", "Analyses"]]
def mkdir(path):
try:
os.makedirs(path)
except OSError as e:
if e.errno == errno.EEXIST and os.path.isdir(path):
pass
else:
raise
def call(additionalArgs, prefix=None):
args=["python","-m","picopore","-y","--print-every","1"]
if prefix is not None:
args.extend(["--prefix",prefix])
args.extend(additionalArgs)
print(" ".join(args))
p = subprocess.call(args)
if prefix is not None:
filename = args[-1]
dirname = os.path.dirname(filename)
basename = os.path.basename(filename)
os.remove(os.path.join(dirname,".".join([prefix,basename])))
return p
def testFile(filename):
result = 0
for run in __test_runs__:
result += call(["--test","--mode",run, filename])
result += call(["--mode",run, filename], prefix=__prefix__)
for run in __raw_runs__:
result += call(["--mode","raw"] + run + [filename], prefix=__prefix__)
return result
def testRealtime(mode, additionalArgs=None, directory="realtime"):
__waittime = 10
mkdir(directory)
args = ["python","-m","picopore","-y","--realtime","--print-every","1"]
if additionalArgs is not None:
args.extend(additionalArgs)
args.extend(["--mode",mode,directory])
print(" ".join(args))
p = subprocess.Popen(args)
time.sleep(__waittime)
for filename in __test_files__:
shutil.copy(filename, directory)
time.sleep(__waittime)
p.send_signal(signal.SIGINT)
p.wait()
shutil.rmtree(directory)
return p.returncode
exitcode = 0
for filename in __test_files__:
exitcode += testFile(filename)
for mode in __test_runs__:
exitcode += testRealtime(mode)
for mode in __raw_runs__:
exitcode += testRealtime("raw", additionalArgs=mode)
exit(exitcode)
|
