refactor(common, spectral-validator): introduce jentic-openapi-common package with subprocess utility

Add new `jentic-openapi-common` package containing shared utilities for OpenAPI-related operations:

- Implement `run_checked` subprocess utility with enhanced error handling.
- Add `SubprocessExecutionError` to handle non-zero exit codes.
- Include type-checking support with `py.typed`.
- Update workspace and CI configurations to include `jentic-openapi-common`.
- Replace direct `subprocess.run` calls with `run_checked` in `spectral_validator.py` for improved error handling and result encapsulation.

This package provides reusable functionality for subprocess management, aiding other OpenAPI tools in the monorepo.

Author: frantuma

.github/workflows/ci.yml

@@ -25,6 +25,25 @@ jobs:
           uv run ruff format --check .
           uv run pyright
 
+  test_common:
+    name: Test (jentic-openapi-Common)
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python: ["3.11", "3.12"]
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+      - name: Set up Python
+        run: uv python install ${{ matrix.python }}
+      - name: Install deps (with dev)
+        run: uv sync
+      - name: Install common package in editable mode
+        run: uv pip install -e jentic-openapi-common
+      - name: Pytest (parser)
+        run: uv run pytest jentic-openapi-common/tests -q
+
   test_parser:
     name: Test (jentic-openapi-parser)
     runs-on: ubuntu-latest

.github/workflows/pre-release.yml

@@ -38,8 +38,12 @@ jobs:
       - name: Install dependencies
         run: uv sync
 
+      - name: Install common package in editable mode
+        run: uv pip install -e jentic-openapi-common
+
       - name: Run tests
         run: |
+          uv run pytest jentic-openapi-common/tests -v
           uv run --package jentic-openapi-parser pytest packages/jentic-openapi-parser/tests -v
           uv run --package jentic-openapi-transformer pytest packages/jentic-openapi-transformer/tests -v
           uv run --package jentic-openapi-validator pytest packages/jentic-openapi-validator/tests -v

.github/workflows/release.yml

@@ -53,6 +53,9 @@ jobs:
       - name: Install dependencies
         run: uv sync
 
+      - name: Install common package in editable mode
+        run: uv pip install -e jentic-openapi-common
+
       - name: Dry Run Notice
         if: env.DRY_RUN == 'true'
         run: |
@@ -122,6 +125,7 @@ jobs:
         if: steps.check_release.outputs.release_needed == 'true'
         run: |
           echo "🧪 Running tests..."
+          uv run pytest jentic-openapi-common/tests -v
           uv run --package jentic-openapi-parser pytest packages/jentic-openapi-parser/tests -v
           uv run --package jentic-openapi-transformer pytest packages/jentic-openapi-transformer/tests -v
           uv run --package jentic-openapi-validator pytest packages/jentic-openapi-validator/tests -v

README.md

@@ -13,6 +13,7 @@ uv sync --all-packages
 ### run per-package tests (from root)
 
 ```bash
+uv run pytest jentic-openapi-common/tests -q
 uv run --package jentic-openapi-parser       pytest packages/jentic-openapi-parser/tests -q
 uv run --package jentic-openapi-transformer  pytest packages/jentic-openapi-transformer/tests -q
 uv run --package jentic-openapi-validator    pytest packages/jentic-openapi-validator/tests -q

jentic-openapi-common/README.md

@@ -0,0 +1,3 @@
+# jentic-openapi-common
+
+Common code for OpenAPI not specific to any package.

jentic-openapi-common/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "jentic-openapi-common"
+version = "0.1.0"
+description = "Jentic OpenAPI Common"
+readme = "README.md"
+authors = [{ name = "Jentic", email = "hello@jentic.com" }]
+license = { text = "Apache-2.0" }
+requires-python = ">=3.11"
+dependencies = []
+
+[project.urls]
+Homepage = "https://github.com/jentic/jentic-openapi-tools"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/jentic_openapi_common"]

jentic-openapi-common/src/jentic_openapi_common/__init__.py

@@ -0,0 +1,3 @@
+from .subprocess import run_checked, SubprocessExecutionResult, SubprocessExecutionError
+
+__all__ = ["run_checked", "SubprocessExecutionResult", "SubprocessExecutionError"]

jentic-openapi-common/src/jentic_openapi_common/py.typed

@@ -0,0 +1,2 @@
+# marker file for type checkers
+

jentic-openapi-common/src/jentic_openapi_common/subprocess.py

@@ -0,0 +1,105 @@
+import subprocess
+from typing import Sequence, Optional
+
+
+class SubprocessExecutionResult:
+    """Returned by a subprocess."""
+
+    def __init__(
+        self,
+        returncode: int,
+        stdout: str = "",
+        stderr: str = "",
+    ):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class SubprocessExecutionError(RuntimeError):
+    """Raised when a subprocess exits with non-zero return code."""
+
+    def __init__(
+        self,
+        cmd: Sequence[str],
+        returncode: int,
+        stdout: Optional[str] = None,
+        stderr: Optional[str] = None,
+    ):
+        self.cmd = list(cmd)
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+        message = (
+            f"Command {cmd!r} failed with exit code {returncode}\n"
+            f"--- stdout ---\n{stdout or ''}\n"
+            f"--- stderr ---\n{stderr or ''}"
+        )
+        super().__init__(message)
+
+
+def run_checked(
+    cmd: Sequence[str],
+    *,
+    fail_on_error: bool = False,
+    timeout: Optional[float] = None,
+    encoding: str = "utf-8",
+    errors: str = "strict",
+) -> SubprocessExecutionResult:
+    """
+    Run a subprocess command and return (stdout, stderr) as text.
+    Raises SubprocessExecutionError if the command fails.
+
+    Parameters
+    ----------
+    cmd : sequence of str
+        The command and its arguments.
+    timeout : float | None
+        Seconds before timing out.
+    encoding : str
+        Passed to subprocess.run so stdout/stderr are decoded as text.
+    errors : str
+        Error handler for text decoding.
+
+    Returns
+    -------
+    (stdout, stderr, returncode) : SubprocessExecutionResult
+    """
+    try:
+        completed = subprocess.run(
+            cmd,
+            check=False,
+            capture_output=True,
+            text=True,
+            encoding=encoding,  # ensure CompletedProcess has str stdout/stderr
+            errors=errors,
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired as e:
+        # e.stdout / e.stderr are bytes|None even with text=True — normalize to str
+        stdout = (
+            e.stdout.decode(encoding, errors)
+            if isinstance(e.stdout, (bytes, bytearray))
+            else (e.stdout or "")
+        )
+        stderr = (
+            e.stderr.decode(encoding, errors)
+            if isinstance(e.stderr, (bytes, bytearray))
+            else (e.stderr or "")
+        )
+        raise SubprocessExecutionError(cmd, -1, stdout, stderr) from e
+    except OSError as e:  # e.g., executable not found, permission denied
+        raise SubprocessExecutionError(cmd, -1, None, str(e)) from e
+
+    if completed.returncode != 0 and fail_on_error:
+        raise SubprocessExecutionError(
+            cmd,
+            completed.returncode,
+            completed.stdout or "",
+            completed.stderr or "",
+        )
+
+    # At this point CompletedProcess stdout/stderr are str due to text=True + encoding
+    return SubprocessExecutionResult(
+        completed.returncode, completed.stdout or "", completed.stderr or ""
+    )

jentic-openapi-common/tests/test_subprocess.py

@@ -0,0 +1,178 @@
+import sys
+import unittest
+import platform
+
+from jentic_openapi_common.subprocess import (
+    run_checked,
+    SubprocessExecutionResult,
+    SubprocessExecutionError,
+)
+
+
+class TestSubprocessModule(unittest.TestCase):
+    """Test cases for the custom subprocess module."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.is_windows = platform.system() == "Windows"
+        self.is_unix = platform.system() in ("Linux", "Darwin")
+
+    @unittest.skipUnless(
+        platform.system() in ("Linux", "Darwin", "Windows"), "Requires a supported operating system"
+    )
+    def test_successful_command(self):
+        """Test running a successful command."""
+        if self.is_windows:
+            cmd = ["cmd", "/c", "echo", "hello"]
+        else:
+            cmd = ["echo", "hello"]
+
+        result = run_checked(cmd)
+
+        self.assertIsInstance(result, SubprocessExecutionResult)
+        self.assertEqual(result.returncode, 0)
+        self.assertIn("hello", result.stdout)
+        self.assertEqual(result.stderr, "")
+
+    @unittest.skipUnless(platform.system() in ("Linux", "Darwin"), "Requires Unix-like system")
+    def test_unix_ls_command(self):
+        """Test ls command on Unix systems."""
+        result = run_checked(["ls", "/"])
+
+        self.assertEqual(result.returncode, 0)
+        self.assertIsInstance(result.stdout, str)
+        self.assertTrue(len(result.stdout) > 0)
+
+    @unittest.skipUnless(platform.system() == "Windows", "Requires Windows system")
+    def test_windows_dir_command(self):
+        """Test dir command on Windows systems."""
+        result = run_checked(["cmd", "/c", "dir", "C:\\"])
+
+        self.assertEqual(result.returncode, 0)
+        self.assertIsInstance(result.stdout, str)
+        self.assertTrue(len(result.stdout) > 0)
+
+    @unittest.skipUnless(
+        platform.system() in ("Linux", "Darwin", "Windows"), "Requires a supported operating system"
+    )
+    def test_python_version_command(self):
+        """Test running python --version command."""
+        # Use sys.executable to ensure we're using the right Python
+        result = run_checked([sys.executable, "--version"])
+
+        self.assertEqual(result.returncode, 0)
+        self.assertIn("Python", result.stdout)
+
+    def test_nonexistent_command(self):
+        """Test running a command that doesn't exist."""
+        with self.assertRaises(SubprocessExecutionError) as cm:
+            run_checked(["nonexistent_command_12345"])
+
+        self.assertEqual(cm.exception.returncode, -1)
+        self.assertEqual(cm.exception.cmd, ["nonexistent_command_12345"])
+
+    @unittest.skipUnless(
+        platform.system() in ("Linux", "Darwin", "Windows"), "Requires a supported operating system"
+    )
+    def test_command_with_non_zero_exit_code(self):
+        """Test command that exits with non-zero code."""
+        if self.is_windows:
+            cmd = ["cmd", "/c", "exit", "1"]
+        else:
+            cmd = ["sh", "-c", "exit 1"]
+
+        # Without fail_on_error, should not raise exception
+        result = run_checked(cmd)
+        self.assertEqual(result.returncode, 1)
+
+        # With fail_on_error=True, should raise exception
+        with self.assertRaises(SubprocessExecutionError) as cm:
+            run_checked(cmd, fail_on_error=True)
+
+        self.assertEqual(cm.exception.returncode, 1)
+        self.assertEqual(cm.exception.cmd, cmd)
+
+    @unittest.skipUnless(
+        platform.system() in ("Linux", "Darwin"), "Requires Unix-like system for stderr test"
+    )
+    def test_command_with_stderr(self):
+        """Test command that produces stderr output."""
+        # Use a command that writes to stderr
+        result = run_checked(["sh", "-c", "echo 'error message' >&2"])
+
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertIn("error message", result.stderr)
+
+    def test_encoding_parameter(self):
+        """Test custom encoding parameter."""
+        if self.is_windows:
+            cmd = ["cmd", "/c", "echo", "hello"]
+        else:
+            cmd = ["echo", "hello"]
+
+        result = run_checked(cmd, encoding="utf-8")
+
+        self.assertEqual(result.returncode, 0)
+        self.assertIn("hello", result.stdout)
+        self.assertIsInstance(result.stdout, str)
+
+    def test_subprocess_execution_result_initialization(self):
+        """Test SubprocessExecutionResult initialization."""
+        result = SubprocessExecutionResult(0, "stdout", "stderr")
+
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "stdout")
+        self.assertEqual(result.stderr, "stderr")
+
+        # Test with defaults
+        result_defaults = SubprocessExecutionResult(1)
+        self.assertEqual(result_defaults.returncode, 1)
+        self.assertEqual(result_defaults.stdout, "")
+        self.assertEqual(result_defaults.stderr, "")
+
+    def test_subprocess_execution_error_initialization(self):
+        """Test SubprocessExecutionError initialization."""
+        cmd = ["test", "command"]
+        error = SubprocessExecutionError(cmd, 1, "out", "err")
+
+        self.assertEqual(error.cmd, ["test", "command"])
+        self.assertEqual(error.returncode, 1)
+        self.assertEqual(error.stdout, "out")
+        self.assertEqual(error.stderr, "err")
+
+        # Test message formatting
+        error_msg = str(error)
+        self.assertIn("['test', 'command']", error_msg)
+        self.assertIn("exit code 1", error_msg)
+        self.assertIn("out", error_msg)
+        self.assertIn("err", error_msg)
+
+    @unittest.skipUnless(platform.system() in ("Linux", "Darwin"), "Requires Unix-like system")
+    def test_empty_command_output(self):
+        """Test command that produces no output."""
+        result = run_checked(["true"])  # Unix command that does nothing and succeeds
+
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(result.stderr, "")
+
+    @unittest.skipUnless(
+        platform.system() in ("Linux", "Darwin", "Windows"), "Requires a supported operating system"
+    )
+    def test_command_with_arguments(self):
+        """Test command with multiple arguments."""
+        if self.is_windows:
+            cmd = ["cmd", "/c", "echo", "hello", "world"]
+        else:
+            cmd = ["echo", "hello", "world"]
+
+        result = run_checked(cmd)
+
+        self.assertEqual(result.returncode, 0)
+        self.assertIn("hello", result.stdout)
+        self.assertIn("world", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()