1#!/usr/local/bin/python3.8
2from __future__ import print_function
3import unittest
4from ruffus import originate, transform, Pipeline, pipeline_run, suffix
5import sys
6
7"""
8    test_softlink_uptodate.py
9"""
10
11
12import os
13tempdir = os.path.relpath(os.path.abspath(os.path.splitext(__file__)[0])) + "/"
14
15# add grandparent to search path for testing
16grandparent_dir = os.path.abspath(
17    os.path.join(os.path.dirname(__file__), "..", ".."))
18sys.path.insert(0, grandparent_dir)
19
20# module name = script name without extension
21module_name = os.path.splitext(os.path.basename(__file__))[0]
22
23
24# funky code to import by file name
25parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
26
27
28# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
29
30#   Tasks
31
32
33# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
34
35#
36#   First task
37#
38@originate(["a.1", "b.1"])
39def start_task(output_file_name):
40    with open(output_file_name,  "w") as f:
41        pass
42
43#
44#   Forwards file names, is always as up to date as its input files...
45#
46
47
48@transform(start_task, suffix(".1"), ".1")
49def same_file_name_task(input_file_name, output_file_name):
50    pass
51
52#
53#   Links file names, is always as up to date if links are not missing
54#
55
56
57@transform(start_task, suffix(".1"), ".linked.1")
58def linked_file_name_task(input_file_name, output_file_name):
59    try:
60        os.symlink(input_file_name, output_file_name)
61    except:
62        print(input_file_name, output_file_name)
63        raise
64
65
66#
67#   Final task linking everything
68#
69@transform([linked_file_name_task, same_file_name_task], suffix(".1"), ".3")
70def final_task(input_file_name, output_file_name):
71    with open(output_file_name,  "w") as f:
72        pass
73
74# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
75
76#   Run pipeline
77
78
79# 88888888888888888888888888888888888888888888888888888888888888888888888888888888888888888
80
81
82try:
83    from StringIO import StringIO
84except:
85    from io import StringIO
86
87
88class Test_ruffus(unittest.TestCase):
89    def setUp(self):
90        for f in ["a.1", "b.1", "a.linked.1", "b.linked.1", "a.3", "b.3", "a.linked.3", "b.linked.3"]:
91            try:
92                if os.path.exists(f):
93                    os.unlink(f)
94            except:
95                print("    !!!!OOPs. Can't unlink %s" % f, file=sys.stderr)
96                pass
97
98    def tearDown(self):
99        for f in ["a.1", "b.1", "a.linked.1", "b.linked.1", "a.3", "b.3", "a.linked.3", "b.linked.3"]:
100            if os.path.lexists(f):
101                os.unlink(f)
102            else:
103                raise Exception("Expected %s missing" % f)
104
105    def test_ruffus(self):
106        pipeline_run(log_exceptions=True, verbose=0, pipeline="main")
107
108    def test_newstyle_ruffus(self):
109        test_pipeline = Pipeline("test")
110        test_pipeline.originate(start_task, ["a.1", "b.1"])
111        test_pipeline.transform(same_file_name_task,
112                                start_task, suffix(".1"), ".1")
113        test_pipeline.transform(linked_file_name_task,
114                                start_task, suffix(".1"), ".linked.1")
115        test_pipeline.transform(
116            final_task, [linked_file_name_task, same_file_name_task], suffix(".1"), ".3")
117        test_pipeline.run(log_exceptions=True, verbose=0)
118
119
120if __name__ == '__main__':
121    unittest.main()
122