Improve typing of LanguageProcessors (#666)

Author: Swatinem

services/report/languages/__init__.py

@@ -1,29 +0,0 @@
-# ruff: noqa: F401
-from services.report.languages.bullseye import BullseyeProcessor
-from services.report.languages.clover import CloverProcessor
-from services.report.languages.cobertura import CoberturaProcessor
-from services.report.languages.coveralls import CoverallsProcessor
-from services.report.languages.csharp import CSharpProcessor
-from services.report.languages.dlst import DLSTProcessor
-from services.report.languages.elm import ElmProcessor
-from services.report.languages.flowcover import FlowcoverProcessor
-from services.report.languages.gap import GapProcessor
-from services.report.languages.gcov import GcovProcessor
-from services.report.languages.go import GoProcessor
-from services.report.languages.jacoco import JacocoProcessor
-from services.report.languages.jetbrainsxml import JetBrainsXMLProcessor
-from services.report.languages.lcov import LcovProcessor
-from services.report.languages.lua import LuaProcessor
-from services.report.languages.mono import MonoProcessor
-from services.report.languages.node import NodeProcessor
-from services.report.languages.pycoverage import PyCoverageProcessor
-from services.report.languages.rlang import RlangProcessor
-from services.report.languages.salesforce import SalesforceProcessor
-from services.report.languages.scala import ScalaProcessor
-from services.report.languages.scoverage import SCoverageProcessor
-from services.report.languages.simplecov import SimplecovProcessor
-from services.report.languages.v1 import VOneProcessor
-from services.report.languages.vb import VbProcessor
-from services.report.languages.vb2 import VbTwoProcessor
-from services.report.languages.xcode import XCodeProcessor
-from services.report.languages.xcodeplist import XCodePlistProcessor

services/report/languages/base.py

@@ -1,4 +1,3 @@
-import typing
 from typing import Any
 
 from shared.reports.resources import Report
@@ -7,10 +6,6 @@
 
 
 class BaseLanguageProcessor(object):
-    @property
-    def name(self):
-        return self.get_processor_name()
-
     def __init__(self, *args, **kwargs) -> None:
         pass
 
@@ -37,9 +32,7 @@ def matches_content(self, content: Any, first_line: str, name: str) -> bool:
         """
         pass
 
-    def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
-    ) -> Report:
+    def process(self, name: str, content: Any, report_builder: ReportBuilder) -> Report:
         """Processes a report uploaded by the user, returning a `Report`
 
         This is the base function which we need to implement
@@ -61,7 +54,3 @@ def process(
             ReportExpiredException: If the report is considered expired
         """
         pass
-
-    @classmethod
-    def get_processor_name(cls) -> str:
-        return cls.__name__

services/report/languages/bullseye.py

@@ -1,6 +1,6 @@
-import typing
 from xml.etree.ElementTree import Element
 
+import sentry_sdk
 from shared.reports.resources import Report
 from timestring import Date
 
@@ -18,17 +18,17 @@ class BullseyeProcessor(BaseLanguageProcessor):
     def matches_content(self, content: Element, first_line: str, name: str) -> bool:
         return "BullseyeCoverage" in content.tag
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: Element, report_builder: ReportBuilder
     ) -> Report:
         return from_xml(content, report_builder.create_report_builder_session(name))
 
 
-def from_xml(xml, report_builder_session: ReportBuilderSession):
-    path_fixer, ignored_lines, sessionid, yaml = (
+def from_xml(xml: Element, report_builder_session: ReportBuilderSession):
+    path_fixer, ignored_lines, yaml = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
         report_builder_session.current_yaml,
     )
     if read_yaml_field(yaml, ("codecov", "max_report_age"), "12h ago"):

services/report/languages/clover.py

@@ -1,5 +1,6 @@
-import typing
+from xml.etree.ElementTree import Element
 
+import sentry_sdk
 from shared.reports.resources import Report
 from timestring import Date
 
@@ -14,11 +15,12 @@
 
 
 class CloverProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
-        return bool(content.tag == "coverage" and content.attrib.get("generated"))
+    def matches_content(self, content: Element, first_line: str, name: str) -> bool:
+        return content.tag == "coverage" and bool(content.attrib.get("generated"))
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: Element, report_builder: ReportBuilder
     ) -> Report:
         report_builder_session = report_builder.create_report_builder_session(name)
         return from_xml(content, report_builder_session)
@@ -38,11 +40,10 @@ def get_end_of_file(filename, xmlfile):
                 pass
 
 
-def from_xml(xml, report_builder_session: ReportBuilderSession) -> Report:
-    path_fixer, ignored_lines, sessionid, yaml = (
+def from_xml(xml: Element, report_builder_session: ReportBuilderSession) -> Report:
+    path_fixer, ignored_lines, yaml = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
         report_builder_session.current_yaml,
     )
 

services/report/languages/cobertura.py

@@ -1,8 +1,9 @@
 import logging
 import re
-import typing
 from typing import List
+from xml.etree.ElementTree import Element
 
+import sentry_sdk
 from shared.reports.resources import Report
 from timestring import Date, TimestringInvalid
 
@@ -19,13 +20,15 @@
 
 
 class CoberturaProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
-        if bool(list(content.iter("coverage"))):
-            return True
-        return bool(list(content.iter("scoverage")))
+    def matches_content(self, content: Element, first_line: str, name: str) -> bool:
+        return bool(
+            next(content.iter("coverage"), None)
+            or next(content.iter("scoverage"), None)
+        )
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: Element, report_builder: ReportBuilder
     ) -> Report:
         report_builder_session = report_builder.create_report_builder_session(name)
         return from_xml(content, report_builder_session)
@@ -43,11 +46,10 @@ def get_sources_to_attempt(xml) -> List[str]:
     return [s for s in sources if isinstance(s, str) and s.startswith("/")]
 
 
-def from_xml(xml, report_builder_session: ReportBuilderSession) -> Report:
-    path_fixer, ignored_lines, sessionid, repo_yaml = (
+def from_xml(xml: Element, report_builder_session: ReportBuilderSession) -> Report:
+    path_fixer, ignored_lines, repo_yaml = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
         report_builder_session.current_yaml,
     )
 

services/report/languages/coveralls.py

@@ -1,5 +1,4 @@
-import typing
-
+import sentry_sdk
 from shared.reports.resources import Report
 
 from services.report.languages.base import BaseLanguageProcessor
@@ -11,26 +10,21 @@
 
 
 class CoverallsProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
-        return detect(content)
+    def matches_content(self, content: dict, first_line: str, name: str) -> bool:
+        return "source_files" in content
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: dict, report_builder: ReportBuilder
     ) -> Report:
         return from_json(content, report_builder.create_report_builder_session(name))
 
 
-def detect(report):
-    return "source_files" in report
-
-
-def from_json(report, report_builder_session: ReportBuilderSession) -> Report:
+def from_json(report: dict, report_builder_session: ReportBuilderSession) -> Report:
     # https://github.com/codecov/support/issues/253
-    path_fixer, ignored_lines, sessionid, repo_yaml = (
+    path_fixer, ignored_lines = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
-        report_builder_session.current_yaml,
     )
     for _file in report["source_files"]:
         filename = path_fixer(_file["name"])

services/report/languages/csharp.py

@@ -1,7 +1,8 @@
-import typing
 from collections import defaultdict
 from itertools import repeat
+from xml.etree.ElementTree import Element
 
+import sentry_sdk
 from shared.reports.resources import Report
 
 from services.report.languages.base import BaseLanguageProcessor
@@ -13,11 +14,12 @@
 
 
 class CSharpProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
-        return bool(content.tag == "CoverageSession")
+    def matches_content(self, content: Element, first_line: str, name: str) -> bool:
+        return content.tag == "CoverageSession"
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: Element, report_builder: ReportBuilder
     ) -> Report:
         report_builder_session = report_builder.create_report_builder_session(name)
         return from_xml(content, report_builder_session)
@@ -38,7 +40,7 @@ def _build_branches(branch_gen):
     return branches
 
 
-def from_xml(xml, report_builder_session: ReportBuilderSession) -> Report:
+def from_xml(xml: Element, report_builder_session: ReportBuilderSession) -> Report:
     """
     https://github.com/OpenCover/opencover/issues/293#issuecomment-94598145
     @sl - start line
@@ -50,13 +52,10 @@ def from_xml(xml, report_builder_session: ReportBuilderSession) -> Report:
     @vc - statement executed
     <SequencePoint vc="2" uspid="3" ordinal="0" offset="0" sl="35" sc="8" el="35" ec="9" bec="0" bev="0" fileid="1" />
     """
-    ignored_lines, sessionid = (
-        report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
-    )
+    ignored_lines = report_builder_session.ignored_lines
+
     # dict of {"fileid": "path"}
     file_by_id = {}
-    file_by_id_get = file_by_id.get
     file_by_name = {None: None}
     for f in xml.iter("File"):
         filename = report_builder_session.path_fixer(

services/report/languages/dlst.py

@@ -1,6 +1,6 @@
-import typing
 from io import BytesIO
 
+import sentry_sdk
 from shared.reports.resources import Report
 from shared.reports.types import ReportLine
 
@@ -12,16 +12,17 @@
 
 
 class DLSTProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
-        return bool(content[-7:] == b"covered")
+    def matches_content(self, content: bytes, first_line: str, name: str) -> bool:
+        return content[-7:] == b"covered"
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: bytes, report_builder: ReportBuilder
     ) -> Report:
         return from_string(content, report_builder.create_report_builder_session(name))
 
 
-def from_string(string, report_builder_session: ReportBuilderSession) -> Report:
+def from_string(string: bytes, report_builder_session: ReportBuilderSession) -> Report:
     path_fixer, ignored_lines, sessionid, filename = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,

services/report/languages/elm.py

@@ -1,5 +1,4 @@
-import typing
-
+import sentry_sdk
 from shared.reports.resources import Report
 
 from services.report.languages.base import BaseLanguageProcessor
@@ -11,21 +10,21 @@
 
 
 class ElmProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
+    def matches_content(self, content: dict, first_line: str, name: str) -> bool:
         return isinstance(content, dict) and bool(content.get("coverageData"))
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: dict, report_builder: ReportBuilder
     ) -> Report:
         report_builder_session = report_builder.create_report_builder_session(name)
         return from_json(content, report_builder_session)
 
 
-def from_json(json, report_builder_session: ReportBuilderSession) -> Report:
-    path_fixer, ignored_lines, sessionid = (
+def from_json(json: dict, report_builder_session: ReportBuilderSession) -> Report:
+    path_fixer, ignored_lines = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
     )
     for name, data in json["coverageData"].items():
         fn = path_fixer(json["moduleMap"][name])

services/report/languages/flowcover.py

@@ -1,5 +1,4 @@
-import typing
-
+import sentry_sdk
 from shared.reports.resources import Report
 
 from services.report.languages.base import BaseLanguageProcessor
@@ -11,23 +10,23 @@
 
 
 class FlowcoverProcessor(BaseLanguageProcessor):
-    def matches_content(self, content, first_line, name):
+    def matches_content(self, content: dict, first_line: str, name: str) -> bool:
         return isinstance(content, dict) and bool(content.get("flowStatus"))
 
+    @sentry_sdk.trace
     def process(
-        self, name: str, content: typing.Any, report_builder: ReportBuilder
+        self, name: str, content: dict, report_builder: ReportBuilder
     ) -> Report:
         report_builder_session = report_builder.create_report_builder_session(
             filepath=name
         )
         return from_json(content, report_builder_session)
 
 
-def from_json(json, report_builder_session: ReportBuilderSession) -> Report:
-    path_fixer, ignored_lines, sessionid = (
+def from_json(json: dict, report_builder_session: ReportBuilderSession) -> Report:
+    path_fixer, ignored_lines = (
         report_builder_session.path_fixer,
         report_builder_session.ignored_lines,
-        report_builder_session.sessionid,
     )
 
     for fn, data in json["files"].items():