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
50
51
52
53
54
55
56
57
58
59
|
import json
import os
import logging
import unittest
from pbcommand.models import ToolContract, ResolvedToolContract
from pbcommand.pb_io import (load_tool_contract_from,
load_resolved_tool_contract_from)
from pbcommand.schemas import validate_rtc, validate_tc
from base_utils import DATA_DIR
log = logging.getLogger(__name__)
def _to_json(path):
with open(path, 'r') as f:
d = json.loads(f.read())
return d
def _filter_rtc(path):
return path.endswith('resolved_tool_contract.json')
def _filter_tc(path):
return path.endswith('tool_contract.json') and not path.endswith('resolved_tool_contract.json')
def _get_all_from(root_dir, filter_func):
for path in os.listdir(root_dir):
if filter_func(path):
yield os.path.join(root_dir, path)
def _to_assertion(path, schema_validate_func):
def test_is_validate(self):
d = _to_json(path)
log.debug(d)
is_valid = schema_validate_func(d)
log.info(" is-valid? {i} {p}".format(i=is_valid, p=path))
self.assertTrue(is_valid, "{p} is not valid with the avro schema".format(p=path))
return test_is_validate
class ValidateResolvedToolContracts(unittest.TestCase):
def test_validate_resolved_tool_contracts(self):
for path in _get_all_from(DATA_DIR, _filter_rtc):
f = _to_assertion(path, validate_rtc)
f(self)
self.assertIsInstance(load_resolved_tool_contract_from(path), ResolvedToolContract)
class ValidateToolContracts(unittest.TestCase):
def test_validate_tool_contracts(self):
for path in _get_all_from(DATA_DIR, _filter_tc):
f = _to_assertion(path, validate_tc)
f(self)
self.assertIsInstance(load_tool_contract_from(path), ToolContract)
|
