Results Management

@dataclass(frozen=True)
class MypyResults:
    """
    Parsed and cached mypy execution results.
    
    Stores all information from a mypy run including command arguments,
    output, exit status, and per-file error lines for efficient access
    by multiple test items.
    """
    
    opts: List[str]  # Mypy command options used
    args: List[str]  # File arguments passed to mypy
    stdout: str      # Complete mypy stdout output
    stderr: str      # Complete mypy stderr output
    status: int      # Mypy exit code
    path_lines: Dict[Optional[Path], List[str]]  # Error lines grouped by file
    
    def dump(self, results_f: IO[bytes]) -> None:
        """
        Cache results to file for persistence across processes.
        
        Serializes the complete results structure to JSON format
        for storage in the temporary results file. Path objects are
        converted to strings for JSON compatibility.
        
        Parameters:
        - results_f: Binary file handle for writing cached results
        """
    
    @classmethod
    def load(cls, results_f: IO[bytes]) -> MypyResults:
        """
        Load cached results from file.
        
        Deserializes results from JSON format and reconstructs
        the complete MypyResults structure.
        
        Parameters:
        - results_f: Binary file handle containing cached results
        
        Returns:
        Fully reconstructed MypyResults instance
        """
    
    @classmethod
    def from_mypy(
        cls,
        paths: List[Path],
        *,
        opts: Optional[List[str]] = None
    ) -> MypyResults:
        """
        Execute mypy and create results from output.
        
        Runs mypy with specified options on the given paths,
        parses the output, and groups error lines by file.
        
        Parameters:
        - paths: List of file paths to check with mypy
        - opts: Optional mypy command line options
        
        Returns:
        MypyResults containing complete execution results
        """
    
    @classmethod
    def from_session(cls, session: pytest.Session) -> MypyResults:
        """
        Load or generate cached mypy results for a pytest session.
        
        Uses file locking to ensure thread safety. If cached results
        exist, loads them; otherwise runs mypy and caches results.
        
        Parameters:
        - session: Pytest session containing configuration and items
        
        Returns:
        MypyResults for all MypyFileItems in the session
        """

@dataclass(frozen=True)
class MypyConfigStash:
    """
    Plugin configuration data stored in pytest.Config stash.
    
    Maintains the path to cached mypy results and provides
    serialization for xdist worker communication.
    """
    
    mypy_results_path: Path  # Path to temporary results cache file
    
    @classmethod
    def from_serialized(cls, serialized: str) -> MypyConfigStash:
        """
        Reconstruct stash from serialized string.
        
        Used by xdist workers to recreate configuration
        from controller-provided serialized data.
        
        Parameters:
        - serialized: String representation of results path
        
        Returns:
        MypyConfigStash instance with reconstructed path
        """
    
    def serialized(self) -> str:
        """
        Serialize stash for transmission to workers.
        
        Converts the results path to string format for
        communication between xdist controller and workers.
        
        Returns:
        String representation suitable for deserialization
        """

# Global stash key for accessing cached configuration
stash_key = {
    "config": pytest.StashKey[MypyConfigStash](),
}

# Example of how results are safely accessed
def get_results_safely(session):
    mypy_results_path = session.config.stash[stash_key["config"]].mypy_results_path
    with FileLock(str(mypy_results_path) + ".lock"):
        try:
            with open(mypy_results_path, mode="rb") as results_f:
                return MypyResults.load(results_f)
        except FileNotFoundError:
            # First access - run mypy and cache results
            results = MypyResults.from_mypy(collected_paths)
            with open(mypy_results_path, mode="wb") as results_f:
                results.dump(results_f)
            return results

class MypyXdistControllerPlugin:
    """
    Plugin active only on xdist controller processes.
    
    Responsible for passing configuration to worker processes
    so they can access the shared results cache.
    """
    
    def pytest_configure_node(self, node: WorkerController) -> None:
        """
        Pass configuration stash to xdist workers.
        
        Serializes the results path and includes it in worker
        input so workers can access the shared cache.
        
        Parameters:
        - node: xdist worker controller being configured
        """

class MypyControllerPlugin:
    """
    Plugin for main/controller processes (not xdist workers).
    
    Handles terminal reporting and cleanup of temporary files.
    """
    
    def pytest_terminal_summary(
        self,
        terminalreporter: TerminalReporter,
        config: pytest.Config
    ) -> None:
        """
        Report mypy results in terminal summary.
        
        Displays mypy output including errors, notes, and status
        information in the pytest terminal report.
        
        Parameters:
        - terminalreporter: Pytest terminal reporter instance
        - config: Pytest configuration object
        """
    
    def pytest_unconfigure(self, config: pytest.Config) -> None:
        """
        Clean up temporary mypy results file.
        
        Removes the cached results file when pytest session ends.
        
        Parameters:
        - config: Pytest configuration object
        """

class CustomMypyFileItem(MypyFileItem):
    def runtest(self):
        # Access cached mypy results
        results = MypyResults.from_session(self.session)
        
        # Get errors for this specific file
        file_errors = results.path_lines.get(self.path.resolve(), [])
        
        # Custom error processing
        if file_errors:
            print(f"Found {len(file_errors)} errors in {self.path}")
            for error in file_errors:
                print(f"  {error}")

def analyze_mypy_results(session):
    """Analyze mypy results for custom reporting."""
    results = MypyResults.from_session(session)
    
    total_errors = sum(len(lines) for lines in results.path_lines.values())
    files_with_errors = len([lines for lines in results.path_lines.values() if lines])
    
    print(f"Mypy found {total_errors} issues across {files_with_errors} files")
    print(f"Exit status: {results.status}")