pytest_wdl/executors/_cromwell.py
def _get_workflow_imports(
        cls,
        import_dirs: Optional[Sequence[Path]] = None,
        imports_file: Optional[Path] = None,
    ) -> Path:
        """
        Creates a ZIP file with all WDL files to be imported.

        Args:
            imports_file: Text file naming import directories/files - one per line.

        Returns:
            Path to the ZIP file.
        """
        write_imports = bool(import_dirs)
        imports_path = None

        if imports_file:
            imports_path = ensure_path(imports_file)

            if imports_path.exists():
                write_imports = False

        if write_imports and import_dirs:
            imports = [
                wdl for path in import_dirs for wdl in glob.glob(str(path / "*.wdl"))
            ]

            if imports:
                if imports_path:
                    ensure_path(imports_path, is_file=True, create=True)
                else:
                    imports_path = Path(tempfile.mkstemp(suffix=".zip")[1])

                imports_str = " ".join(imports)

                LOG.info(f"Writing imports {imports_str} to zip file {imports_path}")

                exe = subby.run(
                    f"zip -j - {imports_str}",
                    mode=bytes,
                    stdout=imports_path,
                    raise_on_error=False,
                )

                if not exe.ok:
                    raise Exception(
                        f"Error creating imports zip file; stdout={exe.output}; "
                        f"stderr={exe.error}"
                    )

        return imports_path

        Creates a ZIP file with all WDL files to be imported.

        Args:
            imports_file: Text file naming import directories/files - one per line.

        Returns:
            Path to the ZIP file.
        
Creates a ZIP file with all WDL files to be imported.
Creates a ZIP file with all WDL files to be imported .
def _get_workflow_imports(
        cls,
        import_dirs: Optional[Sequence[Path]] = None,
        imports_file: Optional[Path] = None,
    ) -> Path:
        write_imports = bool(import_dirs)
        imports_path = None
        if imports_file:
            imports_path = ensure_path(imports_file)
            if imports_path.exists():
                write_imports = False
        if write_imports and import_dirs:
            imports = [
                wdl for path in import_dirs for wdl in glob.glob(str(path / "*.wdl"))
            ]
            if imports:
                if imports_path:
                    ensure_path(imports_path, is_file=True, create=True)
                else:
                    imports_path = Path(tempfile.mkstemp(suffix=".zip")[1])
                imports_str = " ".join(imports)
                LOG.info(f"Writing imports {imports_str} to zip file {imports_path}")
                exe = subby.run(
                    f"zip -j - {imports_str}",
                    mode=bytes,
                    stdout=imports_path,
                    raise_on_error=False,
                )
                if not exe.ok:
                    raise Exception(
                        f"Error creating imports zip file; stdout={exe.output}; "
                        f"stderr={exe.error}"
                    )
        return imports_path
def _get_workflow_imports ( cls , import_dirs : Optional [ Sequence [ Path ] ] = None , imports_file : Optional [ Path ] = None , ) -> Path : write_imports = bool ( import_dirs ) imports_path = None if imports_file : imports_path = ensure_path ( imports_file ) if imports_path . exists ( ) : write_imports = False if write_imports and import_dirs : imports = [ wdl for path in import_dirs for wdl in glob . glob ( str ( path / "*.wdl" ) ) ] if imports : if imports_path : ensure_path ( imports_path , is_file = True , create = True ) else : imports_path = Path ( tempfile . mkstemp ( suffix = ".zip" ) [ 1 ] ) imports_str = " " . join ( imports ) LOG . info ( f"Writing imports {imports_str} to zip file {imports_path}" ) exe = subby . run ( f"zip -j - {imports_str}" , mode = bytes , stdout = imports_path , raise_on_error = False , ) if not exe . ok : raise Exception ( f"Error creating imports zip file; stdout={exe.output}; " f"stderr={exe.error}" ) return imports_path
Creates a ZIP file with all WDL files to be imported.
Creates a ZIP file with all WDL files to be imported .
"""
        Creates a ZIP file with all WDL files to be imported.

        Args:
            imports_file: Text file naming import directories/files - one per line.

        Returns:
            Path to the ZIP file.
        """
Path to the ZIP file.
Path to the ZIP file .
cls
import_dirs
Optional[Sequence[Path]]
imports_file
Optional[Path]
Text file naming import directories/files - one per line.
Text file naming import directories / files - one per line .