File: test_subgraph.py

package info (click to toggle)
cwltool 1.0.20181217162649%2Bdfsg-10
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 2,632 kB
  • sloc: python: 11,008; makefile: 153; sh: 22
file content (40 lines) | stat: -rw-r--r-- 1,403 bytes parent folder | download
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
import os

from six.moves import urllib

import pytest
import json

from cwltool.context import LoadingContext
from cwltool.load_tool import load_tool
from cwltool.workflow import default_make_tool
from cwltool.subgraph import get_subgraph
from schema_salad.utils import convert_to_dict
from cwltool.resolver import Path, resolve_local
from .util import get_data, working_directory
from six import string_types
from .test_fetch import norm

def test_get_subgraph():
    loadingContext = LoadingContext({"construct_tool_object": default_make_tool})
    wf = norm(Path(get_data("tests/subgraph/count-lines1-wf.cwl")).as_uri())
    tool = load_tool(wf, loadingContext)

    sg = norm(Path(get_data("tests/subgraph")).as_uri())

    def clean(val):
        if isinstance(val, string_types):
            if val.startswith(sg):
                return val[len(sg)+1:]
        if isinstance(val, dict):
            return {k: clean(v) for k,v in val.items()}
        if isinstance(val, list):
            return [clean(v) for v in val]
        return val

    for a in ("file1", "file2", "file3", "count_output",
              "output3", "output4", "output5",
              "step1", "step2", "step3", "step4", "step5"):
        extracted = get_subgraph([wf+"#"+a], tool)
        with open(get_data("tests/subgraph/extract_"+a+".json")) as f:
            assert json.load(f) == clean(convert_to_dict(extracted))