lint

Author: karlnyr

microSALT/utils/job_creator.py

@@ -11,21 +11,45 @@
 import shutil
 import subprocess
 import sys
-from importlib.metadata import entry_points
 import time
 from datetime import datetime
+from importlib.metadata import entry_points
 from pathlib import Path
 
 import yaml
 
 from microSALT import __version__
-from microSALT.config import Folders, Threshold, SlurmHeader, Regex, PubMLSTCredentials, PasteurCredentials, Singularity, Containers
+from microSALT.config import (
+    Containers,
+    Folders,
+    PasteurCredentials,
+    PubMLSTCredentials,
+    Regex,
+    Singularity,
+    SlurmHeader,
+    Threshold,
+)
 from microSALT.store.db_manipulator import DB_Manipulator
 from microSALT.utils.referencer import Referencer
 
 
 class Job_Creator:
-    def __init__(self, log, folders: Folders, slurm_header: SlurmHeader, regex: Regex, dry: bool, config_path: str, threshold: Threshold, pubmlst: PubMLSTCredentials, pasteur: PasteurCredentials, singularity: Singularity, containers: Containers, sampleinfo={}, run_settings={}):
+    def __init__(
+        self,
+        log,
+        folders: Folders,
+        slurm_header: SlurmHeader,
+        regex: Regex,
+        dry: bool,
+        config_path: str,
+        threshold: Threshold,
+        pubmlst: PubMLSTCredentials,
+        pasteur: PasteurCredentials,
+        singularity: Singularity,
+        containers: Containers,
+        sampleinfo={},
+        run_settings={},
+    ):
         self.folders = folders
         self.slurm_header = slurm_header
         self.regex = regex
@@ -89,7 +113,15 @@ def __init__(self, log, folders: Folders, slurm_header: SlurmHeader, regex: Rege
             self.finishdir = f"{folders.results}/{self.name}_{self.now}"
         self.db_pusher = DB_Manipulator(log=log, folders=folders, threshold=threshold)
         self.concat_files = dict()
-        self.ref_resolver = Referencer(log=log, folders=folders, threshold=threshold, pubmlst=pubmlst, pasteur=pasteur, singularity=singularity, containers=containers)
+        self.ref_resolver = Referencer(
+            log=log,
+            folders=folders,
+            threshold=threshold,
+            pubmlst=pubmlst,
+            pasteur=pasteur,
+            singularity=singularity,
+            containers=containers,
+        )
 
     def get_sbatch(self):
         """Returns sbatchfile, slightly superflous"""
@@ -141,15 +173,13 @@ def verify_fastq(self):
                     else:
                         pairno = 2 - 1 % int(file_match[1])  # 1->2, 2->1
                         # Construct mate name
-                        pairname = f"{file_match.string[:file_match.end(1) - 1]}{pairno}{file_match.string[file_match.end(1):file_match.end()]}"
+                        pairname = f"{file_match.string[: file_match.end(1) - 1]}{pairno}{file_match.string[file_match.end(1) : file_match.end()]}"
                     if pairname in files:
                         files.pop(files.index(pairname))
                         verified_files.append(file_match[0])
                         verified_files.append(pairname)
                 else:
-                    raise Exception(
-                        f"Some fastq files have no mate in directory {self.indir}."
-                    )
+                    raise Exception(f"Some fastq files have no mate in directory {self.indir}.")
         if verified_files == []:
             raise Exception(
                 f"No files in directory {self.indir} match file_pattern '{self.regex.file_pattern}'."
@@ -163,9 +193,7 @@ def verify_fastq(self):
                 if bsize > 1000:
                     self.logger.warning(f"Input fastq {vfile} exceeds 1000MB")
             except Exception:
-                self.logger.warning(
-                    f"Unable to verify size of input file {self.indir}/{vfile}"
-                )
+                self.logger.warning(f"Unable to verify size of input file {self.indir}/{vfile}")
 
         # Warn about invalid fastq files
         for vfile in verified_files:
@@ -198,9 +226,7 @@ def create_assemblysection(self):
             f"--contigs_out {contigs_file_raw} "
             f"--reads {self.concat_files['f']},{self.concat_files['r']}"
         )
-        batchfile.write(
-            f"mkdir -p {assembly_dir} &" f"{self._singularity_exec('skesa', skesa_cmd)}\n"
-        )
+        batchfile.write(f"mkdir -p {assembly_dir} &{self._singularity_exec('skesa', skesa_cmd)}\n")
 
         # Convert sequence naming in Skesa output into Spades format in the contigs fasta file:
         # ----------------------------------------------
@@ -318,13 +344,10 @@ def create_variantsection(self):
             + "\n"
         )
         batchfile.write(
-            self._singularity_exec("samtools", f"samtools index {outbase}.bam_sort_rmdup")
-            + "\n"
+            self._singularity_exec("samtools", f"samtools index {outbase}.bam_sort_rmdup") + "\n"
         )
         batchfile.write(
-            self._singularity_exec(
-                "samtools", f"samtools idxstats {outbase}.bam_sort_rmdup"
-            )
+            self._singularity_exec("samtools", f"samtools idxstats {outbase}.bam_sort_rmdup")
             + f" &> {outbase}.stats.ref\n"
         )
         # Removal of temp aligment files
@@ -579,9 +602,7 @@ def project_job(self, single_sample=False):
             else:
                 self.create_project(self.name)
         except Exception:
-            self.logger.error(
-                f"LIMS interaction failed. Unable to read/write project {self.name}"
-            )
+            self.logger.error(f"LIMS interaction failed. Unable to read/write project {self.name}")
         # Writes the job creation sbatch
         if single_sample:
             try:
@@ -648,7 +669,9 @@ def project_job(self, single_sample=False):
 
     def _write_mailjob(self, mailfile: str, report: str, custom_conf: str) -> None:
         """Write the mailjob.sh script that runs `microsalt utils finish` after all jobs complete."""
-        _ep = next(ep for ep in entry_points(group="console_scripts") if ep.value == "microSALT.cli:root")
+        _ep = next(
+            ep for ep in entry_points(group="console_scripts") if ep.value == "microSALT.cli:root"
+        )
         microsalt_bin = Path(sys.executable).parent / _ep.name
         with open(mailfile, "w+") as mb:
             mb.write("#!/usr/bin/env bash\n\n")
@@ -788,9 +811,7 @@ def sample_job(self):
             except Exception:
                 self.logger.error(f"Unable to access LIMS info for sample {self.name}")
         except Exception as e:
-            self.logger.error(
-                f"Unable to create job for sample {self.name}\nSource: {e!s}"
-            )
+            self.logger.error(f"Unable to create job for sample {self.name}\nSource: {e!s}")
             shutil.rmtree(self.finishdir, ignore_errors=True)
             raise
 
@@ -824,9 +845,7 @@ def snp_job(self):
         batchfile = open(self.batchfile, "a+")
         batchfile.close()
 
-        headerline = (
-            f"-A {self.slurm_header.project} -p {self.slurm_header.type} -n 1 -t 24:00:00 -J {self.slurm_header.job_prefix}_{self.name} --qos {self.slurm_header.qos} --output {self.finishdir}/slurm_{self.name}.log"
-        )
+        headerline = f"-A {self.slurm_header.project} -p {self.slurm_header.type} -n 1 -t 24:00:00 -J {self.slurm_header.job_prefix}_{self.name} --qos {self.slurm_header.qos} --output {self.finishdir}/slurm_{self.name}.log"
         outfile = self.get_sbatch()
         bash_cmd = f"sbatch {headerline} {outfile}"
         samproc = subprocess.Popen(bash_cmd.split(), stdout=subprocess.PIPE)

tests/utils/test_jobcreator.py

@@ -1,9 +1,10 @@
-#!/usr/bin/env python
-
+from pathlib import Path
 from unittest import mock
-
 from unittest.mock import patch
 
+import pytest
+
+from microSALT.config import MicroSALTConfig
 from microSALT.utils.job_creator import Job_Creator
 
 
@@ -136,6 +137,7 @@ def test_project_job(subproc, config, logger, testdata):
         )
         jc.project_job()
 
+
 def test_singularity_exec_binds_finishdir(config, logger, testdata):
     """finishdir is automatically added to the --bind list of every singularity exec call."""
     jc = Job_Creator(
@@ -160,6 +162,7 @@ def test_singularity_exec_binds_finishdir(config, logger, testdata):
 def test_singularity_exec_does_not_duplicate_finishdir(config, logger, testdata):
     """finishdir is not listed twice when it already appears in singularity.bind_paths."""
     from microSALT.config import Singularity
+
     singularity = Singularity(bind_paths=["/tmp/test_runfolder", "/data"])
     jc = Job_Creator(
         log=logger,
@@ -180,7 +183,9 @@ def test_singularity_exec_does_not_duplicate_finishdir(config, logger, testdata)
     assert cmd.count("/tmp/test_runfolder") == 1
 
 
-def _make_jc(config, logger, testdata, tmp_path) -> Job_Creator:
+def _make_jc(
+    config: MicroSALTConfig, logger: pytest.LogCaptureFixture, testdata: dict, tmp_path: Path
+) -> Job_Creator:
     return Job_Creator(
         log=logger,
         folders=config.folders,
@@ -201,24 +206,22 @@ def _make_jc(config, logger, testdata, tmp_path) -> Job_Creator:
 def test_write_mailjob_uses_existing_executable(config, logger, testdata, tmp_path):
     """The binary embedded in mailjob.sh must exist on the filesystem."""
     import pathlib
+
     jc = _make_jc(config, logger, testdata, tmp_path)
     mailfile = str(tmp_path / "mailjob.sh")
 
     jc._write_mailjob(mailfile, "default", "")
 
     content = pathlib.Path(mailfile).read_text()
     # Extract the first token of the finish command (the binary path)
-    bin_path = next(
-        line.split()[0]
-        for line in content.splitlines()
-        if "utils finish" in line
-    )
+    bin_path = next(line.split()[0] for line in content.splitlines() if "utils finish" in line)
     assert pathlib.Path(bin_path).exists(), f"Binary not found on disk: {bin_path}"
 
 
 def test_write_mailjob_contains_finish_command(config, logger, testdata, tmp_path):
     """mailjob.sh must contain the expected microsalt utils finish invocation."""
     import pathlib
+
     jc = _make_jc(config, logger, testdata, tmp_path)
     mailfile = str(tmp_path / "mailjob.sh")
 
@@ -228,4 +231,3 @@ def test_write_mailjob_contains_finish_command(config, logger, testdata, tmp_pat
     assert "utils finish" in content
     assert "--report qc" in content
     assert str(tmp_path) in content
-