# 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 abc import enum import hashlib import json import os from functools import cached_property import requests import structlog from libmozdata.phabricator import LintResult, UnitResult, UnitResultState from taskcluster.helper import TaskclusterConfig from code_review_bot.config import settings from code_review_bot.stats import InfluxDb from code_review_bot.tasks.base import AnalysisTask logger = structlog.get_logger(__name__) def positive_int(name, x): """Helper to get a positive integer or None""" if isinstance(x, int): if x >= 0: return x else: logger.warning(f"Negative {name} value found, defaults to None", value=x) return None class AnalysisException(Exception): """ Custom exception used in controlled errors """ def __init__(self, code, message): self.code = code super().__init__(message) class Level(enum.Enum): # A critical issue breaks CI and must always be reported Error = "error" # Warnings are reported when they are in patch Warning = "warning" class Issue(abc.ABC): """ Common reported issue interface """ revision = None def __init__( self, analyzer: AnalysisTask, revision, path: str, line: int, nb_lines: int, check: str, column: int = None, message: str = None, level: Level = Level.Warning, fix: str = None, language: str = None, ): # Check while avoiding circular dependencies from code_review_bot.revisions import Revision assert isinstance(revision, Revision) assert isinstance(analyzer, AnalysisTask) # Base required fields for all issues assert not os.path.isabs(path), f"Issue path can not be absolute {path}" self.revision = revision self.analyzer = analyzer self.check = check self.path = path self.line = positive_int("line", line) self.nb_lines = positive_int("nb_lines", nb_lines) # Support line 0 for full file issues like `source-test-mozlint-test-manifest`. if self.line == 0: logger.info("Line 0 is not supported, falling back to full file issue") self.line = None # Optional common fields self.column = column self.message = message self.level = level # Reserved payload for backend self.on_backend = None # Store information when a fix is available self.fix = fix self.language = language if self.fix is not None: assert self.language is not None, "Missing fix language" # Mark the issue as known by default, so only errors are reported # The before/after feature may tag some issues as new, so they are reported self.new_issue = False def __str__(self): line = f"line {self.line}" if self.line is not None else "full file" return f"{self.analyzer.name} issue {self.check}@{self.level.value} {self.path} {line}" @property def display_name(self): """ Issue's base name (by default analyzer's name) But can be overridden by subclasses """ return self.analyzer.display_name def build_extra_identifiers(self): """ Used to add information when building an issue unique hash """ return {} @property def allow_before_and_after_publish(self): """ Allow the possibility for an issue to avoid being published based on before/after. This allow publishing issues based on other criteria, like in_patch. """ if taskcluster.secrets.get( f"{self.analyzer.name.upper()}_DISABLE_PUBLICATION_BEFORE_AFTER", False ): return False return self.revision.before_after_feature def is_publishable(self): """ Is this issue publishable on reporters ? """ assert self.revision is not None, "Missing revision" # Always check specific rules validate if not self.validates(): return False if self.allow_before_and_after_publish: # Only publish new issues or issues inside the diff return self.new_issue or self.in_patch # An error is always published if self.level == Level.Error: return True # Then check if the backend marks this issue as publishable if self.on_backend is not None: return self.on_backend["publishable"] # Fallback to in_patch detection return self.in_patch @property def in_patch(self): return self.revision.contains(self) @cached_property def hash(self): """ Build a unique hash identifying that issue and cache the resulting value The text concerned by the issue is used and not its position in the file Message content is hashed as a single linter may return multiple issues on a single line We make the assumption that the message does not contain the line number If an error occurs reading the file content (locally or remotely), None is returned """ assert self.revision is not None, "Missing revision" # Build the hash only if the file is not autogenerated. # An autogenerated file resides in the build directory that it has the # format `obj-x86_64-pc-linux-gnu` file_content = None if "/obj-" not in self.path: if settings.mercurial_cache_checkout: logger.debug("Using the local repository to build issue's hash") try: with (settings.mercurial_cache_checkout / self.path).open() as f: file_content = f.read() except (FileNotFoundError, IsADirectoryError): logger.warning( "Failed to find issue's related file", path=self.path ) file_content = None else: try: # Load all the lines affected by the issue file_content = self.revision.load_file(self.path) except ValueError: # Build the hash with an empty content in case the path is erroneous file_content = None except requests.exceptions.HTTPError as e: if e.response.status_code == 404: logger.warning( "Failed to download a file with an issue", path=self.path ) # We still build the hash with empty content file_content = None else: # When encountering another HTTP error, raise the issue raise if file_content is None: self._hash = None return self._hash # Build raw content: # 1. lines affected by patch # 2. without any spaces around each line file_lines = file_content.splitlines() if self.line is None or self.nb_lines is None: # Use full file when line is not specified lines = file_lines else: # Use a range of lines start = self.line - 1 # file_lines start at 0, not 1 lines = file_lines[start : start + self.nb_lines] raw_content = "\n".join([line.strip() for line in lines]) # Build hash payload using issue data # excluding file position information (lines & char) extras = json.dumps(self.build_extra_identifiers(), sort_keys=True) payload = ":".join( [ self.analyzer.name, self.path, self.level.value, self.check, extras, raw_content, self.message, ] ).encode("utf-8") # Finally build the MD5 hash return hashlib.md5(payload).hexdigest() @abc.abstractmethod def validates(self): """ Is this issue publishable on reporters using IN_PATCH publication ? Should check specific rules and return a boolean """ raise NotImplementedError @abc.abstractmethod def as_text(self): """ Build the text content for reporters """ raise NotImplementedError @abc.abstractmethod def as_markdown(self): """ Build the Markdown content for debug email """ raise NotImplementedError def as_error(self): """ Build the Markdown content for for build error issues """ raise NotImplementedError def as_dict(self): """ Build the serializable dict representation of the issue Used by debugging tools """ issue_hash = None try: issue_hash = self.hash except Exception as e: logger.warn("Failed to build issue hash", error=str(e), issue=str(self)) return { "analyzer": self.analyzer.name, "path": self.path, "line": self.line, "nb_lines": self.nb_lines, "column": self.column, "check": self.check, "level": self.level.value, "message": self.message, "in_patch": self.in_patch, "validates": self.validates(), "publishable": self.is_publishable(), "hash": issue_hash, "fix": self.fix, } def as_phabricator_lint(self): """ Build the Phabricator LintResult instance """ # Add the level to the issue message if self.level == Level.Error: # We use the IMPORTANT red block silently prefix = "(IMPORTANT) ERROR:" else: prefix = "WARNING:" description = f"{prefix} {self.message}" # Add a fix when available # Prefix each line with 2 spaces as required by phabricator to trigger a code block # with syntax highlighting if self.fix is not None: fix = "\n".join(f" {line}" for line in self.fix.splitlines()) description += f"\n\n lang={self.language}\n{fix}" return LintResult( name=self.display_name, description=description, code=self.check, severity=self.level.value, path=self.path, # Report full file issues on line 1 line=self.line if self.line is not None else 1, char=self.column, ) def as_phabricator_unitresult(self): """ Build a Phabricator UnitResult for build errors """ assert ( self.is_build_error() ), "Only build errors may be published as unit results" return UnitResult( namespace="code-review", name="general", result=UnitResultState.Fail, details=f"Code review bot found a **build error**: \n{self.message}", format="remarkup", ) def is_build_error(self): """ Is this issue a build error? Default is False """ return False class Reliability(enum.Enum): Unknown = "unknown" High = "high" Medium = "medium" Low = "low" @property def invert(self): """ Verbalize the opposite of `value` of reliability to be used in coherent sentences. """ inversions = {"high": "low", "medium": "medium", "low": "high"} return inversions.get(self.value, "unknown") # Create common stats instance stats = InfluxDb() # Create common taskcluster config taskcluster = TaskclusterConfig("https://firefox-ci-tc.services.mozilla.com")