danger.py

import json
import sys
from functools import partial
from operator import attrgetter
from typing import Any, Callable, Dict, List, Optional

from pydantic import BaseModel

from .models import (
    BitBucketCloudJSONDSL,
    BitBucketServerJSONDSL,
    DangerDSLJSONType,
    DangerDSLJSONTypeSettings,
    GitHubDSL,
    GitJSONDSL,
    GitLabDSL,
)


class Violation(BaseModel):
    message: str
    file_name: Optional[str] = None
    line: Optional[int] = None

    class Config:
        allow_population_by_field_name = True
        fields = {"message": "message", "file_name": "file", "line": "line"}


class DangerMeta(BaseModel):
    href: str = "https://danger.systems/python"
    name: str = "danger-python"

    class Config:
        fields = {"href": "runtimeHref", "name": "runtimeName"}


class DangerResults(BaseModel):
    fails: List[Violation] = []
    warnings: List[Violation] = []
    messages: List[Violation] = []
    markdowns: List[Violation] = []
    meta: DangerMeta = DangerMeta()


def serialize_violation(violation: Violation) -> Dict[str, Any]:
    return violation.dict(exclude_none=True, by_alias=True)


def serialize_results(results: DangerResults) -> Dict[str, Any]:
    return results.dict(exclude_none=True, by_alias=True)


def load_dsl() -> DangerDSLJSONType:
    file_url = sys.stdin.read().split("danger://dsl/")[-1]

    with open(file_url, "r") as json_file:
        input_json = json.loads(json_file.read())
        return DangerDSLJSONType(**input_json["danger"])


class Danger:
    def __init__(self):
        if not Danger.dsl:
            Danger.dsl = load_dsl()
        if not Danger.results:
            Danger.results = DangerResults()

    dsl: Optional[DangerDSLJSONType] = None
    results: Optional[DangerResults] = None

    @property
    def git(self) -> Optional[GitJSONDSL]:
        assert Danger.dsl is not None
        return Danger.dsl.git

    @property
    def github(self) -> Optional[GitHubDSL]:
        assert Danger.dsl is not None
        return Danger.dsl.github

    @property
    def bitbucket_cloud(self) -> Optional[BitBucketCloudJSONDSL]:
        assert Danger.dsl is not None
        return Danger.dsl.bitbucket_cloud

    @property
    def bitbucket_server(self) -> Optional[BitBucketServerJSONDSL]:
        assert Danger.dsl is not None
        return Danger.dsl.bitbucket_server

    @property
    def gitlab(self) -> Optional[GitLabDSL]:
        assert Danger.dsl is not None
        return Danger.dsl.gitlab

    @property
    def settings(self) -> Optional[DangerDSLJSONTypeSettings]:
        assert Danger.dsl is not None
        return Danger.dsl.settings


def _add_to_results(
    field: Callable[[DangerResults], List[Violation]],
    message: str,
    file_name: Optional[str] = None,
    line: Optional[int] = None,
):
    assert Danger.results is not None
    violation = Violation(message=message, file_name=file_name, line=line)
    field(Danger.results).append(violation)


message = partial(_add_to_results, attrgetter("messages"))
markdown = partial(_add_to_results, attrgetter("markdowns"))
warn = partial(_add_to_results, attrgetter("warnings"))
fail = partial(_add_to_results, attrgetter("fails"))