File indexing completed on 2024-05-12 05:43:36

0001 """
0002 This module contains code for the pipeline command
0003 """
0004 
0005 # SPDX-FileCopyrightText: 2021 Leon Morten Richter <github@leonmortenrichter.de>
0006 #
0007 # SPDX-License-Identifier: GPL-2.0-or-later
0008 
0009 import argparse
0010 import os
0011 import sys
0012 from enum import Enum
0013 from typing import List, Optional, Dict
0014 
0015 from gitlab.exceptions import GitlabGetError
0016 from gitlab.v4.objects import ProjectPipeline
0017 
0018 from lab.repositoryconnection import RepositoryConnection
0019 from lab.table import Table
0020 from lab.utils import TextFormatting, Utils, LogType
0021 
0022 
0023 class PipelineStatus(Enum):
0024     """
0025     Currently supported Pipeline Status from Gitlab.
0026 
0027     Allowed : https://docs.gitlab.com/ce/api/pipelines.html#list-project-pipelines
0028     """
0029 
0030     WAITING = "waiting_for_resource"
0031     PENDING = "pending"
0032     RUNNING = "running"
0033     SUCCESS = "success"
0034     FAILED = "failed"
0035     CANCELED = "canceled"
0036     SKIPPED = "skipped"
0037     SCHEDULED = "scheduled"
0038 
0039     @classmethod
0040     def _missing_(cls, value: object) -> None:
0041         """
0042         Called if a given value does not exist.
0043         Supported in all version of Python since 3.6.
0044         """
0045         Utils.log(LogType.WARNING, f"Invalid status '{str(value)}'")
0046         sys.exit(1)
0047 
0048     @staticmethod
0049     def format(status: str) -> str:
0050         """
0051         Transform a status into a formatted string, that is a colored string.
0052         If no color is defined for the given status, this is a NO-OP.
0053         """
0054         formatting = {
0055             "success": TextFormatting.GREEN,
0056             "failed": TextFormatting.RED,
0057             "running": TextFormatting.BLUE,
0058             "canceled": TextFormatting.PURPLE,
0059             "pending": TextFormatting.BLUE,
0060             "waiting_for_resource": TextFormatting.YELLOW,
0061         }
0062 
0063         if status in formatting:
0064             # Only format if there is a defined color for the status
0065             return f"{formatting.get(status, '')}{status}{TextFormatting.END}"
0066 
0067         # If color is associated with the status, do nothing
0068         return status
0069 
0070     @property
0071     def formatted(self) -> str:
0072         """
0073         Formatted string representation of a status.
0074         """
0075         return self.format(str(self))
0076 
0077     @property
0078     def finished(self) -> bool:
0079         """
0080         Is the pipeline finished, that is either skipped, canceled, failed or succeeded.
0081         """
0082         finished_states = (
0083             PipelineStatus.SKIPPED,
0084             PipelineStatus.SUCCESS,
0085             PipelineStatus.FAILED,
0086             PipelineStatus.CANCELED,
0087         )
0088         return self in finished_states
0089 
0090     def __str__(self) -> str:
0091         return str(self.value)
0092 
0093     def __repr__(self) -> str:
0094         return str(self)
0095 
0096 
0097 def parser(
0098     subparsers: argparse._SubParsersAction,  # pylint: disable=protected-access
0099 ) -> argparse.ArgumentParser:
0100     """
0101     Subparser for pipeline command
0102     :param subparsers: subparsers object from global parser
0103     :return: search subparser
0104     """
0105     pipeline_parser: argparse.ArgumentParser = subparsers.add_parser(
0106         "pipelines", help="Fetch pipeline status from GitLab."
0107     )
0108 
0109     # Optionally filter by status. This is None by default.
0110     pipeline_parser.add_argument(
0111         "--status",
0112         help=f"Filter pipelines by status, one of: [{', '.join(str(s) for s in PipelineStatus)}]",
0113         metavar="",
0114         type=PipelineStatus,
0115         default=None,
0116         choices=list(PipelineStatus),
0117     )
0118 
0119     pipeline_parser.add_argument(
0120         "--ref",
0121         help="Filter pipelines by reference, e.g. 'master'",
0122         metavar="ref",
0123         type=str,
0124         nargs="?",
0125     )
0126 
0127     pipeline_parser.add_argument(
0128         "pipeline_id",
0129         help="Show pipeline by id if provided",
0130         metavar="pipeline_id",
0131         type=int,
0132         nargs="?",
0133     )
0134 
0135     return pipeline_parser
0136 
0137 
0138 def run(args: argparse.Namespace) -> None:
0139     """
0140     :param args: parsed arguments
0141     """
0142     if args.pipeline_id is not None:
0143         pipeline: PipelineShow = PipelineShow(args.pipeline_id)
0144         print(pipeline)
0145 
0146     else:
0147         lister: PipelineList = PipelineList(args.status, ref=args.ref)
0148         lister.print_formatted_list()
0149 
0150 
0151 class PipelineShow(RepositoryConnection):
0152     """
0153     Show single pipeline
0154     """
0155 
0156     def __init__(self, pipeline_id: int) -> None:
0157         RepositoryConnection.__init__(self)
0158         try:
0159             self.pipeline: ProjectPipeline = self._remote_project.pipelines.get(
0160                 pipeline_id, lazy=False
0161             )
0162         except GitlabGetError:
0163             Utils.log(LogType.WARNING, f"No pipeline with ID {pipeline_id}")
0164             sys.exit(1)
0165 
0166     def __str__(self) -> str:
0167         """
0168         Human readable representation of a pipeline.
0169         Output looks like:
0170         "Pipeline #25794 for master triggered by some one(Someone) 2 months ago
0171         finished after 1m 52s with status: success"
0172         """
0173         status: PipelineStatus = PipelineStatus(self.pipeline.status)
0174 
0175         text_buffer: str = ""
0176         text_buffer += (
0177             f"Pipeline {TextFormatting.BOLD}#{self.pipeline.id}"
0178             + f"{TextFormatting.END} for {self.pipeline.ref} "
0179             + f"triggered by {self.pipeline.user['name']}({self.pipeline.user['username']}) "
0180             + f"{Utils.pretty_date(self.pipeline.created_at)} "
0181         )
0182 
0183         if status.finished:
0184             text_buffer += (
0185                 f"finished after {Utils.pretty_time_delta(self.pipeline.duration or 0)} "
0186                 + f"with status: {status.formatted}"
0187             )
0188         else:
0189             text_buffer += f"is currently {status.formatted}"
0190 
0191         text_buffer += os.linesep
0192         return text_buffer
0193 
0194 
0195 class PipelineList(RepositoryConnection):
0196     """
0197     Search class
0198     """
0199 
0200     def __init__(self, status: Optional[PipelineStatus] = None, ref: Optional[str] = None) -> None:
0201         RepositoryConnection.__init__(self)
0202 
0203         self.status: Optional[PipelineStatus] = status
0204         self.ref: Optional[str] = ref
0205 
0206         if ref is not None and ref not in self._local_repo.refs:
0207             # Print a warning, if the ref is not found LOCALLY
0208             # The remote may contain refs, that do not exists inside the local copy,
0209             # therefore only a warning is printed.
0210             Utils.log(LogType.WARNING, f"Ref '{ref}' is not found locally.")
0211 
0212     def print_formatted_list(self) -> None:
0213         """
0214         Print the list of pipelines to terminal formatted as a table
0215         """
0216         table = Table()
0217 
0218         # Compute args that are sent to GitLab
0219         args: Dict[str, str] = {}
0220         if self.status is not None:
0221             # Only yield pipelines that have the specific status
0222             # If empty all pipelines will be returned by GitLab
0223             args["status"] = str(self.status)
0224 
0225         if self.ref is not None:
0226             # Only yield pipeline that match a given reference
0227             args["ref"] = self.ref
0228 
0229         # Build the printable table
0230         pipelines: List[ProjectPipeline] = self._remote_project.pipelines.list(**args)
0231         for pipeline in pipelines:
0232             row: List[str] = [
0233                 TextFormatting.BOLD + "#" + str(pipeline.id) + TextFormatting.END,
0234                 pipeline.ref,
0235                 Utils.pretty_date(pipeline.created_at),
0236                 PipelineStatus.format(pipeline.status),
0237             ]
0238             table.add_row(row)
0239 
0240         table.print()