panaetius/duties.py

from __future__ import annotations

import importlib
import os
import pathlib
import re
import shutil
import sys
from io import StringIO

from duty import duty

PACKAGE_NAME = "panaetius"


@duty
def update_deps(ctx, dry: bool = False):
    """
    Update the dependencies using Poetry.

    Args:
        ctx: The context instance (passed automatically).
        dry (bool, optional) = If True will update the `poetry.lock` without updating the
            dependencies themselves. Defaults to False.

    Example:
        `duty update_deps dry=False`
    """
    dry_run = "--dry-run" if dry else ""
    ctx.run(
        ["poetry", "update", dry_run],
        title=f"Updating poetry deps {dry_run}",
    )


@duty
def test(ctx):
    """
    Run tests using pytest.

    Args:
        ctx: The context instance (passed automatically).
    """
    pytest_results = ctx.run(["pytest", "-v"], pty=True)
    print(pytest_results)


@duty
def coverage(ctx):
    """
    Generate a coverage report and save to XML and HTML.

    Args:
        ctx: The context instance (passed automatically).

    Example:
        `duty coverage`
    """
    ctx.run(["coverage", "run", "--source", PACKAGE_NAME, "-m", "pytest"])
    res = ctx.run(["coverage", "report"], pty=True)
    print(res)
    ctx.run(["coverage", "html"])
    ctx.run(["coverage", "xml"])


@duty
def version(ctx, bump: str = "patch"):
    """
    Bump the version using Poetry and update _version.py.

    Args:
        ctx: The context instance (passed automatically).
        bump (str, optional) = poetry version flag. Available options are:
            patch, minor, major, prepatch, preminor, premajor, prerelease.
            Defaults to patch.

    Example:
        `duty version bump=major`
    """

    # bump with poetry
    result = ctx.run(["poetry", "version", bump])
    new_version = re.search(r"(?:.*)(?:\s)(\d+\.\d+\.\d+)$", result)
    print(new_version.group(0))

    # update _version.py
    version_file = pathlib.Path(PACKAGE_NAME) / "_version.py"
    with version_file.open("w", encoding="utf-8") as version_file:
        version_file.write(
            f'"""Module containing the version of {PACKAGE_NAME}."""\n\n' + f'__version__ = "{new_version.group(1)}"\n'
        )
    print(f"Bumped _version.py to {new_version.group(1)}")


@duty
def build(ctx):
    """
    Build with poetry and extract the setup.py and copy to project root.

    Args:
        ctx: The context instance (passed automatically).

    Example:
        `duty build`
    """

    repo_root = pathlib.Path(".")

    # build with poetry
    result = ctx.run(["poetry", "build"])
    print(result)

    # extract the setup.py from the tar
    extracted_tar = re.search(r"(?:.*)(?:Built\s)(.*)", result)
    tar_file = pathlib.Path(f"./dist/{extracted_tar.group(1)}")
    shutil.unpack_archive(tar_file, tar_file.parents[0])

    # copy setup.py to repo root
    extracted_path = tar_file.parents[0] / os.path.splitext(tar_file.stem)[0]
    setup_py = extracted_path / "setup.py"
    shutil.copyfile(setup_py, (repo_root / "setup.py"))

    # cleanup
    shutil.rmtree(extracted_path)


@duty
def export(ctx):
    """
    Export the dependencies to a requirements.txt file.

    Args:
        ctx: The context instance (passed automatically).

    Example:
        `duty export`
    """
    requirements_content = ctx.run(
        [
            "poetry",
            "export",
            "-f",
            "requirements.txt",
            "--without-hashes",
        ]
    )
    requirements_dev_content = ctx.run(
        [
            "poetry",
            "export",
            "-f",
            "requirements.txt",
            "--without-hashes",
            "--dev",
        ]
    )

    requirements = pathlib.Path(".") / "requirements.txt"
    requirements_dev = pathlib.Path(".") / "requirements_dev.txt"

    with requirements.open("w", encoding="utf-8") as req:
        req.write(requirements_content)

    with requirements_dev.open("w", encoding="utf-8") as req:
        req.write(requirements_dev_content)


@duty
def publish(ctx, password: str):
    """
    Publish the package to pypi.org.

    Args:
        ctx: The context instance (passed automatically).
        password (str): pypi.org password.

    Example:
        `duty publish password=$my_password`
    """
    dist_dir = pathlib.Path(".") / "dist"
    rm_result = rm_tree(dist_dir)
    print(rm_result)

    publish_result = ctx.run(["poetry", "publish", "-u", "dtomlinson", "-p", password, "--build"])
    print(publish_result)


@duty(silent=True)
def clean(ctx):
    """
    Delete temporary files.

    Args:
        ctx: The context instance (passed automatically).
    """
    ctx.run("rm -rf .mypy_cache")
    ctx.run("rm -rf .pytest_cache")
    ctx.run("rm -rf tests/.pytest_cache")
    ctx.run("rm -rf build")
    ctx.run("rm -rf dist")
    ctx.run("rm -rf pip-wheel-metadata")
    ctx.run("rm -rf site")
    ctx.run("rm -rf coverage.xml")
    ctx.run("rm -rf pytest.xml")
    ctx.run("rm -rf htmlcov")
    ctx.run("find . -iname '.coverage*' -not -name .coveragerc | xargs rm -rf")
    ctx.run("find . -type d -name __pycache__ | xargs rm -rf")
    ctx.run("find . -name '*.rej' -delete")


@duty
def format(ctx):
    """
    Format code using Black and isort.

    Args:
        ctx: The context instance (passed automatically).
    """
    res = ctx.run(["black", "--line-length=99", PACKAGE_NAME], pty=True, title="Running Black")
    print(res)

    res = ctx.run(["isort", PACKAGE_NAME])
    print(res)


@duty(pre=["check_code_quality", "check_types", "check_docs", "check_dependencies"])
def check(ctx):
    """
    Check the code quality, check types, check documentation builds and check dependencies for vulnerabilities.

    Args:
        ctx: The context instance (passed automatically).
    """


@duty
def check_code_quality(ctx):
    """
    Check the code quality using prospector.

    Args:
        ctx: The context instance (passed automatically).
    """
    ctx.run(["prospector", PACKAGE_NAME], pty=True, title="Checking code quality with prospector")


@duty
def check_types(ctx):
    """
    Check the types using mypy.

    Args:
        ctx: The context instance (passed automatically).
    """
    ctx.run(["mypy", PACKAGE_NAME], pty=True, title="Checking types with MyPy")


@duty
def check_docs(ctx):
    """
    Check the documentation builds successfully.

    Args:
        ctx: The context instance (passed automatically).
    """
    ctx.run(["mkdocs", "build"], title="Building documentation")


@duty
def check_dependencies(ctx):
    """
    Check dependencies with safety for vulnerabilities.

    Args:
        ctx: The context instance (passed automatically).
    """
    for module in sys.modules:
        if module.startswith("safety.") or module == "safety":
            del sys.modules[module]

    importlib.invalidate_caches()

    from safety import safety
    from safety.formatter import report
    from safety.util import read_requirements

    requirements = ctx.run(
        "poetry export --dev --without-hashes",
        title="Exporting dependencies as requirements",
        allow_overrides=False,
    )

    def check_vulns():
        packages = list(read_requirements(StringIO(requirements)))
        vulns = safety.check(packages=packages, ignore_ids="41002", key="", db_mirror="", cached=False, proxy={})
        output_report = report(vulns=vulns, full=True, checked_packages=len(packages))
        print(vulns)
        if vulns:
            print(output_report)

    ctx.run(
        check_vulns,
        stdin=requirements,
        title="Checking dependencies",
        pty=True,
    )


def rm_tree(directory: pathlib.Path):
    """
    Recursively delete a directory and all its contents.

    Args:
        directory (pathlib.Path): The directory to delete.
    """
    for child in directory.glob("*"):
        if child.is_file():
            child.unlink()
        else:
            rm_tree(child)
    directory.rmdir()