Skip to content
6 changes: 4 additions & 2 deletions snakemake_executor_plugin_slurm/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,9 @@ def run_job(self, job: JobExecutorInterface):
group_or_rule = f"group_{job.name}" if job.is_group() else f"rule_{job.name}"

try:
wildcard_str = "_".join(job.wildcards) if job.wildcards else ""
wildcard_str = (
"_".join(job.wildcards).replace("/", "_") if job.wildcards else ""
)
except AttributeError:
wildcard_str = ""

Expand Down Expand Up @@ -264,10 +266,10 @@ def run_job(self, job: JobExecutorInterface):
"specified. Assuming 'tasks_per_node=1'."
"Probably not what you want."
)

# we need to set cpus-per-task OR cpus-per-gpu, the function
# will return a string with the corresponding value
call += f" {get_cpu_setting(job, gpu_job)}"

if job.resources.get("slurm_extra"):
self.check_slurm_extra(job)
call += f" {job.resources.slurm_extra}"
Expand Down
44 changes: 44 additions & 0 deletions tests/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,3 +110,47 @@ def test_both_gres_and_gpu_set(self, mock_job):
WorkflowError, match="GRES and GPU are set. Please only set one of them."
):
set_gres_string(job)


class TestWildcardsWithSlashes(snakemake.common.tests.TestWorkflowsLocalStorageBase):
"""Test handling of wildcards with slashes to ensure log directories are correctly constructed."""

__test__ = True

def get_executor(self) -> str:
return "slurm"

def get_executor_settings(self) -> Optional[ExecutorSettingsBase]:
return ExecutorSettings(logdir="test_logdir")

def test_wildcard_slash_replacement(self):
"""Test that slashes in wildcards are correctly replaced with underscores in log directory paths."""

# Create a mock job with wildcards containing slashes
class MockJob:
def __init__(self):
self.name = "test_job"
self.wildcards = ["/leading_slash", "middle/slash", "trailing/"]
self.is_group = lambda: False

# Create an executor instance
executor = Executor()
executor.workflow = snakemake.common.tests.DummyWorkflow()
executor.workflow.executor_settings = self.get_executor_settings()
executor.slurm_logdir = os.path.abspath("test_logdir")

# Manually call the wildcard sanitization logic from run_job method
try:
wildcard_str = (
"_".join(MockJob().wildcards).replace("/", "_")
if MockJob().wildcards
else ""
)
except AttributeError:
wildcard_str = ""

# Assert that slashes are correctly replaced with underscores
assert wildcard_str == "_leading_slash_middle_slash_trailing_"

# Verify no slashes remain in the wildcard string
assert "/" not in wildcard_str
Loading