Merge pull request #472 from lyz-code/feat/pipeline_times

feat/pipeline times

.cruft.json

@@ -1,6 +1,6 @@
 {
   "template": "git@github.com:lyz-code/cookiecutter-python-project.git",
-  "commit": "8a05453b3d6c9022546fba512ffc0df42da192de",
+  "commit": "85e6779b497a75fd7bd25690a2e00763e3b1152f",
   "context": {
     "cookiecutter": {
       "project_name": "Drode",

pyproject.toml

@@ -24,7 +24,7 @@ authors = [
     {name = "Lyz", email = "lyz-code-security-advisories@riseup.net"},
 ]
 license = {text = "GPL-3.0-only"}
-requires-python = ">=3.7.2"
+requires-python = ">=3.7"
 dependencies = [
     "click>=8.1.3",
     "goodconf[yaml]>=2.0.1",
@@ -65,15 +65,13 @@ drode = "drode.entrypoints.cli:cli"
 version = {from = "src/drode/version.py"}
 package-dir = "src"
 source-includes = ["tests/"]
+allow_prereleases = true
 
 [tool.pdm.overrides]
 
 # To be removed once https://github.com/flakeheaven/flakeheaven/issues/55 is solved
 importlib-metadata = ">=3.10"
 
-# To be removed once https://github.com/mkdocs/mkdocs/issues/2892 is solved
-markdown = "<3.4"
-
 [tool.pdm.build]
 editable-backend = "path"
 
@@ -112,7 +110,6 @@ test = [
     "pytest-freezegun>=0.4.2",
     "pydantic-factories>=1.6.1",
     "requests-mock>=1.9.3",
-    "feedparser>=6.0.10",
 ]
 doc = [
     "mkdocs>=1.3.1",
@@ -121,12 +118,12 @@ doc = [
     "mkdocs-minify-plugin>=0.5.0",
     "mkdocs-autolinks-plugin>=0.6.0",
     "mkdocs-material>=8.4.2",
-    "mkdocstrings[python]>=0.18",
     "markdown-include>=0.7.0",
     "mkdocs-section-index>=0.3.4",
+    "mkdocstrings[python]>=0.18.0",
 ]
 security = [
-    "safety>=2.1.1",
+    "safety>=2.3.1",
     "bandit>=1.7.3",
 ]
 fixers = [
@@ -137,17 +134,16 @@ fixers = [
 ]
 typing = [
     "mypy>=0.971",
-    "types-beautifulsoup4>=4.11.2",
-    "types-requests>=2.28.11.2",
-    "types-tabulate>=0.9.0.0",
+    "types-click>=7.1.8",
+    "types-requests>=2.27.16",
+    "types-tabulate>=0.8.6",
 ]
 dev = [
     "pre-commit>=2.20.0",
     "twine>=4.0.1",
     "commitizen>=2.32.2",
 ]
 
-
 [build-system]
 requires = ["pdm-pep517"]
 build-backend = "pdm.pep517.api"
@@ -182,7 +178,7 @@ exclude = '''
 [tool.pytest.ini_options]
 minversion = "6.0"
 addopts = "-vv --tb=short -n auto"
-log_level = "debug"
+log_level = "info"
 norecursedirs = [
     ".tox",
     ".git",

src/drode/adapters/aws.py

@@ -70,6 +70,7 @@ def get_autoscaling_group(autoscaling_name: str) -> AutoscalerInfo:
                         'Template': str = LaunchConfiguration or
                             LaunchTemplate:LaunchTemplateVersion that generated the
                             instance.
+
         Raises:
             AWSStateError: If no autoscaling groups are found with that name.
         """

src/drode/adapters/drone.py

@@ -26,8 +26,8 @@ class DronePromoteError(Exception):
     """Exception to gather job promotion errors."""
 
 
-@dataclass
 # R0902: Too many attributes, but it's a model, so it doesn't mind
+@dataclass  # noqa: R0902
 class BuildInfo:  # noqa: R0902
     """Build information schema."""
 
@@ -69,6 +69,14 @@ def from_kwargs(cls, **kwargs: Any) -> "BuildInfo":  # noqa: ANN401
 
         return entity
 
+    def __gt__(self, other: "BuildInfo") -> bool:
+        """Return if we're greater than other."""
+        return self.id > other.id
+
+    def __lt__(self, other: "BuildInfo") -> bool:
+        """Return if we're smaller than other."""
+        return self.id < other.id
+
 
 class Drone:
     """Drone adapter.
@@ -83,25 +91,23 @@ def __init__(self, drone_url: str, drone_token: str) -> None:
         self.drone_url = drone_url
         self.drone_token = drone_token
 
-    def check_configuration(self) -> None:
-        """Check if the client is able to interact with the server.
+    def builds(self, project_pipeline: str) -> List[BuildInfo]:
+        """Return the builds of a project pipeline.
 
-        Makes sure that an API call works as expected.
+        Args:
+            project_pipeline: Drone pipeline identifier.
+                In the format of `repo_owner/repo_name`.
 
-        Raises:
-            DroneConfigurationError: if any of the checks fail.
+        Returns:
+            info: all builds information.
         """
-        try:
-            self.get(f"{self.drone_url}/api/user/repos")
-        except DroneAPIError as error:
-            log.error("Drone: KO")
-            raise DroneConfigurationError(
-                "There was a problem contacting the Drone server. \n\n"
-                "\t  Please make sure the DRONE_SERVER and DRONE_TOKEN "
-                "environmental variables are set. \n"
-                "\t  https://docs.drone.io/cli/configure/"
-            ) from error
-        log.info("Drone: OK")
+        build_history = self.get(
+            f"{self.drone_url}/api/repos/{project_pipeline}/builds"
+        ).json()
+
+        builds = [BuildInfo.from_kwargs(**build_data) for build_data in build_history]
+
+        return builds
 
     def build_info(self, project_pipeline: str, build_number: int) -> BuildInfo:
         """Return the information of the build.
@@ -149,13 +155,13 @@ def get(
                     response = requests.post(
                         url,
                         headers={"Authorization": f"Bearer {self.drone_token}"},
-                        timeout=10,
+                        timeout=2,
                     )
                 else:
                     response = requests.get(
                         url,
                         headers={"Authorization": f"Bearer {self.drone_token}"},
-                        timeout=10,
+                        timeout=2,
                     )
 
                 if response.status_code == 200:
@@ -170,19 +176,37 @@ def get(
             f"{response.status_code} error while trying to access {url}"
         )
 
+    def check_configuration(self) -> None:
+        """Check if the client is able to interact with the server.
+
+        Makes sure that an API call works as expected.
+
+        Raises:
+            DroneConfigurationError: if any of the checks fail.
+        """
+        try:
+            self.get(f"{self.drone_url}/api/user/repos")
+        except DroneAPIError as error:
+            log.error("Drone: KO")
+            raise DroneConfigurationError(
+                "There was a problem contacting the Drone server. \n\n"
+                "\t  Please make sure the DRONE_SERVER and DRONE_TOKEN "
+                "environmental variables are set. \n"
+                "\t  https://docs.drone.io/cli/configure/"
+            ) from error
+        log.info("Drone: OK")
+
     def last_build_info(self, project_pipeline: str) -> BuildInfo:
         """Return the information of the last build.
 
         Args:
             project_pipeline: Drone pipeline identifier.
                 In the format of `repo_owner/repo_name`.
+
         Returns:
             info: Last build information.
         """
-        build_data = self.get(
-            f"{self.drone_url}/api/repos/{project_pipeline}/builds"
-        ).json()[0]
-        return BuildInfo.from_kwargs(**build_data)
+        return self.builds(project_pipeline)[0]
 
     def last_success_build_info(
         self, project_pipeline: str, branch: str = "master"
@@ -197,17 +221,13 @@ def last_success_build_info(
         Returns:
             info: last successful build number information.
         """
-        build_history = self.get(
-            f"{self.drone_url}/api/repos/{project_pipeline}/builds"
-        ).json()
-
-        for build_data in build_history:
+        for build in self.builds(project_pipeline):
             if (
-                build_data["status"] == "success"
-                and build_data["target"] == branch
-                and build_data["event"] == "push"
+                build.status == "success"
+                and build.target == branch
+                and build.event == "push"
             ):
-                return BuildInfo.from_kwargs(**build_data)
+                return build
         raise DroneBuildError(
             f"There are no successful jobs with target branch {branch}"
         )

src/drode/entrypoints/cli.py

@@ -144,6 +144,37 @@ def status(ctx: Context) -> None:
         sys.exit(1)
 
 
+@cli.command()
+@click.option(
+    "-p",
+    "--pipeline",
+    type=str,
+    help="Analyze a pipeline that's not from the active project",
+)
+@click.option("-n", "--number_builds", type=int, help="Successful builds to analyze")
+@click.pass_context
+def time(
+    ctx: Context, pipeline: Optional[str] = None, number_builds: Optional[int] = None
+) -> None:
+    """Print the mean and standard deviation time of successful builds."""
+    try:
+        config = ctx.obj["config"]
+        if pipeline is None:
+            project_pipeline = config.get(
+                f"projects.{config['active_project']}.pipeline"
+            )
+        else:
+            project_pipeline = pipeline
+
+        pipeline_times = services.pipeline_times(
+            project_pipeline, ctx.obj["drone"], number_builds
+        )
+        views.print_times(project_pipeline, pipeline_times)
+    except ConfigError as error:
+        log.error(error)
+        sys.exit(1)
+
+
 @cli.command(hidden=True)
 def null() -> None:
     """Do nothing.

src/drode/services.py

@@ -6,7 +6,8 @@
 
 import logging
 import time
-from typing import Dict, Optional
+from math import sqrt
+from typing import Dict, Optional, Tuple
 
 from .adapters import Drone
 from .adapters.aws import AWS, AutoscalerInfo
@@ -190,3 +191,37 @@ def project_status(config: Config, aws: AWS) -> ProjectStatus:
         project[environment] = autoscaler_info
 
     return project
+
+
+PipelineTimes = Tuple[float, float, int]
+
+
+def pipeline_times(
+    project_pipeline: str, drone: Drone, number_builds: Optional[int] = None
+) -> PipelineTimes:
+    """Calculate the mean and standard deviation times of the successful builds.
+
+    Args:
+        config: Program configuration.
+        drone: Drone adapter.
+
+    Returns:
+        mean_time:
+        standard_deviation:
+            number_builds:
+    """
+    successful_builds = [
+        build for build in drone.builds(project_pipeline) if build.status == "success"
+    ]
+    if number_builds is not None:
+        successful_builds = successful_builds[:number_builds]
+
+    number_builds = len(successful_builds)
+    build_times = [build.finished - build.started for build in successful_builds]
+    mean_time = sum(build_times) / number_builds
+    standard_deviation = (
+        sqrt(sum((build_time - mean_time) ** 2 for build_time in build_times))
+        / number_builds
+    )
+
+    return mean_time, standard_deviation, number_builds

src/drode/views.py

@@ -6,7 +6,7 @@
 
 if TYPE_CHECKING:
     from drode.adapters.aws import AutoscalerInfo
-    from drode.services import ProjectStatus
+    from drode.services import PipelineTimes, ProjectStatus
 
 
 def print_autoscaling_group_info(autoscaler_info: "AutoscalerInfo") -> None:
@@ -25,3 +25,23 @@ def print_status(project_status: "ProjectStatus") -> None:
         print(f"# {environment}")
         print_autoscaling_group_info(autoscaler_info)
         print()
+
+
+def print_times(pipeline_name: str, pipeline_times: "PipelineTimes") -> None:
+    """Print the information of the pipeline times."""
+    mean_time, standard_deviation, number_builds = pipeline_times
+
+    print(f"Analyzing pipeline {pipeline_name}")
+    print(f"Using {number_builds} successful builds")
+    print(f"Mean build time: {_print_time(mean_time)}")
+    print(f"Standard deviation time: {_print_time(standard_deviation)}")
+
+
+def _print_time(time: float) -> str:
+    """Print the time with a nice format.
+
+    Examples:
+    >>> _print_time(63.1)
+    "01:03"
+    """
+    return f"{round(time)//60:02}:{round(time)%60:02}"

tests/conftest.py

@@ -1,10 +1,10 @@
 """Store the classes and fixtures used throughout the tests."""
 
 import os
+from pathlib import Path
 from shutil import copyfile
 
 import pytest
-from _pytest.tmpdir import TempPathFactory
 
 from drode.config import Config
 
@@ -15,12 +15,13 @@
 
 
 @pytest.fixture(name="config")
-def fixture_config(tmpdir_factory: TempPathFactory) -> Config:
+def fixture_config(tmp_path: Path) -> Config:
     """Configure the Config object for the tests."""
-    data = tmpdir_factory.mktemp("data")
-    config_file = str(data.join("config.yaml"))  # type: ignore
+    data = tmp_path / "data"
+    data.mkdir()
+    config_file = data / "config.yaml"
     copyfile("tests/assets/config.yaml", config_file)
-    config = Config(config_file)
+    config = Config(str(config_file))
 
     return config