Deduplicate dependency version pins from backend code/tests
Some checks failed
CICD / Build and Push CICD Images (push) Successful in 37m42s
CICD / Build CICD Image Failure Postmortem (push) Has been skipped
CICD / Dependency Audits (Informational) (push) Successful in 15m24s
CICD / Source Checks (push) Failing after 18m19s
CICD / Build Release Images (push) Has been skipped
CICD / Build Tester Images (push) Has been skipped
CICD / Production Image Failures Postmortem (push) Has been skipped
CICD / Production Images Complete (push) Failing after 11s
CICD / Runtime Black-Box Integration Tests (push) Has been skipped
CICD / End-to-End Tests (push) Has been skipped
CICD / Integration Tests Failure Postmortem (push) Has been skipped
CICD / E2E Tests Failure Postmortem (push) Has been skipped
CICD / Promote Staging Images To Release (push) Has been skipped
CICD / Source Lanes Failure Postmortem (push) Successful in 16s
CICD / CICD Tests Complete (push) Failing after 28s

This commit is contained in:
copilotcoder
2026-07-20 15:16:28 -04:00
parent 4b2c46e1d8
commit fb82e3947d
3 changed files with 69 additions and 19 deletions

View File

@@ -6,9 +6,11 @@ from __future__ import annotations
import os
import sys
import tomllib
from contextlib import asynccontextmanager
from importlib import metadata
from typing import Any
from pathlib import Path
from typing import Any, cast
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
@@ -20,12 +22,65 @@ from backend.database import (
probe_database,
)
REQUIRED_PACKAGE_PINS: dict[str, str] = {
"fastapi": "0.139.0",
"psycopg": "3.2.12",
"sqlalchemy": "2.0.44",
"uvicorn": "0.51.0",
}
PROJECT_DISTRIBUTION_NAME = "plex-playlist-backend"
def _parse_exact_pin(requirement: str) -> tuple[str, str] | None:
"""Extract normalized package name and exact version from a requirement string."""
requirement_without_marker = requirement.split(";", 1)[0].strip()
if "==" not in requirement_without_marker:
return None
package_part, version = requirement_without_marker.split("==", 1)
package_name = package_part.strip().split("[", 1)[0].lower()
pinned_version = version.strip()
if not package_name or not pinned_version:
return None
return package_name, pinned_version
def _required_package_pins_from_dependencies(dependencies: list[str]) -> dict[str, str]:
"""Build exact pins from project dependency declarations."""
required_pins: dict[str, str] = {}
for requirement in dependencies:
parsed_pin = _parse_exact_pin(requirement)
if parsed_pin is None:
continue
package_name, version = parsed_pin
required_pins[package_name] = version
return required_pins
def _required_package_pins() -> dict[str, str]:
"""Load required package pins from project metadata derived from pyproject."""
try:
dependencies = metadata.requires(PROJECT_DISTRIBUTION_NAME) or []
pins = _required_package_pins_from_dependencies(dependencies)
if pins:
return pins
except metadata.PackageNotFoundError:
pass
pyproject_file = Path(__file__).resolve().parents[2] / "pyproject.toml"
pyproject = cast(
"dict[str, object]",
tomllib.loads(pyproject_file.read_text(encoding="utf-8")),
)
project_section = pyproject.get("project")
if not isinstance(project_section, dict):
return {}
project_config = cast("dict[str, object]", project_section)
dependencies = project_config.get("dependencies")
if not isinstance(dependencies, list):
return {}
dependency_entries = cast("list[object]", dependencies)
dependency_strings = [
dependency for dependency in dependency_entries if isinstance(dependency, str)
]
return _required_package_pins_from_dependencies(dependency_strings)
def _parse_required_python(required_python: str) -> tuple[int, int] | None:
@@ -56,10 +111,11 @@ def compatibility_status() -> dict[str, object]:
required_python = os.getenv("BACKEND_REQUIRED_PYTHON", "3.14")
current_python = f"{sys.version_info.major}.{sys.version_info.minor}"
required_python_parsed = _parse_required_python(required_python)
required_package_pins = _required_package_pins()
package_checks: dict[str, bool] = {}
package_errors: dict[str, str] = {}
for package_name, required_version in REQUIRED_PACKAGE_PINS.items():
for package_name, required_version in required_package_pins.items():
try:
package_checks[package_name] = (
_installed_version(package_name) == required_version
@@ -78,7 +134,7 @@ def compatibility_status() -> dict[str, object]:
"required_python": required_python,
"current_python": current_python,
"python_policy_valid": required_python_parsed is not None,
"required_packages": REQUIRED_PACKAGE_PINS,
"required_packages": required_package_pins,
"package_checks": package_checks,
"package_errors": package_errors,
}

View File

@@ -84,6 +84,7 @@ class TestAPIIntegration:
) -> None:
"""Compatibility endpoint should expose runtime and pinning policy status."""
monkeypatch.setenv("BACKEND_REQUIRED_PYTHON", "3.14")
expected_required_packages = compatibility_status()["required_packages"]
with TestClient(app) as local_client:
response = local_client.get("/compatibility")
@@ -94,9 +95,7 @@ class TestAPIIntegration:
assert payload["required_python"] == "3.14"
assert "current_python" in payload
assert payload["python_policy_valid"] is True
assert "required_packages" in payload
assert payload["required_packages"]["fastapi"] == "0.139.0"
assert payload["required_packages"]["uvicorn"] == "0.51.0"
assert payload["required_packages"] == expected_required_packages
assert payload["package_errors"] == {}
monkeypatch.delenv("BACKEND_REQUIRED_PYTHON", raising=False)

View File

@@ -98,15 +98,10 @@ def test_compatibility_status_success(monkeypatch: pytest.MonkeyPatch):
"""Compatibility status should be healthy when all checks match policy."""
monkeypatch.setenv("BACKEND_REQUIRED_PYTHON", "3.14")
required_pins = cast("dict[str, str]", compatibility_status()["required_packages"])
def installed_version(package_name: str) -> str:
versions = {
"fastapi": "0.139.0",
"psycopg": "3.2.12",
"sqlalchemy": "2.0.44",
"uvicorn": "0.51.0",
}
return versions[package_name]
return required_pins[package_name]
monkeypatch.setattr("backend.main._installed_version", installed_version)
status = cast("dict[str, Any]", compatibility_status())