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