tools/git-if-needed/python/gifn_apply/repo_url.py

# SPDX-FileCopyrightText: StorPool <support@storpool.com>
# SPDX-License-Identifier: BSD-2-Clause
"""Parse base/URL repository location pairs."""

from __future__ import annotations

import dataclasses
import os
import re
import urllib.parse as uparse

from . import defs
from . import util


_RE_ENV_BASE = re.compile(r"^ [A-Z][A-Z0-9_]* $", re.X)


@dataclasses.dataclass(frozen=True)
class RepoURLResult:
    """Base class for the OK/error parsed URL dichotomy."""


@dataclasses.dataclass(frozen=True)
class RepoURLOK(RepoURLResult):
    """Successfully parsed a base URL for repositories."""

    url: defs.RepoURL


@dataclasses.dataclass(frozen=True)
class RepoURLError(RepoURLResult):
    """Could not parse a base URL for repositories."""

    name: str
    value: str
    err: ValueError


@dataclasses.dataclass(frozen=True)
class RepoURLPair:
    """A base/URL pair for an URL obtained from the command line."""

    base: str
    url: defs.RepoURL


def _slash_extend(url: uparse.ParseResult) -> uparse.ParseResult:
    """Add a / at the end of the path if there is none."""
    if url.path.endswith("/"):
        return url

    return url._replace(path=url.path + "/")


def _validate_file(name: str, value: str, url: uparse.ParseResult) -> RepoURLResult:
    """Make sure a file:// URL has no host and an absolute path."""
    if url.netloc:
        return RepoURLError(name, value, ValueError("No hostname expected for a 'file' URL"))
    if not url.path.startswith("/"):
        return RepoURLError(name, value, ValueError("Expected an absolute path for a 'file' URL"))

    url = _slash_extend(url)
    return RepoURLOK(defs.RepoURL(url))


def _validate_http(name: str, value: str, url: uparse.ParseResult) -> RepoURLResult:
    """Make sure a http(s):// URL has a host, slash-terminate the path."""
    if not url.netloc:
        return RepoURLError(
            name,
            value,
            ValueError("Expected a hostname for 'http' or 'https' URLs"),
        )

    url = _slash_extend(url)
    return RepoURLOK(defs.RepoURL(url))


_SCHEME_VALIDATORS = {
    "file": _validate_file,
    "http": _validate_http,
    "https": _validate_http,
}


def parse_url(name: str, base: str, value: str) -> RepoURLResult:
    """Parse and validate a single base/URL pair."""
    if not _RE_ENV_BASE.match(base):
        return RepoURLError(name, value, ValueError(f"Invalid URL base {base!r}"))

    try:
        url = uparse.urlparse(value)
    except ValueError as err:
        return RepoURLError(name, value, err)

    validator = _SCHEME_VALIDATORS.get(url.scheme)
    if validator is None:
        return RepoURLError(
            name,
            value,
            ValueError("Expected 'http', 'https', or 'file' as the URL scheme"),
        )
    return validator(name, value, url)


def get_env_repo_urls(environ: dict[str, str] | None = None) -> dict[str, RepoURLResult]:
    """Parse the REPO_URL_<base> environment variables."""
    if environ is None:
        environ = dict(os.environ)

    res: dict[str, RepoURLResult] = {
        "OPENSTACK": RepoURLOK(defs.RepoURL(uparse.urlparse("https://github.com/openstack"))),
    }
    for name, value in environ.items():
        base = util.str_removeprefix(name, "REPO_URL_")
        if base == name:
            continue
        res[base] = parse_url(name, base, value)

    return res


def parse_base_url_pair(arg: str) -> RepoURLPair:
    """Parse a `--repo-url base=url` command-line argument."""
    raise NotImplementedError(repr(arg))