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
|
"""
Test mapfile generation.
"""
import os
import os.path as op
import re
import subprocess as S
from drslib.drs_tree import DRSTree
from .drs_tree_shared import TestEg, test_dir
from . import gen_drs
from drslib.cmip5 import CMIP5FileSystem
from drslib.mapfile import calc_md5
import io
class TestMapfile(TestEg):
__test__ = True
def setUp(self):
super(TestMapfile, self).setUp()
gen_drs.write_eg1(self.tmpdir)
drs_fs = CMIP5FileSystem(self.tmpdir)
dt = DRSTree(drs_fs)
dt.discover(self.incoming, activity='cmip5',
product='output1', institute='MOHC', model='HadCM3')
self.pt = list(dt.pub_trees.values())[0]
self.pt.do_version()
assert self.pt.state == self.pt.STATE_VERSIONED
def test1(self):
fh = io.StringIO()
self.pt.version_to_mapfile(self.pt.latest, fh, checksum_func=calc_md5)
# Verify checksums match standard md5 tool
fh.seek(0)
for line in fh:
ds_id, path, checksum = re.match(r'([^ ]+) \| ([^ ]+) \|.*checksum=([^ ]+)', line.strip()).groups()
p = S.Popen('md5sum %s' % path, shell=True, stdout=S.PIPE)
output = p.stdout.read()
if not checksum in output:
print ('LINE:', line.strip())
print ('MD5: ', output.strip())
assert False
|
