github.com_openSUSE_osc/osc/git_scm/store.py

import fcntl
import fnmatch
import json
import os
from pathlib import Path
import typing
import urllib.parse
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union

from .. import oscerr
from ..util.models import BaseModel
from ..util.models import Field


if typing.TYPE_CHECKING:
    from .core import Repo


class Header(BaseModel):
    type: str = Field()
    version: str = Field()


class Meta(BaseModel):
    """
    Metadata about a project or a package managed in git that is stored in .git/obs/<branch>/meta.json
    """

    header: Header = Field(default={"type": "obs-metadata-store", "version": "1"})
    apiurl: Optional[str] = Field()
    project: Optional[str] = Field()
    package: Optional[str] = Field()

    def update(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)


class Lock:
    def __init__(self, path: str):
        self.path = os.path.abspath(path)
        self.handle = None

    def __enter__(self):
        os.makedirs(os.path.dirname(self.path), exist_ok=True)
        self.handle = open(self.path, "w")
        fcntl.flock(self.handle, fcntl.LOCK_EX)

    def __exit__(self, type, value, traceback):
        if self.handle:
            fcntl.flock(self.handle, fcntl.LOCK_UN)
            self.handle.close()
            try:
                os.remove(self.path)
            except OSError:
                pass


class BuildRoot(BaseModel):
    """
    Model that encapsulates last_buildroot values, providing better API than the original tuple.

    For compatibility, assigning values to repo, arch, vm_type variables is also supported via __iter__()
    and __eq__ is capable of comparing with a tuple with (repo, arch, vm_type).
    """
    repo: str = Field()
    arch: str = Field()
    vm_type: Optional[str] = Field()

    def __iter__(self):
        for field in self.__fields__:
            yield getattr(self, field)

    def __eq__(self, other):
        if isinstance(other, tuple) and len(other) == 3:
            return (self.repo, self.arch, self.vm_type) == other
        return super().__eq__(other)


class LocalGitStore:
    """
    A class for managing OBS metadata in .git.
    It is not supposed to be used directly, it's a base class for GitStore that adds logic for resolving the values from multiple places.
    """

    @classmethod
    def is_project_dir(cls, path):
        try:
            store = cls(path)
        except oscerr.NoWorkingCopy:
            return False
        return store.is_project

    @classmethod
    def is_package_dir(cls, path):
        try:
            store = cls(path)
        except oscerr.NoWorkingCopy:
            return False
        return store.is_package

    def __init__(self, path: str, *, check: bool = True):
        from ..gitea_api import Git
        from ..obs_scm import Store
        from .manifest import Manifest
        from .manifest import Subdirs

        self._git = Git(path)

        if not self._git.topdir:
            msg = f"Directory '{path}' is not a Git SCM working copy"
            raise oscerr.NoWorkingCopy(msg)

        if not self._git.current_branch:
            # branch is required for determining and storing metadata
            msg = f"Directory '{path}' is not a Git SCM working copy because it has no branch or is in a detached HEAD state"
            raise oscerr.NoWorkingCopy(msg)

        # 'package' is the default type that applies to all git repos that are not projects (we have no means of detecting packages)
        self._type = "package"
        self._topdir = self._git.topdir

        self.manifest = None

        # we detect projects by looking for certain file names next to .git
        files = ["_manifest", "_config", "_pbuild", "_subdirs"]
        for fn in files:
            path = os.path.join(self._git.topdir, fn)
            if os.path.exists(path):
                self._type = "project"
                break

        if self.type == "project":
            if self._git.topdir != self.abspath:
                manifest_path = os.path.join(self._git.topdir, "_manifest")
                subdirs_path = os.path.join(self._git.topdir, "_subdirs")

                if os.path.exists(manifest_path):
                    self.manifest = Manifest.from_file(manifest_path)
                elif os.path.exists(subdirs_path):
                    self.manifest = Subdirs.from_file(subdirs_path)
                else:
                    # empty manifest considers all top-level directories as packages
                    self.manifest = Manifest()

                package_topdir = self.manifest.resolve_package_path(project_path=self._git.topdir, package_path=self.abspath)
                if package_topdir:
                    self._type = "package"
                    self._topdir = package_topdir

        self.project_store = None
        if self.type == "package":
            # load either .osc or .git project store from the directory above topdir
            for cls in (Store, self.__class__):
                try:
                    store = cls(os.path.join(self.topdir, ".."))
                    store.assert_is_project()
                    self.project_store = store
                    break
                except oscerr.NoWorkingCopy:
                    pass
        elif self.type == "project":
            # load .osc project store that is next to .git and may provide medatata we don't have
            try:
                store = Store(self.topdir)
                store.assert_is_project()
                self.project_store = store
            except oscerr.NoWorkingCopy:
                pass

    @property
    def abspath(self) -> str:
        return self._git.abspath

    @property
    def topdir(self) -> str:
        return self._topdir

    def reset(self, *, branch: Optional[str] = None):
        self._delete_meta(branch=branch)

    def _get_path(self, path: List[str], *, branch: Optional[str] = None):
        assert isinstance(path, list)
        # sanitization for os.path.join()
        branch = branch if branch is not None else self._git.current_branch
        branch = branch.strip("/")
        branch = branch.replace("/", "__")
        path = [i.strip("/") for i in path]

        result = self._git._run_git(["rev-parse", "--path-format=absolute", "--git-path", "obs"])
        result = os.path.join(result, branch, *path)
        return result

    def _lock(self):
        return Lock(self._get_path([".lock"]))

    def _delete_meta(self, *, branch: Optional[str] = None):
        path = self._get_path(["meta.json"], branch=branch)
        try:
            os.unlink(path)
        except FileNotFoundError:
            pass

    def _read_meta(self, *, branch: Optional[str] = None) -> Meta:
        path = self._get_path(["meta.json"], branch=branch)
        try:
            with self._lock():
                return Meta.from_file(path)
        except FileNotFoundError:
            return Meta()

    def _write_meta(self, *, branch: Optional[str] = None, **kwargs):
        path = self._get_path(["meta.json"], branch=branch)
        with self._lock():
            os.makedirs(os.path.dirname(path), exist_ok=True)
            try:
                meta = Meta.from_file(path)
            except FileNotFoundError:
                meta = Meta()
            meta.update(**kwargs)
            meta.to_file(path)

    @property
    def type(self):
        return self._type

    @property
    def is_project(self) -> bool:
        return self.type == "project"

    @property
    def is_package(self) -> bool:
        return self.type == "package"

    def assert_is_project(self):
        if not self.is_project:
            msg = f"Directory '{self.abspath}' is not a Git SCM working copy of a project"
            raise oscerr.NoWorkingCopy(msg)

    def assert_is_package(self):
        if not self.is_package:
            msg = f"Directory '{self.abspath}' is not a Git SCM working copy of a package"
            raise oscerr.NoWorkingCopy(msg)

    # APIURL

    @property
    def apiurl(self) -> Optional[str]:
        return self.get_apiurl()

    @apiurl.setter
    def apiurl(self, value: Optional[str]):
        self.set_apiurl(value)

    def get_apiurl(self, *, branch: Optional[str] = None) -> Optional[str]:
        return self._read_meta(branch=branch).apiurl

    def set_apiurl(self, value: Optional[str], *, branch: Optional[str] = None):
        self._write_meta(apiurl=value, branch=branch)

    # PROJECT

    @property
    def project(self) -> Optional[str]:
        return self.get_project()

    @project.setter
    def project(self, value: Optional[str]):
        self.set_project(value)

    def get_project(self, *, branch: Optional[str] = None) -> Optional[str]:
        return self._read_meta(branch=branch).project

    def set_project(self, value: Optional[str], *, branch: Optional[str] = None):
        self._write_meta(project=value, branch=branch)

    # PACKAGE

    @property
    def package(self) -> Optional[str]:
        return self.get_package()

    @package.setter
    def package(self, value: Optional[str]):
        self.set_package(value)

    def get_package(self, *, branch: Optional[str] = None) -> Optional[str]:
        return self._read_meta(branch=branch).package

    def set_package(self, value: Optional[str], *, branch: Optional[str] = None):
        self.assert_is_package()
        self._write_meta(package=value, branch=branch)

    # CACHE
    # buildinfo and buildconfig files are considered a cache, they can be safely deleted at any time

    def cache_list_files(self, *, pattern: Optional[str] = None, branch: Optional[str] = None):
        path = self._get_path(["cache"], branch=branch)
        files = os.listdir(path)
        if pattern:
            files = [i for i in files if fnmatch.fnmatch(i, pattern)]
        return files

    def cache_read_file(self, filename: str, *, branch: Optional[str] = None) -> Optional[bytes]:
        assert "/" not in filename
        branch = branch or self._git.current_branch
        path = self._get_path(["cache", filename], branch=branch)
        with self._lock():
            try:
                with open(path, "rb") as f:
                    return f.read()
            except FileNotFoundError:
                return None

    def cache_write_file(self, filename: str, data: bytes, *, branch: Optional[str] = None):
        assert "/" not in filename
        branch = branch or self._git.current_branch
        path = self._get_path(["cache", filename], branch=branch)
        with self._lock():
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, "wb") as f:
                f.write(data)

    def cache_delete_files(self, filenames: List[str], *, branch: Optional[str] = None):
        branch = branch or self._git.current_branch
        with self._lock():
            for filename in filenames:
                assert "/" not in filename
                path = self._get_path(["cache", filename], branch=branch)
                os.unlink(path)

    # LAST BUILDROOT

    _LAST_BUILDROOT_FILE = "last-buildroot.json"
    _LAST_BUILDROOT_VALUE_TYPE = Union[Optional[Tuple[str, str, Optional[str]]], BuildRoot]

    @property
    def last_buildroot(self) -> _LAST_BUILDROOT_VALUE_TYPE:
        return self.get_last_buildroot()

    @last_buildroot.setter
    def last_buildroot(self, value: _LAST_BUILDROOT_VALUE_TYPE):
        self.set_last_buildroot(value)

    def get_last_buildroot(self, *, branch: Optional[str] = None) -> _LAST_BUILDROOT_VALUE_TYPE:
        result = self.cache_read_file(self._LAST_BUILDROOT_FILE, branch=branch)
        if result is None:
            return None
        obj = BuildRoot.from_string(result.decode("utf-8"))
        return obj

    def set_last_buildroot(self, value: _LAST_BUILDROOT_VALUE_TYPE, *, branch: Optional[str] = None):
        if value is None:
            self.cache_delete_files([self._LAST_BUILDROOT_FILE])
            return

        if isinstance(value, tuple):
            if len(value) != 3:
                raise ValueError("A tuple with exactly 3 items is expected: (repo, arch, vm_type)")
            obj = BuildRoot(repo=value[0], arch=value[1], vm_type=value[2])
        else:
            obj = value

        data = json.dumps(obj.dict()).encode("utf-8")
        self.cache_write_file(self._LAST_BUILDROOT_FILE, data, branch=branch)

    # BUILD REPOSITORIES

    _BUILD_REPOSITORIES_FILE = "build-repositories.json"
    _BUILD_REPOSITORIES_VALUE_TYPE = Optional[List["Repo"]]

    @property
    def build_repositories(self) -> _BUILD_REPOSITORIES_VALUE_TYPE:
        return self.get_build_repositories()

    @build_repositories.setter
    def build_repositories(self, value: Optional[List["Repo"]]):
        return self.set_build_repositories(value)

    def get_build_repositories(self, *, branch: Optional[str] = None) -> Optional[List["Repo"]]:
        from ..core import Repo

        result = self.cache_read_file(self._BUILD_REPOSITORIES_FILE, branch=branch)
        if result is None:
            return None
        result = json.loads(result)
        return [Repo(**i) for i in result]

    def set_build_repositories(self, value: Optional[List["Repo"]], *, branch: Optional[str] = None):
        from ..core import Repo

        if value is None:
            self.cache_delete_files([self._BUILD_REPOSITORIES_FILE])
            return

        repos = []
        if value is not None:
            for i in value:
                if not isinstance(i, Repo):
                    raise ValueError(f"The value is not an instance of 'Repo': {i}")
                repos.append(i.dict())
        self.cache_write_file(self._BUILD_REPOSITORIES_FILE, json.dumps(repos).encode("utf-8"), branch=branch)


class GitStore(LocalGitStore):
    """
    A class for managing OBS metadata in .git that also reads the values from additional locations
    such as the parent project's metadata or the _manifest file.
    """
    def __init__(self, path: str, check: bool = True, *, cached: bool = True):
        super().__init__(path, check=check)
        self.cached = cached
        self._cache = {}

    def _resolve_meta(self, field_name: str, *, allow_none: bool = False):
        result = None

        # values cached in the object
        if self.cached:
            result = self._cache.get(field_name, None)

        # local git store
        if result is None:
            result = getattr(super(), field_name)

        # _manifest file in the project
        if result is None and self.is_project and self.manifest:
            result = getattr(self.manifest, f"obs_{field_name}", None) or None

        # project.build file in the project
        if result is None and self.is_project and field_name == "project":
            path = os.path.join(self.topdir, "project.build")
            if os.path.exists(path):
                with open(path, "r", encoding="utf-8") as f:
                    result = f.read().strip() or None

        # package = repo name from the current remote url
        if result is None and self.is_package and field_name == "package":
            remote_url = self._git.get_remote_url()
            if remote_url:
                result = Path(urllib.parse.urlsplit(remote_url).path).stem

        # project: get value from .osc which is next to .git
        # package: get value from the parent project's store
        if result is None and self.project_store:
            result = getattr(self.project_store, field_name, None)

        if self.cached:
            self._cache[field_name] = result

        return result

    @property
    def apiurl(self) -> Optional[str]:
        return self._resolve_meta("apiurl")

    @property
    def project(self) -> Optional[str]:
        return self._resolve_meta("project")

    @property
    def package(self) -> Optional[str]:
        return self._resolve_meta("package")

    @property
    def scmurl(self) -> Optional[str]:
        return self._git.get_remote_url()