Clinical-Genomics / trailblazer / 11681030687

coverage: 88.24%. First build
11681030687

Pull #501

github

henrikstranneheim 
feat(update): Werkzeug
Pull Request #501: feat(update): Werkzeug

2056 of 2330 relevant lines covered (88.24%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

53.33
/trailblazer/services/slurm/slurm_cli_service/slurm_cli_service.py
1 
from trailblazer.clients.slurm_cli_client.models import SqueueJob, SqueueResult
1✔
2 
from trailblazer.clients.slurm_cli_client.slurm_cli_client import SlurmCLIClient
1✔
3 
from trailblazer.clients.slurm_cli_client.mapper import create_job_info_dto
1✔
4 
from trailblazer.services.slurm.dtos import SlurmJobInfo
1✔
5 
from trailblazer.services.slurm.slurm_api_service.mappers import create_job
1✔
6 
from trailblazer.services.slurm.slurm_service import SlurmService
1✔
7 
from trailblazer.services.slurm.utils import get_slurm_job_ids
1✔
8 
from trailblazer.store.models import Analysis
1✔
9 
from trailblazer.store.store import Store
1✔
10 









11












12



class SlurmCLIService(SlurmService):




1✔





13



    def __init__(self, client: SlurmCLIClient, store: Store):




1✔





14



        self.client = client




1✔





15



        self.store = store




1✔





16












17



    def get_job(self, job_id: int) -> SlurmJobInfo:




1✔





18



        queue: SqueueResult = self.client.get_slurm_queue([job_id])




×







19



        job: SqueueJob = queue.jobs[0]




×







20



        return create_job_info_dto(job)




×







21












22



    def update_jobs(self, analysis_id: int) -> None:




1✔





23



        analysis: Analysis = self.store.get_analysis_with_id(analysis_id)




×







24



        job_ids: list[int] = get_slurm_job_ids(analysis.config_path)




×







25



        queue: SqueueResult = self.client.get_slurm_queue(job_ids)




×







26



        dtos = [create_job_info_dto(job) for job in queue.jobs]




×







27



        jobs = [create_job(dto) for dto in dtos]




×







28



        self.store.replace_jobs(analysis_id=analysis_id, jobs=jobs)




×







29












30



    def cancel_jobs(self, analysis_id: int) -> None:




1✔





31



        analysis: Analysis = self.store.get_analysis_with_id(analysis_id)




×







32



        job_ids: list[int] = get_slurm_job_ids(analysis.config_path)




×







33



        for job_id in job_ids:




×







34



            self.client.cancel_job(job_id)




×







35



        self.update_jobs(analysis_id)




×





















    

  • 

    •