Quelle lint.py Sprache: Python

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import enum
from pathlib import Path
import re
import sys
from typing import (
    Any,
    Callable,
    Dict,
    Generator,
    List,
    Iterable,
    Optional,
    Tuple,
    Union,
)  # noqa

from . import metrics
from . import parser
from . import pings
from . import tags
from . import util

# Yield only an error message
LintGenerator = Generator[str, None, None]

# Yield fully constructed GlinterNits
NitGenerator = Generator["GlinterNit", None, None]

class CheckType(enum.Enum):
    warning = 0
    error = 1

def _split_words(name: str) -> List[str]:
    """
    Helper function to split words on either `.` or `_`.
    """
    return re.split("[._-]", name)

def _english_list(items: List[str]) -> str:
    """
    Helper function to format a list [A, B, C] as "'A', 'B', or 'C'".
    """
    if len(items) == 0:
        return ""
    elif len(items) == 1:
        return f"'{items[0]}'"
    else:
        return "{}, or '{}'".format(
            ", ".join([f"'{x}'" for x in items[:-1]]), items[-1]
        )

def _hamming_distance(str1: str, str2: str) -> int:
    """
    Count the # of differences between strings str1 and str2,
    padding the shorter one with whitespace
    """

    diffs = 0
    if len(str1) < len(str2):
        str1, str2 = str2, str1
    len_dist = len(str1) - len(str2)
    str2 += " " * len_dist

    for ch1, ch2 in zip(str1, str2):
        if ch1 != ch2:
            diffs += 1
    return diffs

def check_common_prefix(
    category_name: str, metrics: Iterable[metrics.Metric]
) -> LintGenerator:
    """
    Check if all metrics begin with a common prefix.
    """
    metric_words = sorted([_split_words(metric.name) for metric in metrics])

    if len(metric_words) < 2:
        return

    first = metric_words[0]
    last = metric_words[-1]

    for i in range(min(len(first), len(last))):
        if first[i] != last[i]:
            break

    if i > 0:
        common_prefix = "_".join(first[:i])
        yield (
            f"Within category '{category_name}', all metrics begin with "
            f"prefix '{common_prefix}'."
            "Remove the prefixes on the metric names and (possibly) "
            "rename the category."
        )

def check_unit_in_name(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    """
    The metric name ends in a unit.
    """
    TIME_UNIT_ABBREV = {
        "nanosecond": "ns",
        "microsecond": "us",
        "millisecond": "ms",
        "second": "s",
        "minute": "m",
        "hour": "h",
        "day": "d",
    }

    MEMORY_UNIT_ABBREV = {
        "byte": "b",
        "kilobyte": "kb",
        "megabyte": "mb",
        "gigabyte": "gb",
    }

    name_words = _split_words(metric.name)
    unit_in_name = name_words[-1]

    time_unit = getattr(metric, "time_unit", None)
    memory_unit = getattr(metric, "memory_unit", None)
    unit = getattr(metric, "unit", None)

    if time_unit is not None:
        if (
            unit_in_name == TIME_UNIT_ABBREV.get(time_unit.name)
            or unit_in_name == time_unit.name
        ):
            yield (
                f"Suffix '{unit_in_name}' is redundant with time_unit "
                f"'{time_unit.name}'. Only include time_unit."
            )
        elif (
            unit_in_name in TIME_UNIT_ABBREV.keys()
            or unit_in_name in TIME_UNIT_ABBREV.values()
        ):
            yield (
                f"Suffix '{unit_in_name}' doesn't match time_unit "
                f"'{time_unit.name}'. "
                "Confirm the unit is correct and only include time_unit."
            )

    elif memory_unit is not None:
        if (
            unit_in_name == MEMORY_UNIT_ABBREV.get(memory_unit.name)
            or unit_in_name == memory_unit.name
        ):
            yield (
                f"Suffix '{unit_in_name}' is redundant with memory_unit "
                f"'{memory_unit.name}'. "
                "Only include memory_unit."
            )
        elif (
            unit_in_name in MEMORY_UNIT_ABBREV.keys()
            or unit_in_name in MEMORY_UNIT_ABBREV.values()
        ):
            yield (
                f"Suffix '{unit_in_name}' doesn't match memory_unit "
                f"{memory_unit.name}'. "
                "Confirm the unit is correct and only include memory_unit."
            )

    elif unit is not None:
        if unit_in_name == unit:
            yield (
                f"Suffix '{unit_in_name}' is redundant with unit param "
                f"'{unit}'. "
                "Only include unit."
            )

def check_category_generic(
    category_name: str, metrics: Iterable[metrics.Metric]
) -> LintGenerator:
    """
    The category name is too generic.
    """
    GENERIC_CATEGORIES = ["metrics", "events"]

    if category_name in GENERIC_CATEGORIES:
        yield (
            f"Category '{category_name}' is too generic. "
            f"Don't use {_english_list(GENERIC_CATEGORIES)} for category names"
        )

def check_bug_number(
    metric: Union[metrics.Metric, pings.Ping], parser_config: Dict[str, Any]
) -> LintGenerator:
    number_bugs = [str(bug) for bug in metric.bugs if isinstance(bug, int)]

    if len(number_bugs):
        yield (
            f"For bugs {', '.join(number_bugs)}: "
            "Bug numbers are deprecated and should be changed to full URLs. "
            f"For example, use 'http://bugzilla.mozilla.org/{number_bugs[0]}' "
            f"instead of '{number_bugs[0]}'."
        )

def check_valid_in_baseline(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    allow_reserved = parser_config.get("allow_reserved", False)

    if not allow_reserved and "baseline" in metric.send_in_pings:
        yield (
            "The baseline ping is Glean-internal. "
            "Remove 'baseline' from the send_in_pings array."
        )

def check_misspelled_pings(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    for ping in metric.send_in_pings:
        for builtin in pings.RESERVED_PING_NAMES:
            distance = _hamming_distance(ping, builtin)
            if distance == 1:
                yield f"Ping '{ping}' seems misspelled. Did you mean '{builtin}'?"

def check_tags_required(
    metric_or_ping: Union[metrics.Metric, pings.Ping], parser_config: Dict[str, Any]
) -> LintGenerator:
    if parser_config.get("require_tags", False) and not len(
        metric_or_ping.metadata.get("tags", [])
    ):
        yield "Tags are required but no tags specified"

def check_user_lifetime_expiration(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    if metric.lifetime == metrics.Lifetime.user and metric.expires != "never":
        yield (
            "Metrics with 'user' lifetime cannot have an expiration date. "
            "They live as long as the user profile does. "
            "Set expires to 'never'."
        )

def check_expired_date(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    try:
        metric.validate_expires()
    except ValueError as e:
        yield (str(e))

def check_expired_metric(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    if metric.is_expired():
        yield ("Metric has expired. Please consider removing it.")

def check_old_event_api(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    # Glean v52.0.0 removed the old events API.
    # The metrics-2-0-0 schema still supports it.
    # We want to warn about it.
    # This can go when we introduce 3-0-0

    if not isinstance(metric, metrics.Event):
        return

    if not all("type" in x for x in metric.extra_keys.values()):
        yield ("The old event API is gone. Extra keys require a type.")

def check_metric_on_events_lifetime(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    """A non-event metric on the Events ping only makes sense if its value
    is immutable over the life of the ping."""
    if (
        "events" in metric.send_in_pings
        and "all_pings" not in metric.send_in_pings
        and metric.type != "event"
        and metric.lifetime == metrics.Lifetime.ping
    ):
        yield (
            "Non-event metrics sent on the Events ping should not have the ping"
            " lifetime."
        )

def check_unexpected_unit(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    """
    `unit` was allowed on all metrics and recently disallowed.
    We now warn about its use on all but quantity and custom distribution
    metrics.
    """
    allowed_types = [metrics.Quantity, metrics.CustomDistribution]
    if not any([isinstance(metric, ty) for ty in allowed_types]) and metric.unit:
        yield (
            "The `unit` property is only allowed for quantity "
            + "and custom distribution metrics."
        )

def check_empty_datareview(
    metric: metrics.Metric, parser_config: Dict[str, Any]
) -> LintGenerator:
    disallowed_datareview = ["", "todo"]
    data_reviews = [dr.lower() in disallowed_datareview for dr in metric.data_reviews]

    if any(data_reviews):
        yield "List of data reviews should not contain empty strings or TODO markers."

def check_redundant_ping(
    pings: pings.Ping, parser_config: Dict[str, Any]
) -> LintGenerator:
    """
    Check if the pings contains 'ping' as the prefix or suffix, or 'ping' or 'custom'
    """
    ping_words = _split_words(pings.name)

    if len(ping_words) != 0:
        ping_first_word = ping_words[0]
        ping_last_word = ping_words[-1]

        if ping_first_word == "ping":
            yield ("The prefix 'ping' is redundant.")
        elif ping_last_word == "ping":
            yield ("The suffix 'ping' is redundant.")
        elif "ping" in ping_words:
            yield ("The word 'ping' is redundant.")
        elif "custom" in ping_words:
            yield ("The word 'custom' is redundant.")

def check_unknown_ping(
    check_name: str,
    check_type: CheckType,
    all_pings: Dict[str, pings.Ping],
    metrics: Dict[str, metrics.Metric],
    parser_config: Dict[str, Any],
) -> NitGenerator:
    """
    Check that all pings in `send_in_pings` for all metrics are either a builtin ping
    or in the list of defined custom pings.
    """
    available_pings = [p for p in all_pings]

    for _, metric in metrics.items():
        if check_name in metric.no_lint:
            continue

        send_in_pings = metric.send_in_pings
        for target_ping in send_in_pings:
            if target_ping in pings.RESERVED_PING_NAMES:
                continue

            if target_ping not in available_pings:
                msg = f"Ping `{target_ping} `in `send_in_pings` is unknown."
                name = ".".join([metric.category, metric.name])
                nit = GlinterNit(
                    check_name,
                    name,
                    msg,
                    check_type,
                )
                yield nit

def check_name_too_similar(
    check_name: str,
    check_type: CheckType,
    all_pings: Dict[str, pings.Ping],
    all_metrics: Dict[str, metrics.Metric],
    parser_config: Dict[str, Any],
) -> NitGenerator:
    """
    Check that all metrics identifiers are suitably distinct.
    Require that at least n-1 of the similarly-named metrics must be no_lint'd to dismiss the lint.

    Current similarity test: the fully-qualified identifier differs solely in punctuation.
    e.g. formautofill.credit_cards and formautofill.creditcards
    """
    seen_metrics: Dict[str, metrics.Metric] = dict()

    for _, metric in all_metrics.items():
        if check_name in metric.no_lint:
            continue

        no_punc = metric.identifier().replace("_", "").replace(".", "")
        if no_punc in seen_metrics:
            msg = f"Metric `{metric.identifier()}`'s name is too similar to existing metric `{seen_metrics[no_punc].identifier()}`"
            nit = GlinterNit(check_name, metric.identifier(), msg, check_type)
            yield nit

        seen_metrics[no_punc] = metric

# The checks that operate on an entire category of metrics:
#    {NAME: (function, is_error)}
CATEGORY_CHECKS: Dict[
    str, Tuple[Callable[[str, Iterable[metrics.Metric]], LintGenerator], CheckType]
] = {
    "COMMON_PREFIX": (check_common_prefix, CheckType.error),
    "CATEGORY_GENERIC": (check_category_generic, CheckType.error),
}

# The checks that operate on individual metrics:
#     {NAME: (function, is_error)}
METRIC_CHECKS: Dict[
    str, Tuple[Callable[[metrics.Metric, dict], LintGenerator], CheckType]
] = {
    "UNIT_IN_NAME": (check_unit_in_name, CheckType.error),
    "BUG_NUMBER": (check_bug_number, CheckType.error),
    "BASELINE_PING": (check_valid_in_baseline, CheckType.error),
    "MISSPELLED_PING": (check_misspelled_pings, CheckType.error),
    "TAGS_REQUIRED": (check_tags_required, CheckType.error),
    "EXPIRATION_DATE_TOO_FAR": (check_expired_date, CheckType.warning),
    "USER_LIFETIME_EXPIRATION": (check_user_lifetime_expiration, CheckType.warning),
    "EXPIRED": (check_expired_metric, CheckType.warning),
    "OLD_EVENT_API": (check_old_event_api, CheckType.warning),
    "METRIC_ON_EVENTS_LIFETIME": (check_metric_on_events_lifetime, CheckType.error),
    "UNEXPECTED_UNIT": (check_unexpected_unit, CheckType.warning),
    "EMPTY_DATAREVIEW": (check_empty_datareview, CheckType.warning),
}

# The checks that operate on individual pings:
#     {NAME: (function, is_error)}
PING_CHECKS: Dict[
    str, Tuple[Callable[[pings.Ping, dict], LintGenerator], CheckType]
] = {
    "BUG_NUMBER": (check_bug_number, CheckType.error),
    "TAGS_REQUIRED": (check_tags_required, CheckType.error),
    "REDUNDANT_PING": (check_redundant_ping, CheckType.error),
}

ALL_OBJECT_CHECKS: Dict[
    str,
    Tuple[
        Callable[
            # check name, check type, pings, metrics, config
            [str, CheckType, dict, dict, dict],
            NitGenerator,
        ],
        CheckType,
    ],
] = {
    "UNKNOWN_PING_REFERENCED": (check_unknown_ping, CheckType.error),
    "NAME_TOO_SIMILAR": (check_name_too_similar, CheckType.error),
}

class GlinterNit:
    def __init__(self, check_name: str, name: str, msg: str, check_type: CheckType):
        self.check_name = check_name
        self.name = name
        self.msg = msg
        self.check_type = check_type

    def format(self):
        return (
            f"{self.check_type.name.upper()}: {self.check_name}: "
            f"{self.name}: {self.msg}"
        )

def _lint_item_tags(
    item_name: str,
    item_type: str,
    item_tag_names: List[str],
    valid_tag_names: List[str],
) -> List[GlinterNit]:
    invalid_tags = [tag for tag in item_tag_names if tag not in valid_tag_names]
    return (
        [
            GlinterNit(
                "INVALID_TAGS",
                item_name,
                f"Invalid tags specified in {item_type}: {', '.join(invalid_tags)}",
                CheckType.error,
            )
        ]
        if len(invalid_tags)
        else []
    )

def _lint_pings(
    category: Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag]],
    parser_config: Dict[str, Any],
    valid_tag_names: List[str],
) -> List[GlinterNit]:
    nits: List[GlinterNit] = []

    for ping_name, ping in sorted(list(category.items())):
        assert isinstance(ping, pings.Ping)
        for check_name, (check_func, check_type) in PING_CHECKS.items():
            new_nits = list(check_func(ping, parser_config))
            if len(new_nits):
                if check_name not in ping.no_lint:
                    nits.extend(
                        GlinterNit(
                            check_name,
                            ping_name,
                            msg,
                            check_type,
                        )
                        for msg in new_nits
                    )
        nits.extend(
            _lint_item_tags(
                ping_name,
                "ping",
                ping.metadata.get("tags", []),
                valid_tag_names,
            )
        )
    return nits

def _lint_all_objects(
    objects: Dict[str, Dict[str, Union[metrics.Metric, pings.Ping, tags.Tag]]],
    parser_config: Dict[str, Any],
) -> List[GlinterNit]:
    nits: List[GlinterNit] = []

    pings = objects.get("pings")
    if not pings:
        return []

    metrics = objects.get("all_metrics")
    if not metrics:
        return []

    for check_name, (check_func, check_type) in ALL_OBJECT_CHECKS.items():
        new_nits = list(
            check_func(check_name, check_type, pings, metrics, parser_config)
        )
        nits.extend(new_nits)

    return nits

def lint_metrics(
    objs: metrics.ObjectTree,
    parser_config: Optional[Dict[str, Any]] = None,
    file=sys.stderr,
) -> List[GlinterNit]:
    """
    Performs glinter checks on a set of metrics objects.

    :param objs: Tree of metric objects, as returns by `parser.parse_objects`.
    :param file: The stream to write errors to.
    :returns: List of nits.
    """
    if parser_config is None:
        parser_config = {}

    nits: List[GlinterNit] = []
    valid_tag_names = [tag for tag in objs.get("tags", [])]

    nits.extend(_lint_all_objects(objs, parser_config))

    for category_name, category in sorted(list(objs.items())):
        if category_name == "pings":
            nits.extend(_lint_pings(category, parser_config, valid_tag_names))
            continue

        if category_name == "tags":
            # currently we have no linting for tags
            continue

        # Make sure the category has only Metrics, not Pings or Tags
        category_metrics = dict(
            (name, metric)
            for (name, metric) in category.items()
            if isinstance(metric, metrics.Metric)
        )

        for cat_check_name, (cat_check_func, check_type) in CATEGORY_CHECKS.items():
            if any(
                cat_check_name in metric.no_lint for metric in category_metrics.values()
            ):
                continue
            nits.extend(
                GlinterNit(cat_check_name, category_name, msg, check_type)
                for msg in cat_check_func(category_name, category_metrics.values())
            )

        for _metric_name, metric in sorted(list(category_metrics.items())):
            for check_name, (check_func, check_type) in METRIC_CHECKS.items():
                new_nits = list(check_func(metric, parser_config))
                if len(new_nits):
                    if check_name not in metric.no_lint:
                        nits.extend(
                            GlinterNit(
                                check_name,
                                ".".join([metric.category, metric.name]),
                                msg,
                                check_type,
                            )
                            for msg in new_nits
                        )

            # also check that tags for metric are valid
            nits.extend(
                _lint_item_tags(
                    ".".join([metric.category, metric.name]),
                    "metric",
                    metric.metadata.get("tags", []),
                    valid_tag_names,
                )
            )

    if len(nits):
        print("Sorry, Glean found some glinter nits:", file=file)
        for nit in nits:
            print(nit.format(), file=file)
        print("", file=file)
        print("Please fix the above nits to continue.", file=file)
        print(
            "To disable a check, add a `no_lint` parameter "
            "with a list of check names to disable.\n"
            "This parameter can appear with each individual metric, or at the "
            "top-level to affect the entire file.",
            file=file,
        )

    return nits

def glinter(
    input_filepaths: Iterable[Path],
    parser_config: Optional[Dict[str, Any]] = None,
    file=sys.stderr,
) -> int:
    """
    Commandline helper for glinter.

    :param input_filepaths: List of Path objects to load metrics from.
    :param parser_config: Parser configuration object, passed to
      `parser.parse_objects`.
    :param file: The stream to write the errors to.
    :return: Non-zero if there were any glinter errors.
    """
    if parser_config is None:
        parser_config = {}

    errors = 0

    objs = parser.parse_objects(input_filepaths, parser_config)
    errors += util.report_validation_errors(objs)

    nits = lint_metrics(objs.value, parser_config=parser_config, file=file)
    errors += len([nit for nit in nits if nit.check_type == CheckType.error])

    if errors == 0:
        print("✨ Your metrics are Glean! ✨", file=file)
        return 0

    print(f"❌ Found {errors} errors.")

    return 1

quality79%

¤ Dauer der Verarbeitung: 0.27 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.