From f655b1195d2b2904b8d69dbd963970b4a72783d5 Mon Sep 17 00:00:00 2001 From: Diane Trout Date: Wed, 14 Jan 2009 01:18:42 +0000 Subject: [PATCH] add some testing code for the runner daemon --- htsworkflow/automation/test/test_runner.py | 46 ++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 htsworkflow/automation/test/test_runner.py diff --git a/htsworkflow/automation/test/test_runner.py b/htsworkflow/automation/test/test_runner.py new file mode 100644 index 0000000..6c3b9df --- /dev/null +++ b/htsworkflow/automation/test/test_runner.py @@ -0,0 +1,46 @@ +import unittest + + +import os +from htsworkflow.automation.copier import runfolder_validate + +def extract_runfolder_path(watchdir, event): + runfolder_path = watchdir + path = event.path + if not path.startswith(watchdir): + return None + + fragments = path[len(watchdir):].split(os.path.sep) + for f in fragments: + runfolder_path = os.path.join(runfolder_path, f) + if runfolder_validate(f): + return runfolder_path + return None + +class Event(object): + def __init__(self, path=None, name=None): + self.path = path + self.name = name + +class testRunner(unittest.TestCase): + + def test_extract_runfolder(self): + watchdir = os.path.join('root', 'server', 'mount') + runfolder = os.path.join(watchdir, '080909_HWI-EAS229_0052_1234ABCD') + ipar = os.path.join(runfolder, 'Data', 'IPAR_1.01') + other = os.path.join(watchdir, 'other') + + event = Event( path=runfolder ) + self.failUnlessEqual(extract_runfolder_path(watchdir, event), runfolder) + + event = Event( path=ipar ) + self.failUnlessEqual(extract_runfolder_path(watchdir, event), runfolder) + + event = Event( path=other) + self.failUnlessEqual(extract_runfolder_path(watchdir, event), None ) + +def suite(): + return unittest.makeSuite(testRunner,'test') + +if __name__ == "__main__": + unittest.main(defaultTest="suite") -- 2.30.2