Declarative YAML test coverage for gxformat2.cytoscape

HISTORY.rst

@@ -12,6 +12,9 @@ History
 * Declarative YAML-driven test coverage for ``gxformat2.mermaid``; fix
   ``_input_type_str`` to accept string input types (previous ``.value``
   attribute access crashed on real workflows).
+* Declarative YAML-driven test coverage for ``gxformat2.cytoscape``; stop
+  ``test_interop_generation`` from leaking ``tmp*.gxwf.*`` files into
+  ``tests/examples/cytoscape/`` on each run.
 
 
 ---------------------

gxformat2/examples/expectations/cytoscape.yml

@@ -0,0 +1,144 @@
+# Cytoscape.js element generation from Galaxy workflows.
+# cytoscape_elements_to_list returns the flat list-of-dicts cytoscape.js consumes:
+#   [{group: "nodes"|"edges", data: {...}, classes?: [...], position?: {x,y}}, ...]
+# Inputs come first (in declaration order), then steps, then edges (one per resolved source).
+# cytoscape_node_ids / cytoscape_edge_ids are list-of-string conveniences for
+# value_set and value_any_contains assertions.
+
+test_simple_graph_emits_two_nodes_one_edge:
+  fixture: synthetic-graph-simple.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [$length]
+      value: 3
+    - path: [0, "group"]
+      value: "nodes"
+    - path: [2, "group"]
+      value: "edges"
+
+test_simple_graph_input_node_shape:
+  fixture: synthetic-graph-simple.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [0, "data", "id"]
+      value: "the_input"
+    - path: [0, "data", "step_type"]
+      value: "data"
+    - path: [0, "data", "tool_id"]
+      value: null
+    - path: [0, "classes"]
+      value_set: ["type_data", "input"]
+
+test_simple_graph_tool_step_shape:
+  fixture: synthetic-graph-simple.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [1, "data", "id"]
+      value: "cat"
+    - path: [1, "data", "tool_id"]
+      value: "cat1"
+    - path: [1, "data", "step_type"]
+      value: "tool"
+    - path: [1, "classes"]
+      value_set: ["type_tool", "runnable"]
+
+test_simple_graph_edge_shape:
+  fixture: synthetic-graph-simple.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [2, "data", "id"]
+      value: "cat__input1__from__the_input"
+    - path: [2, "data", "source"]
+      value: "the_input"
+    - path: [2, "data", "target"]
+      value: "cat"
+    - path: [2, "data", "input"]
+      value: "input1"
+    - path: [2, "data", "output"]
+      value: null
+
+test_int_input_step_type:
+  fixture: synthetic-int-input.gxwf.yml
+  operation: cytoscape_node_ids
+  assertions:
+    - path: []
+      value_set: ["input_d", "num_lines", "random_lines"]
+
+test_int_input_carries_int_step_type:
+  fixture: synthetic-int-input.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [1, "data", "id"]
+      value: "num_lines"
+    - path: [1, "data", "step_type"]
+      value: "int"
+    - path: [1, "classes"]
+      value_set: ["type_int", "input"]
+
+test_subworkflow_step_classes:
+  fixture: synthetic-graph-with-subworkflow.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    - path: [2, "data", "id"]
+      value: "nested_workflow"
+    - path: [2, "data", "step_type"]
+      value: "subworkflow"
+    - path: [2, "classes"]
+      value_set: ["type_subworkflow", "runnable"]
+
+test_subworkflow_edges:
+  fixture: synthetic-graph-with-subworkflow.gxwf.yml
+  operation: cytoscape_edge_ids
+  assertions:
+    - path: []
+      value_set:
+        - "first_cat__input1__from__outer_input"
+        - "nested_workflow__inner_input__from__first_cat"
+
+test_subworkflow_edge_named_output:
+  fixture: synthetic-graph-with-subworkflow.gxwf.yml
+  operation: cytoscape_elements_to_list
+  assertions:
+    # Edge from first_cat (named output 'out_file1') to nested_workflow.
+    - path: [4, "data", "id"]
+      value: "nested_workflow__inner_input__from__first_cat"
+    - path: [4, "data", "output"]
+      value: "out_file1"
+
+test_multi_data_inputs_distinct_ids:
+  fixture: synthetic-multi-data-input.gxwf.yml
+  operation: cytoscape_node_ids
+  assertions:
+    - path: []
+      value_set: ["optional", "required", "count_multi_file"]
+
+test_multi_data_inputs_two_edges:
+  fixture: synthetic-multi-data-input.gxwf.yml
+  operation: cytoscape_edge_ids
+  assertions:
+    - path: [$length]
+      value: 2
+    - path: []
+      value_any_contains: "from__required"
+    - path: []
+      value_any_contains: "from__optional"
+
+test_real_hacked_unicycler_repo_link_populated:
+  fixture: real-hacked-unicycler-assembly-extra-annotations.ga
+  operation: cytoscape_elements_to_list
+  assertions:
+    # First tool step in this workflow is fastQC; its tool_shed_repository should
+    # produce a tool-shed view URL on the corresponding cytoscape node.
+    - path: [3, "data", "id"]
+      value: "fastQC"
+    - path: [3, "data", "repo_link"]
+      value_matches: "^https://toolshed\\.g2\\.bx\\.psu\\.edu/view/devteam/fastqc/"
+
+test_real_sars_cov2_smoke:
+  fixture: real-sars-cov2-variant-calling.ga
+  operation: cytoscape_node_ids
+  assertions:
+    - path: []
+      value_any_contains: "Accessions file"
+    - path: []
+      value_any_contains: "SnpSift"

tests/test_cytoscape.py

@@ -1,11 +1,19 @@
+"""CLI and HTML-render smoke tests for gxformat2.cytoscape.
+
+Behavioral coverage of cytoscape_elements lives in declarative YAML
+(gxformat2/examples/expectations/cytoscape.yml) and runs from
+test_interop_tests.py — keep this file for things that aren't a pure
+function of (workflow dict) -> result.
+"""
+
 import json
 import os
 import shutil
 import tempfile
 
-from gxformat2.cytoscape import cytoscape_elements, CytoscapeElements, main
+from gxformat2.cytoscape import cytoscape_elements, main, render_html
 from gxformat2.cytoscape._builder import _input_type_str
-from gxformat2.normalized import ensure_format2, normalized_format2
+from gxformat2.normalized import normalized_format2
 from gxformat2.yaml import ordered_load
 
 from ._helpers import example_path, TEST_INTEROP_EXAMPLES
@@ -31,57 +39,46 @@ def test_main_output_html():
     assert "</body>" in open(out_file.name).read()
 
 
-def test_interop_generation():
-    # not much of a test case but it will generate a directory of interoperability examples to
-    # test Java against.
-    write_cytoscape_elements(EXAMPLE_PATH)
-    write_cytoscape_elements_for_string(WITH_REPORT)
-
-
-def test_cytoscape_elements_returns_typed_model():
-    elements = cytoscape_elements(EXAMPLE_PATH)
-    assert isinstance(elements, CytoscapeElements)
-    assert len(elements.nodes) > 0
-    assert len(elements.edges) > 0
-    # to_list round-trips to the flat format cytoscape.js expects
-    flat = elements.to_list()
-    assert isinstance(flat, list)
-    assert all(e["group"] in ("nodes", "edges") for e in flat)
-
-
-def test_cytoscape_elements_from_nf2():
-    nf2 = ensure_format2(EXAMPLE_PATH)
-    elements = cytoscape_elements(nf2)
-    assert isinstance(elements, CytoscapeElements)
-    assert len(elements.nodes) == len(nf2.inputs) + len(nf2.steps)
-
-
 def test_render_html():
-    from gxformat2.cytoscape import render_html
-
     elements = cytoscape_elements(EXAMPLE_PATH)
     html = render_html(elements)
     assert "</body>" in html
     assert "cytoscape" in html
 
 
 def test_multi_string_input_type():
+    # Non-function-of-(wf_dict) helper: exercises the typed input directly,
+    # which the declarative harness can't reach.
     nf2 = normalized_format2(ordered_load(MULTI_STRING_INPUT_WORKFLOW))
     inp = nf2.inputs[0]
     assert _input_type_str(inp) == "string[]"
 
 
-def write_cytoscape_elements_for_string(workflow_content):
-    f = tempfile.NamedTemporaryFile(mode="w", suffix=".gxwf.yml")
-    f.write(workflow_content)
-    f.flush()
-    write_cytoscape_elements(f.name)
+def test_interop_generation():
+    # not much of a test case but it will generate a directory of interoperability examples to
+    # test Java against.
+    write_cytoscape_elements(EXAMPLE_PATH)
+    write_cytoscape_elements_for_string(WITH_REPORT, "with_report.gxwf.yml")
+
+
+def write_cytoscape_elements_for_string(workflow_content, filename):
+    # Stage the inline workflow at a stable, descriptive path before copying into
+    # TEST_CYTOSCAPE_EXAMPLES — anonymous tempfile names leak into the interop fixture
+    # directory and accumulate one tmp*.gxwf.yml per test run.
+    if not os.path.exists(TEST_CYTOSCAPE_EXAMPLES):
+        os.makedirs(TEST_CYTOSCAPE_EXAMPLES)
+    workflow_path = os.path.join(TEST_CYTOSCAPE_EXAMPLES, filename)
+    with open(workflow_path, "w") as f:
+        f.write(workflow_content)
+    write_cytoscape_elements(workflow_path)
 
 
 def write_cytoscape_elements(workflow_path):
     if not os.path.exists(TEST_CYTOSCAPE_EXAMPLES):
         os.makedirs(TEST_CYTOSCAPE_EXAMPLES)
     base_name, ext = os.path.splitext(os.path.basename(workflow_path))
-    shutil.copyfile(workflow_path, os.path.join(TEST_CYTOSCAPE_EXAMPLES, base_name + ext))
+    staged_path = os.path.join(TEST_CYTOSCAPE_EXAMPLES, base_name + ext)
+    if os.path.abspath(workflow_path) != os.path.abspath(staged_path):
+        shutil.copyfile(workflow_path, staged_path)
     elements_path = os.path.join(TEST_CYTOSCAPE_EXAMPLES, base_name + ".cytoscape.json")
     main([workflow_path, elements_path])

tests/test_interop_tests.py

@@ -8,6 +8,7 @@
 import os
 from typing import Any, Callable, Dict
 
+from gxformat2.cytoscape import cytoscape_elements as _cytoscape_impl
 from gxformat2.examples import EXAMPLES_DIR, load
 from gxformat2.lint import lint_best_practices_format2 as _lint_bp_format2_impl
 from gxformat2.lint import lint_best_practices_ga as _lint_bp_ga_impl
@@ -112,6 +113,20 @@ def _workflow_to_mermaid_with_comments_lines(wf_dict):
     return _mermaid_impl(wf_dict, comments=True).split("\n")
 
 
+def _cytoscape_elements_to_list(wf_dict):
+    return _cytoscape_impl(wf_dict).to_list()
+
+
+def _cytoscape_node_ids(wf_dict):
+    flat = _cytoscape_impl(wf_dict).to_list()
+    return [el["data"]["id"] for el in flat if el["group"] == "nodes"]
+
+
+def _cytoscape_edge_ids(wf_dict):
+    flat = _cytoscape_impl(wf_dict).to_list()
+    return [el["data"]["id"] for el in flat if el["group"] == "edges"]
+
+
 EXPECTATIONS_DIR = os.path.join(EXAMPLES_DIR, "expectations")
 OPERATIONS: Dict[str, Callable[..., Any]] = {
     "normalized_format2": normalized_format2,
@@ -133,6 +148,9 @@ def _workflow_to_mermaid_with_comments_lines(wf_dict):
     "workflow_to_mermaid": _workflow_to_mermaid,
     "workflow_to_mermaid_lines": _workflow_to_mermaid_lines,
     "workflow_to_mermaid_with_comments_lines": _workflow_to_mermaid_with_comments_lines,
+    "cytoscape_elements_to_list": _cytoscape_elements_to_list,
+    "cytoscape_node_ids": _cytoscape_node_ids,
+    "cytoscape_edge_ids": _cytoscape_edge_ids,
 }
 
 suite = DeclarativeTestSuite(