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
|
#!/usr/bin/env python
from __future__ import print_function
"""
test_pausing.py
test time.sleep keeping input files and output file times correct
"""
import os
import sys
# add grandparent to search path for testing
grandparent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
sys.path.insert(0, grandparent_dir)
# module name = script name without extension
module_name = os.path.splitext(os.path.basename(__file__))[0]
# funky code to import by file name
parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
import ruffus
from ruffus import pipeline_run, pipeline_printout, Pipeline, files
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
# imports
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
import re
import operator
try:
import StringIO as io
except:
import io as io
from collections import defaultdict
import random
# use simplejson in place of json for python < 2.6
try:
import json
except ImportError:
import simplejson
json = simplejson
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
# Main logic
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
# Tasks
#88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
a = []
#
# task1
#
@files(a)
def task1():
"""
First task
"""
print("Task1", file=sys.stderr)
import unittest
class t_save_to_str_logger:
"""
Everything to stderr
"""
def __init__ (self):
self.info_str = ""
self.warning_str = ""
self.debug_str = ""
def info (self, message):
self.info_str += message
def warning (self, message):
self.warning_str += message
def debug (self, message):
self.debug_str += message
class Test_task(unittest.TestCase):
def test_task (self):
save_to_str_logger = t_save_to_str_logger()
pipeline_run(multiprocess = 10,
logger = save_to_str_logger,
verbose = 1,
pipeline= "main")
self.assertTrue("@files() was empty" in save_to_str_logger.warning_str)
print("\n Warning printed out correctly", file=sys.stderr)
def test_newstyle_task (self):
test_pipeline = Pipeline("test")
test_pipeline.files(task1, a)
save_to_str_logger = t_save_to_str_logger()
test_pipeline.run(multiprocess = 10,
logger = save_to_str_logger,
verbose = 1)
self.assertTrue("@files() was empty" in save_to_str_logger.warning_str)
print("\n Warning printed out correctly", file=sys.stderr)
#
# Necessary to protect the "entry point" of the program under windows.
# see: http://docs.python.org/library/multiprocessing.html#multiprocessing-programming
#
if __name__ == '__main__':
unittest.main()
|
