refactor(context): add ConnectorType enum, python_sources/trees on context, severity helpers

Author: jabesq

shared/connector_linter/connector_linter/formatters.py

@@ -4,9 +4,8 @@
 from pathlib import Path
 from typing import TextIO
 
-from connector_linter.models import CheckResult, Severity
+from connector_linter.models import SEVERITY_COLOR, CheckResult, Severity
 
-# ANSI color codes
 _COLORS = {
     "green": "\033[32m",
     "red": "\033[31m",
@@ -19,17 +18,30 @@
 
 
 def _use_color(stream: TextIO) -> bool:
-    """Determine if we should use ANSI colors on this stream."""
     return hasattr(stream, "isatty") and stream.isatty()
 
 
 def _c(text: str, color: str, stream: TextIO) -> str:
-    """Colorize text if the stream supports it."""
     if not _use_color(stream):
         return text
     return f"{_COLORS.get(color, '')}{text}{_COLORS['reset']}"
 
 
+def _group_results(
+    results: list[CheckResult],
+) -> tuple[list[CheckResult], list[CheckResult], list[CheckResult]]:
+    """Return (failed, advisory, passed_normal).
+
+    - failed       — ``passed=False``, any severity
+    - advisory     — ``passed=True`` + ``WARNING`` (informational, connector still compliant)
+    - passed_normal — ``passed=True`` + non-WARNING
+    """
+    failed = [r for r in results if r.severity == Severity.ERROR]
+    advisory = [r for r in results if r.severity == Severity.WARNING]
+    passed_normal = [r for r in results if r.severity == Severity.INFO]
+    return failed, advisory, passed_normal
+
+
 def _repo_relative_path(connector_path: Path, file_path: Path | None) -> str:
     """Resolve a file_path to be relative to the git repository root.
 
@@ -104,7 +116,7 @@ def _format_result_line(
     else:
         status = _c("FAIL", "red", stream)
 
-    code = _c(result.code, "cyan", stream)
+    code = _c(result.code, SEVERITY_COLOR[result.severity], stream)
     return f"  {location}: {code} [{status}] {result.message}"
 
 
@@ -153,25 +165,27 @@ def format_text(
 ) -> None:
     """Format results as human-readable text with colors.
 
-    By default, only failures (FAIL), warnings (WARN), and the score summary
-    are displayed.  Use ``verbose=True`` to also show passing checks (PASS).
+    Grouping rules (default mode, i.e. ``verbose=False``):
+
+    - **failed**   (“FAIL” / “WARN”) — all results with ``passed=False``, any severity.
+      Suggestions are shown below each failing line.
+    - **advisory** (“WARN”)           — results with ``passed=True`` and
+      ``severity=WARNING``.  These carry notes but do not fail the connector.
+    - **passed**   (“PASS”)           — only shown when ``verbose=True``.
     """
-    failed = [r for r in results if r.severity == Severity.ERROR]
-    warnings = [r for r in results if r.severity == Severity.WARNING]
-    passed_normal = [r for r in results if r.severity == Severity.INFO]
+    failed, advisory, passed_normal = _group_results(results)
 
     def _write_result(result: CheckResult) -> None:
         stream.write(
             f"{_format_result_line(result, connector_path, stream, abspath=abspath)}\n",
         )
         if result.suggestion:
-            suggestion = _c(f"    ↳ {result.suggestion}", "dim", stream)
-            stream.write(f"{suggestion}\n")
+            stream.write(f"{_c(f'    \u21b3 {result.suggestion}', 'dim', stream)}\n")
 
     for result in failed:
         _write_result(result)
 
-    for result in warnings:
+    for result in advisory:
         _write_result(result)
 
     if verbose:
@@ -188,7 +202,6 @@ def format_json(
     stream: TextIO,
 ) -> None:
     """Format results as JSON."""
-    output_results = results
     total = len(results)
     passed_count = len([r for r in results if r.severity != Severity.ERROR])
 
@@ -212,7 +225,7 @@ def format_json(
                 "line": r.line,
                 "suggestion": r.suggestion,
             }
-            for r in output_results
+            for r in results
         ],
     }
     json.dump(output, stream, indent=2)
@@ -235,9 +248,9 @@ def format_markdown(
     stream.write(f"# Connector Linter Report — `{connector_name}`\n\n")
 
     total = len(results)
-    passed_count = len([r for r in results if r.passed])
+    passed_count = len([r for r in results if r.severity != Severity.ERROR])
     failed_count = total - passed_count
-    errors = len([r for r in results if not r.passed and r.severity == Severity.ERROR])
+    errors = len([r for r in results if r.severity == Severity.ERROR])
     warnings = len([r for r in results if r.severity == Severity.WARNING])
     pct = (passed_count / total) * 100 if total else 0
 
@@ -253,9 +266,7 @@ def format_markdown(
         summary_parts.append(f"{warnings} warning(s)")
     stream.write(f"{', '.join(summary_parts)}\n\n")
 
-    failed = [r for r in results if not r.passed]
-    warn_results = [r for r in results if r.passed and r.severity == Severity.WARNING]
-    passed_normal = [r for r in results if r.passed and r.severity != Severity.WARNING]
+    failed, advisory, passed_normal = _group_results(results)
 
     def _md_path(r: CheckResult) -> str:
         if abspath:
@@ -276,9 +287,9 @@ def _md_line(r: CheckResult, icon: str) -> str:
             stream.write(f"{_md_line(r, '❌')}\n")
         stream.write("\n")
 
-    if warn_results:
-        stream.write("## ⚠️ Warnings\n\n")
-        for r in warn_results:
+    if advisory:
+        stream.write("## ⚠️ Advisories\n\n")
+        for r in advisory:
             stream.write(f"{_md_line(r, '⚠️')}\n")
         stream.write("\n")
 

shared/connector_linter/connector_linter/models.py

@@ -1,12 +1,32 @@
 """Data models for the connector linter."""
 
+import ast
 import json
 from dataclasses import dataclass, field
 from enum import StrEnum
+from functools import cached_property
 from pathlib import Path
 from typing import Any
 
 
+class ConnectorType(StrEnum):
+    """Known OpenCTI connector types."""
+
+    EXTERNAL_IMPORT = "EXTERNAL_IMPORT"
+    INTERNAL_ENRICHMENT = "INTERNAL_ENRICHMENT"
+    INTERNAL_EXPORT_FILE = "INTERNAL_EXPORT_FILE"
+    INTERNAL_IMPORT_FILE = "INTERNAL_IMPORT_FILE"
+    STREAM = "STREAM"
+
+    @property
+    def label(self) -> str:
+        """Human-readable label derived from the value.
+
+        Examples: EXTERNAL_IMPORT → 'External Import', STREAM → 'Stream'.
+        """
+        return self.value.replace("_", " ").title()
+
+
 class Severity(StrEnum):
     """Severity levels for check results."""
 
@@ -18,6 +38,23 @@ def symbol(self) -> str:
         """Get a short symbol for the severity level."""
         return {"error": "E", "warning": "W", "info": "I"}[self.value]
 
+    def rank(self) -> int:
+        """Numeric rank for ordering (INFO=0, WARNING=1, ERROR=2)."""
+        return {"info": 0, "warning": 1, "error": 2}[self.value]
+
+
+# Shared severity → display mappings. Keyed by Severity enum for direct lookup.
+SEVERITY_EMOJI: dict["Severity", str] = {
+    Severity.ERROR: "🔴",
+    Severity.WARNING: "🟡",
+    Severity.INFO: "🔵",
+}
+SEVERITY_COLOR: dict["Severity", str] = {
+    Severity.ERROR: "red",
+    Severity.WARNING: "yellow",
+    Severity.INFO: "cyan",
+}
+
 
 @dataclass
 class CheckFinding:
@@ -48,12 +85,30 @@ class CheckResult:
     suggestion: str | None = None
 
 
+def no_python_sources_finding(suggestion: str | None = None) -> "CheckFinding":
+    """Standard finding for checks that require Python source files but find none."""
+    return CheckFinding(
+        message="No Python source files found in src/",
+        severity=Severity.ERROR,
+        suggestion=suggestion or "Connector must have Python source files under src/",
+    )
+
+
+_DIR_TO_CONNECTOR_TYPE: dict[str, ConnectorType] = {
+    "external-import": ConnectorType.EXTERNAL_IMPORT,
+    "internal-enrichment": ConnectorType.INTERNAL_ENRICHMENT,
+    "internal-export-file": ConnectorType.INTERNAL_EXPORT_FILE,
+    "internal-import-file": ConnectorType.INTERNAL_IMPORT_FILE,
+    "stream": ConnectorType.STREAM,
+}
+
+
 @dataclass
 class ConnectorContext:
     """Contextual data about a connector, loaded once and shared across checks."""
 
     path: Path
-    connector_type: str | None = None
+    connector_type: ConnectorType | None = None
     manifest: dict[str, Any] = field(default_factory=dict)
     config_schema: dict[str, Any] = field(default_factory=dict)
     has_tests: bool = False
@@ -63,34 +118,65 @@ class ConnectorContext:
     src_files: list[Path] = field(default_factory=list)
     all_files: list[Path] = field(default_factory=list)
 
+    @cached_property
+    def python_sources(self) -> dict[Path, str]:
+        """All Python source files under src/, keyed by path relative to connector root.
+
+        Computed once and cached for the lifetime of this context.
+        Uses src_files populated at load time to avoid re-scanning the filesystem.
+        """
+        sources: dict[Path, str] = {}
+        for rel_path in self.src_files:
+            abs_path = self.path / rel_path
+            try:
+                sources[rel_path] = abs_path.read_text(
+                    encoding="utf-8", errors="replace"
+                )
+            except OSError:
+                continue
+        return sources
+
+    @cached_property
+    def python_trees(self) -> dict[Path, ast.Module]:
+        """Parsed AST modules for all Python source files.
+
+        Computed once and cached for the lifetime of this context.
+        Files with syntax errors are silently skipped.
+        """
+        trees: dict[Path, ast.Module] = {}
+        for file_path, content in self.python_sources.items():
+            try:
+                trees[file_path] = ast.parse(content, filename=str(file_path))
+            except SyntaxError:
+                continue
+        return trees
+
     @classmethod
     def load(cls, connector_path: Path) -> "ConnectorContext":
         """Load connector context from its directory."""
         ctx = cls(path=connector_path.resolve())
 
         # Detect connector type from parent directory name
-        parent_name = ctx.path.parent.name
-        type_mapping = {
-            "external-import": "EXTERNAL_IMPORT",
-            "internal-enrichment": "INTERNAL_ENRICHMENT",
-            "internal-export-file": "INTERNAL_EXPORT_FILE",
-            "internal-import-file": "INTERNAL_IMPORT_FILE",
-            "stream": "STREAM",
-        }
-        ctx.connector_type = type_mapping.get(parent_name)
+        ctx.connector_type = _DIR_TO_CONNECTOR_TYPE.get(ctx.path.parent.name)
         # Fallback only for template layout: templates/<connector-kind>
-        if ctx.connector_type is None and parent_name == "templates":
-            ctx.connector_type = type_mapping.get(ctx.path.name)
+        if ctx.connector_type is None and ctx.path.parent.name == "templates":
+            ctx.connector_type = _DIR_TO_CONNECTOR_TYPE.get(ctx.path.name)
 
         # Load manifest
         manifest_path = ctx.path / "__metadata__" / "connector_manifest.json"
         if manifest_path.exists():
-            with manifest_path.open() as f:
-                ctx.manifest = json.load(f)
+            try:
+                with manifest_path.open() as f:
+                    ctx.manifest = json.load(f)
+            except (json.JSONDecodeError, OSError):
+                pass  # malformed or unreadable — checks that need it will report missing fields
 
         # Fallback: use container_type from manifest
         if ctx.connector_type is None and ctx.manifest.get("container_type"):
-            ctx.connector_type = ctx.manifest["container_type"]
+            try:
+                ctx.connector_type = ConnectorType(ctx.manifest["container_type"])
+            except ValueError:
+                pass  # unknown type string — leave as None
 
         if ctx.connector_type is None:
             raise ValueError(
@@ -105,8 +191,11 @@ def load(cls, connector_path: Path) -> "ConnectorContext":
         # Load config schema
         schema_path = ctx.path / "__metadata__" / "connector_config_schema.json"
         if schema_path.exists():
-            with schema_path.open() as f:
-                ctx.config_schema = json.load(f)
+            try:
+                with schema_path.open() as f:
+                    ctx.config_schema = json.load(f)
+            except (json.JSONDecodeError, OSError):
+                pass  # malformed or unreadable — leave as empty dict
 
         # Detect structural elements
         ctx.has_metadata_dir = (ctx.path / "__metadata__").is_dir()

shared/connector_linter/connector_linter/noqa.py

@@ -67,29 +67,21 @@ def get_noqa_directives(file_path: Path) -> dict[int, set[str] | None]:
 def is_suppressed(result: CheckResult, file_path: Path, line: int) -> bool:
     """Check if a result is suppressed by a noqa directive on its line."""
     directives = get_noqa_directives(file_path)
-    directive = directives.get(line)
-
-    if directive is None and line in directives:
-        return True  # bare noqa — suppress everything
-
-    return directive is not None and result.code.upper() in directive
+    if line not in directives:
+        return False
+    codes = directives[line]
+    return codes is None or result.code.upper() in codes  # None = bare noqa
 
 
 def filter_noqa(
     results: list[CheckResult],
     connector_path: Path,
 ) -> list[CheckResult]:
-    """Filter results that are suppressed by noqa directives.
-
-    Only results with both ``file_path`` and ``line`` set are eligible
-    for suppression.  Results without location info pass through unchanged.
+    """Filter results suppressed by noqa directives.
 
-    *connector_path* is the connector root directory.  When a result carries
-    a relative ``file_path`` (common — most checks report paths relative to
-    the connector root), it is resolved against *connector_path* so that
-    ``_read_file_lines`` opens the correct file on disk.
+    Results without both ``file_path`` and ``line`` pass through unchanged.
+    Relative paths are resolved against *connector_path*.
     """
-    _read_file_lines.cache_clear()
     resolved_root = connector_path.resolve()
     filtered: list[CheckResult] = []