Working draft for CodeTFv3 data model (#1012)

* Move current CodeTF implementation to v2

* Add initial draft of CodeTF v3 model

Author: drdavella

src/codemodder/codetf/__init__.py

@@ -0,0 +1 @@
+from .v2.codetf import *  # noqa: F403

src/codemodder/codetf/common.py

@@ -0,0 +1,28 @@
+from abc import ABCMeta
+from enum import Enum
+from pathlib import Path
+
+from pydantic import BaseModel
+
+from codemodder.logging import logger
+
+
+class CaseInsensitiveEnum(str, Enum):
+    @classmethod
+    def _missing_(cls, value: object):
+        if not isinstance(value, str):
+            return super()._missing_(value)
+
+        return cls.__members__.get(value.upper())
+
+
+class CodeTFWriter(BaseModel, metaclass=ABCMeta):
+    def write_report(self, outfile: Path | str) -> int:
+        try:
+            Path(outfile).write_text(self.model_dump_json(exclude_none=True))
+        except Exception:
+            logger.exception("failed to write report file.")
+            # Any issues with writing the output file should exit status 2.
+            return 2
+        logger.debug("wrote report to %s", outfile)
+        return 0

src/codemodder/codetf.py src/codemodder/codetf/v2/codetf.py

@@ -9,27 +9,18 @@
 import os
 import sys
 from enum import Enum
-from pathlib import Path
 from typing import TYPE_CHECKING, Optional
 
 from pydantic import BaseModel, ConfigDict, model_validator
 
 from codemodder import __version__
-from codemodder.logging import logger
+
+from ..common import CaseInsensitiveEnum, CodeTFWriter
 
 if TYPE_CHECKING:
     from codemodder.context import CodemodExecutionContext
 
 
-class CaseInsensitiveEnum(str, Enum):
-    @classmethod
-    def _missing_(cls, value: object):
-        if not isinstance(value, str):
-            return super()._missing_(value)
-
-        return cls.__members__.get(value.upper())
-
-
 class Action(CaseInsensitiveEnum):
     ADD = "add"
     REMOVE = "remove"
@@ -221,7 +212,7 @@ class Run(BaseModel):
     sarifs: list[Sarif] = []
 
 
-class CodeTF(BaseModel):
+class CodeTF(CodeTFWriter, BaseModel):
     run: Run
     results: list[Result]
 
@@ -247,13 +238,3 @@ def build(
             sarifs=[],
         )
         return cls(run=run, results=results)
-
-    def write_report(self, outfile: Path | str):
-        try:
-            Path(outfile).write_text(self.model_dump_json(exclude_none=True))
-        except Exception:
-            logger.exception("failed to write report file.")
-            # Any issues with writing the output file should exit status 2.
-            return 2
-        logger.debug("wrote report to %s", outfile)
-        return 0

src/codemodder/codetf/v3/codetf.py

@@ -0,0 +1,181 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, model_validator
+
+from ..common import CaseInsensitiveEnum, CodeTFWriter
+
+
+class Run(BaseModel):
+    """Metadata about the analysis run that produced the results"""
+
+    vendor: str
+    tool: str
+    version: str
+    # Optional free-form metadata about the project being analyzed
+    # e.g. project name, directory, commit SHA, etc.
+    projectMetadata: Optional[str] = None
+    # Analysis duration in milliseconds
+    elapsed: Optional[int] = None
+    # Optional free-form metadata about the inputs used for the analysis
+    # e.g. command line, environment variables, etc.
+    inputMetadata: Optional[dict] = None
+    # Optional free-form metadata about the analysis itself
+    # e.g. timeouts, memory usage, etc.
+    analysisMetadata: Optional[dict] = None
+
+
+class FixStatusType(str, Enum):
+    """Status of a fix"""
+
+    fixed = "fixed"
+    skipped = "skipped"
+    failed = "failed"
+    wontfix = "wontfix"
+
+
+class FixStatus(BaseModel):
+    """Metadata describing fix outcome"""
+
+    status: FixStatus
+    reason: Optional[str]
+    details: Optional[str]
+
+
+class Rule(BaseModel):
+    id: str
+    name: str
+    url: Optional[str] = None
+
+
+class Finding(BaseModel):
+    id: str
+    rule: Optional[Rule] = None
+
+
+class Action(CaseInsensitiveEnum):
+    ADD = "add"
+    REMOVE = "remove"
+
+
+class PackageResult(CaseInsensitiveEnum):
+    COMPLETED = "completed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+
+class DiffSide(CaseInsensitiveEnum):
+    LEFT = "left"
+    RIGHT = "right"
+
+
+class PackageAction(BaseModel):
+    action: Action
+    result: PackageResult
+    package: str
+
+
+class Change(BaseModel):
+    lineNumber: int
+    description: Optional[str]
+    diffSide: DiffSide = DiffSide.RIGHT
+    properties: Optional[dict] = None
+    packageActions: Optional[list[PackageAction]] = None
+
+    @model_validator(mode="after")
+    def validate_lineNumber(self):
+        if self.lineNumber < 1:
+            raise ValueError("lineNumber must be greater than 0")
+        return self
+
+    @model_validator(mode="after")
+    def validate_description(self):
+        if self.description is not None and not self.description:
+            raise ValueError("description must not be empty")
+        return self
+
+
+class ChangeSet(BaseModel):
+    path: str
+    diff: str
+    changes: list[Change]
+
+
+class Reference(BaseModel):
+    url: str
+    description: Optional[str] = None
+
+    @model_validator(mode="after")
+    def validate_description(self):
+        self.description = self.description or self.url
+        return self
+
+
+class Strategy(Enum):
+    ai = "ai"
+    hybrid = "hybrid"
+    deterministic = "deterministic"
+
+
+class AIMetadata(BaseModel):
+    provider: Optional[str] = None
+    models: Optional[list[str]] = None
+    total_tokens: Optional[int] = None
+    completion_tokens: Optional[int] = None
+    prompt_tokens: Optional[int] = None
+
+
+class GenerationMetadata(BaseModel):
+    strategy: Strategy
+    ai: Optional[AIMetadata] = None
+    provisional: bool
+
+
+class FixMetadata(BaseModel):
+    # Fix provider ID, corresponds to legacy codemod ID
+    id: str
+    # A brief summary of the fix
+    summary: str
+    # A detailed description of the fix
+    description: str
+    references: list[Reference]
+    generation: GenerationMetadata
+
+
+class Rating(BaseModel):
+    score: int
+    description: Optional[str] = None
+
+
+class FixQuality(BaseModel):
+    safetyRating: Rating
+    effectivenessRating: Rating
+    cleanlinessRating: Rating
+
+
+class FixResult(BaseModel):
+    """Result corresponding to a single finding"""
+
+    finding: Finding
+    fixStatus: FixStatus
+    changeSets: list[ChangeSet]
+    fixMetadata: Optional[FixMetadata] = None
+    fixQuality: Optional[FixQuality] = None
+    # A description of the reasoning process that led to the fix
+    reasoningSteps: Optional[list[str]] = None
+
+    @model_validator(mode="after")
+    def validate_fixMetadata(self):
+        if self.fixStatus.status == FixStatusType.fixed:
+            if not self.changeSets:
+                raise ValueError("changeSets must be provided for fixed results")
+            if not self.fixMetadata:
+                raise ValueError("fixMetadata must be provided for fixed results")
+        return self
+
+
+class CodeTF(CodeTFWriter, BaseModel):
+    run: Run
+    results: list[FixResult]