TRS ID Tests Review: Mocking/Patching Analysis and Refactoring Plan

trs_test_review.md

TRS ID Tests Review: Mocking/Patching Analysis

Overview

Review of tests/test_trs_id.py from the TRS ID support branch (claude/add-trs-id-support-mefAv). The branch adds support for running workflows via TRS IDs like workflow/github.com/iwc-workflows/parallel-accession-download/main.

Current Test Structure

Tests That Are Good (No Mocking)

These test pure string parsing logic - legitimate unit tests:

• test_parse_trs_id_workflow_with_version
• test_parse_trs_id_with_hash_prefix
• test_parse_trs_id_tool
• test_parse_trs_id_invalid
• test_parse_trs_uri_with_version
• test_parse_trs_uri_from_full_url
• test_parse_trs_uri_invalid

Problematic Mocking Patterns

1. HTTP Mocking (requests.get)

Test	Problem
test_parse_trs_id_workflow_without_version	Mocks requests.get, uses mock_get.assert_called_once() (interaction verification)
test_parse_trs_id_version_fetch_failure	Mocks exception - legitimate but could use recorded fixtures
test_parse_trs_uri_workflow	Same pattern

2. Galaxy Client Mocking (TestTRSWorkflowImport) - Most Problematic

All three tests mock the entire Galaxy client AND parse_trs_uri:

@patch("planemo.galaxy.workflows.parse_trs_uri")
def test_import_workflow_from_trs(self, mock_parse):
    mock_parse.return_value = {"trs_url": expected_trs_url}
    
    mock_gi = Mock()
    mock_gi.workflows._make_url.return_value = "..."
    mock_gi.workflows._post.return_value = {"id": "test_workflow_id"}
    
    result = import_workflow_from_trs(trs_uri, mock_gi)
    
    # Interaction verification (anti-pattern)
    mock_workflows._post.assert_called_once()
    call_args = mock_workflows._post.call_args
    assert call_args[1]["payload"]["trs_url"] == expected_trs_url

Problem: These tests verify the mock was called correctly, not that the code works. They essentially test:

if mock_parse_returns_none:
    raise ValueError()
else:
    mock_gi.workflows._post(url, payload={"trs_url": mock_parse_return_value})

3. translate_alias Mocking (TestTRSIdIntegration)

All 4 tests mock translate_alias to passthrough:

@patch("planemo.runnable_resolve.translate_alias")
def test_for_runnable_identifier_with_trs_id(self, mock_translate_alias):
    mock_translate_alias.side_effect = lambda ctx, identifier, profile: identifier

Problem: Bypasses alias resolution entirely - doesn't test real integration.

---

Implementation Plan

Part 1: HTTP Fixtures for Dockstore API

Goal: Replace @patch("planemo.galaxy.workflows.requests.get") with recorded fixtures

Step 1: Add test dependency

# setup.cfg or pyproject.toml
tests_require = responses>=0.23.0

Step 2: Create fixture directory structure

tests/
  fixtures/
    dockstore/
      versions_iwc_parallel_accession.json    # Real recorded response
      versions_empty.json                      # Edge case: no versions

Step 3: Record real responses (one-time)

# Script to record fixtures (run manually)
import requests
import json

url = "https://dockstore.org/api/ga4gh/trs/v2/tools/%23workflow%2Fgithub.com%2Fiwc-workflows%2Fparallel-accession-download%2Fmain/versions"
response = requests.get(url)
with open("tests/fixtures/dockstore/versions_iwc_parallel_accession.json", "w") as f:
    json.dump(response.json(), f, indent=2)

Step 4: Refactor tests

Before:

@patch("planemo.galaxy.workflows.requests.get")
def test_parse_trs_id_workflow_without_version(self, mock_get):
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = [{"name": "v0.2.0"}, {"name": "v0.1.14"}]
    mock_get.return_value = mock_response
    
    result = parse_trs_id(trs_id)
    mock_get.assert_called_once()  # Interaction verification

After:

import responses
from pathlib import Path

FIXTURES = Path(__file__).parent / "fixtures" / "dockstore"

@responses.activate
def test_parse_trs_id_workflow_without_version(self):
    fixture = json.loads((FIXTURES / "versions_iwc_parallel_accession.json").read_text())
    
    responses.add(
        responses.GET,
        "https://dockstore.org/api/ga4gh/trs/v2/tools/%23workflow%2Fgithub.com%2Fiwc-workflows%2Fparallel-accession-download%2Fmain/versions",
        json=fixture,
        status=200
    )
    
    trs_id = "workflow/github.com/iwc-workflows/parallel-accession-download/main"
    result = parse_trs_id(trs_id)
    
    # State verification - verify the output, not the call
    assert result["trs_url"] == f"https://dockstore.org/api/ga4gh/trs/v2/tools/#workflow/github.com/iwc-workflows/parallel-accession-download/main/versions/{fixture[0]['name']}"

Tests affected: 3 tests

• test_parse_trs_id_workflow_without_version
• test_parse_trs_id_version_fetch_failure
• test_parse_trs_uri_workflow

---

Part 2: TrsImporter Protocol (Dependency Injection)

Goal: Replace mocked GalaxyInstance with injectable interface

Step 1: Define protocol in planemo/galaxy/workflows.py

from typing import Protocol, runtime_checkable

@runtime_checkable
class TrsImporter(Protocol):
    """Interface for importing workflows from TRS."""
    def import_from_trs(self, trs_url: str) -> Dict[str, Any]: ...

Step 2: Create Galaxy implementation

class GalaxyTrsImporter:
    """Import TRS workflows via Galaxy API."""
    
    def __init__(self, gi: "GalaxyInstance"):
        self._gi = gi
    
    def import_from_trs(self, trs_url: str) -> Dict[str, Any]:
        url = self._gi.workflows._make_url() + "/upload"
        return self._gi.workflows._post(url=url, payload={"trs_url": trs_url})

Step 3: Create fake for testing

# tests/fake_trs.py
class FakeTrsImporter:
    """Test fake that records imports without calling Galaxy."""
    
    def __init__(self, return_workflow: Optional[Dict[str, Any]] = None):
        self.imported_urls: List[str] = []
        self._return = return_workflow or {"id": "fake_wf_id", "name": "Fake Workflow"}
    
    def import_from_trs(self, trs_url: str) -> Dict[str, Any]:
        self.imported_urls.append(trs_url)
        return self._return

Step 4: Refactor import_workflow_from_trs

def import_workflow_from_trs(
    trs_uri: str, 
    importer: TrsImporter
) -> Dict[str, Any]:
    """Import a workflow from a TRS endpoint.
    
    Args:
        trs_uri: TRS URI (trs://workflow/github.com/org/repo/name[/version])
        importer: TrsImporter implementation (GalaxyTrsImporter in production)
    """
    trs_info = parse_trs_uri(trs_uri)
    if not trs_info:
        raise ValueError(f"Invalid TRS URI: {trs_uri}")
    
    return importer.import_from_trs(trs_info["trs_url"])

Step 5: Update call sites

Find all callers and change from:

workflow = import_workflow_from_trs(trs_uri, user_gi)

To:

importer = GalaxyTrsImporter(user_gi)
workflow = import_workflow_from_trs(trs_uri, importer)

Step 6: Refactor tests to use fake

def test_import_workflow_from_trs(self):
    fake = FakeTrsImporter(return_workflow={"id": "test_id", "name": "Test"})
    trs_uri = "trs://workflow/github.com/org/repo/main/v0.1.14"
    
    result = import_workflow_from_trs(trs_uri, fake)
    
    # State verification (good)
    assert result["id"] == "test_id"
    assert len(fake.imported_urls) == 1
    assert "workflow/github.com/org/repo/main/versions/v0.1.14" in fake.imported_urls[0]

---

Part 3: Fix translate_alias Tests

Before:

@patch("planemo.runnable_resolve.translate_alias")
def test_for_runnable_identifier_with_trs_id(self, mock_translate_alias):
    mock_translate_alias.side_effect = lambda ctx, identifier, profile: identifier
    ctx = Mock()
    ...

After:

def test_for_runnable_identifier_with_trs_id(self, tmp_path):
    # Create minimal real context with no aliases
    ctx = planemo.cli.PlanemoCliContext()
    ctx.workspace = str(tmp_path)
    
    trs_id = "workflow/github.com/iwc-workflows/parallel-accession-download/main"
    runnable = for_runnable_identifier(ctx, trs_id, {})
    
    assert runnable.type == RunnableType.galaxy_workflow
    assert runnable.uri == f"{TRS_WORKFLOWS_PREFIX}{trs_id}"

---

Part 4: Additional Integration Tests (test_run.py)

The existing test_run_trs_id is good. Add coverage for:

@skip_if_environ("PLANEMO_SKIP_GALAXY_TESTS")
@mark.tests_galaxy_branch
def test_run_trs_id_with_version(self):
    """Test running workflow with explicit TRS version."""
    with self._isolate() as f:
        trs_id = "workflow/github.com/iwc-workflows/parallel-accession-download/main/v0.1.14"
        # ... same job content ...
        test_cmd = ["--verbose", "run", "--no_dependency_resolution", 
                    "--galaxy_branch", target_galaxy_branch(),
                    "--test_data", TEST_DATA_DIR, trs_id, job_path]
        self._check_exit_code(test_cmd)

@skip_if_environ("PLANEMO_SKIP_GALAXY_TESTS")
@mark.tests_galaxy_branch  
def test_run_trs_id_with_hash_prefix(self):
    """Test running workflow with # prefix in TRS ID."""
    with self._isolate() as f:
        trs_id = "#workflow/github.com/iwc-workflows/parallel-accession-download/main"
        # ... rest same ...

---

Summary: Files to Change

File	Changes
setup.cfg	Add responses test dependency
tests/fixtures/dockstore/	Create fixture JSON files
planemo/galaxy/workflows.py	Add TrsImporter protocol, GalaxyTrsImporter class
tests/fake_trs.py (new)	FakeTrsImporter class
tests/test_trs_id.py	Refactor all tests per above
tests/test_run.py	Add 2 integration tests
Call sites of import_workflow_from_trs	Update to use GalaxyTrsImporter

---

Open Questions

1. responses vs pytest-httpx vs respx - which HTTP mocking lib preferred?
2. Should FakeTrsImporter live in test code or production code (for downstream testing)?
3. Delete TestTRSWorkflowImport entirely after adding protocol tests, or keep for fast feedback?
4. How many Dockstore fixtures to record (just one workflow or multiple)?

---

Key Principles Applied

1. Prefer Dependency Injection over Patching - TrsImporter protocol
2. Replace Mocks with Fakes - FakeTrsImporter records state instead of verifying calls
3. Use Real Objects with Controlled Inputs - Real translate_alias with empty config
4. Favor State Verification over Interaction Verification - Assert results, not mock calls
5. Use Contract Tests for External Services - Recorded Dockstore fixtures