|
1
|
|
|
import unittest, os, shutil, sys |
|
2
|
|
|
from cStringIO import StringIO |
|
3
|
|
|
absp = os.path.abspath |
|
4
|
|
|
|
|
5
|
|
|
|
|
6
|
|
|
class TerminalApiTests(unittest.TestCase): |
|
7
|
|
|
|
|
8
|
|
|
def setUp(self): |
|
9
|
|
|
self.dbpath = os.path.expanduser(os.path.join('~','provenance_test.json')) |
|
10
|
|
|
if os.path.exists(self.dbpath): |
|
11
|
|
|
os.remove(self.dbpath) |
|
12
|
|
|
os.mkdir('temp') |
|
13
|
|
|
from niprov import ProvenanceContext |
|
14
|
|
|
self.provenance = ProvenanceContext() |
|
15
|
|
|
self.provenance.config.database_type = 'file' |
|
16
|
|
|
self.provenance.config.database_url = self.dbpath |
|
17
|
|
|
self.provenance.config.verbose = False |
|
18
|
|
|
self.oldstdout = sys.stdout |
|
19
|
|
|
sys.stdout = self.stdout = StringIO() |
|
20
|
|
|
|
|
21
|
|
|
def tearDown(self): |
|
22
|
|
|
sys.stdout = self.oldstdout |
|
23
|
|
|
self.stdout.close() |
|
24
|
|
|
shutil.rmtree('temp') |
|
25
|
|
|
|
|
26
|
|
|
@unittest.skip("Doesn't work on travis") |
|
27
|
|
|
def test_Report_pipeline(self): |
|
28
|
|
|
self.provenance.discover('testdata') |
|
29
|
|
|
raw = os.path.abspath('testdata/eeg/stub.cnt') |
|
30
|
|
|
self.provenance.log(absp('p1a.f'), 'test', raw, transient=True) |
|
31
|
|
|
self.provenance.log(absp('p1b.f'), 'test', raw, transient=True) |
|
32
|
|
|
self.provenance.log(absp('p2.f'), 'test', absp('p1a.f'), transient=True) |
|
33
|
|
|
self.provenance.export(None, 'stdout', 'simple', pipeline=True) |
|
34
|
|
|
exp = '' |
|
35
|
|
|
exp += '+---stub.cnt\n' |
|
36
|
|
|
exp += '| +---p1a.f\n' |
|
37
|
|
|
exp += '| | +---p2.f\n' |
|
38
|
|
|
#exp += '| +---p1b.f\n' # p1b is not related to p2f |
|
39
|
|
|
import time |
|
40
|
|
|
time.sleep(0.3) |
|
41
|
|
|
self.assertIn(exp, self.stdout.getvalue()) |
|
42
|
|
|
|
|
43
|
|
|
|
|
44
|
|
|
if __name__ == '__main__': |
|
45
|
|
|
unittest.main() |
|
46
|
|
|
|
|
47
|
|
|
|
ilogue /
niprov
