Add unit tests for pipeline QA job script

#  Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>

#

#  Licensed under the Apache License, Version 2.0 (the "License");

#  you may not use this file except in compliance with the License.

#  You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

#  Unless required by applicable law or agreed to in writing, software

#  distributed under the License is distributed on an "AS IS" BASIS,

#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#  See the License for the specific language governing permissions and

#  limitations under the License.

"""

Temporary file and directory helper module

"""

from __future__ import annotations

import tempfile

from pathlib import Path

from typing import TypeVar

TEMP_DIR = Path(tempfile.gettempdir())

class TempDir(Path):

	"""

	A temporary directory context manager Path subclass

	This class implements almost identical behaviour to tempfile.TemporaryDirectory, but as

	a pathlib.Path subclass.

	Note: The "name" attribute is from Path; it is the base name of the directory, not the

	full path as in tempfile.TemporaryDirectory.

	"""

	_flavour = type(Path())._flavour  # type: ignore

	P = TypeVar("P", bound="TempDir")

	def __new__(cls: type[P]) -> P:  # noqa: D102 - https://github.com/PyCQA/pydocstyle/issues/515

		return Path.__new__(cls, tempfile.mkdtemp())

	def __enter__(self: P) -> P:

		return self

	def __exit__(self, *_: object) -> None:

		self.cleanup()

	def cleanup(self) -> None:

		"""

		Remove the directory and all its contents

		"""

		if TEMP_DIR not in self.parents:

			raise RuntimeError(f"not removing {self} as it's not in {TEMP_DIR}")

		if not self.exists():

			return

		for path in reversed([*self.glob("**/*")]):

			path.rmdir() if path.is_dir() else path.unlink()

		self.rmdir()

#  Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>

#

#  Licensed under the Apache License, Version 2.0 (the "License");

#  you may not use this file except in compliance with the License.

#  You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

#  Unless required by applicable law or agreed to in writing, software

#  distributed under the License is distributed on an "AS IS" BASIS,

#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

#  See the License for the specific language governing permissions and

#  limitations under the License.

"""

Functions and types for fixtures used by the tests

"""

from __future__ import annotations

import random

import subprocess

from pathlib import Path

from typing import NewType

from tests.temp import TempDir

Ref = NewType("Ref", str)

Sha = NewType("Sha", str)

MAIN = Ref("main")

FORK = Ref("fork")

ORPHAN = Ref("orphan")

class Branch:

	"""

	Class representing a repository branch

	"""

	def __init__(self, repo: Repo, name: str, root: Sha) -> None:

		self.repo = repo

		self.name = Ref(name)

		self.head = root

		self.root = root

		self.lcam = dict[Ref, Sha]()

	@property

	def head(self) -> Sha:

		"""

		The commit in the commit graph pointed to by this reference

		"""  # noqa: D401 - Until pydocstyle 6.2 is released

		return self._head

	@head.setter

	def head(self, commit: Sha) -> None:

		self.repo.command("branch", "-f", self.name, commit)

		self._head = commit

class Repo:

	"""

	Manage a git repository

	"""

	def __init__(self, *, init: bool = False) -> None:

		self.directory = TempDir()

		self.url = f"file://{self.directory}"

		self._env = {

			"GIT_DIR": self.directory.as_posix(),

			"GIT_AUTHOR_NAME": "A. Coder",

			"GIT_AUTHOR_EMAIL": "coder@example.com",

			"GIT_COMMITTER_NAME": "May N. Tainer",

			"GIT_COMMITTER_EMAIL": "mnt@example.com",

		}

		self._initialised = False

		if init:

			self.init()

	def __str__(self) -> str:

		return self.url

	def init(self, branch: str = MAIN) -> None:

		"""

		Initialise (git-init) a bare repository, with "branch" checked out

		"""

		if self._initialised:

			return

		self.command("init", "--bare", "-qb", branch, self.directory)

	def cleanup(self) -> None:

		"""

		Remove the repository and its directory

		"""

		self.directory.cleanup()

	def command(self, command: str, *args: str|Path, input: bytes = b"") -> None:

		"""

		Run the given git command with optional stdin input, expecting a success code

		"""

		subprocess.run(

			["git", command, *args],

			env=self._env,

			check=True,

			input=input,

		)

	def create(self, command: str, *args: str|Path, input: bytes = b"") -> Sha:

		"""

		Like "git" but expect a SHA1 key on stdout, which is returned

		"""

		proc = subprocess.run(

			["git", command, *args],

			env=self._env,

			check=True,

			stdout=subprocess.PIPE,

			input=input,

		)

		return Sha(proc.stdout.decode().strip())

	def get_sha(self, ref: str) -> Sha:

		"""

		Get the SHA1 key from a reference or symbolic reference

		"""

		proc = subprocess.run(

			["git", "rev-parse", "--revs-only", "--verify", ref],

			env=self._env,

			stdout=subprocess.PIPE,

		)

		if proc.returncode == 128:

			raise LookupError(f"revision not found: {ref}")

		proc.check_returncode()

		return Sha(proc.stdout.decode().strip())

def setup_fixture_repo(repo: Repo) -> dict[Ref, Branch]:

	r"""

	Create a sample git repository as a test fixture

	The repository consists of three branches, two with common ancestors and one with no

	commonality with the other two:

	┌──────────────────────────────────┐

	│          Last common ancestor    │

	│           ↓                      │

	│   ○─○─○─○─○─○─○─○─○─● "main"     │

	│           └─○─○─○─○─● "fork"     │

	│   ○─○─○─○─○─○─○─○─○─● "orphan"   │

	└──────────────────────────────────┘

	"""

	random.seed(1)

	root = create_commit(repo)

	main = Branch(repo, MAIN, root)

	fork = Branch(repo, FORK, root)

	lca = create_commit_path(repo, root, 5)

	main.lcam[fork.name] = fork.lcam[main.name] = lca

	main.head = create_commit_path(repo, lca, 5)

	fork.head = create_commit_path(repo, lca, 5)

	orphan = Branch(repo, ORPHAN, create_commit(repo))

	orphan.head = create_commit_path(repo, orphan.root, 5)

	return {b.name: b for b in (main, fork, orphan)}

def create_commit_path(repo: Repo, source: Sha, number: int) -> Sha:

	"""

	Create "n(umber)" commits from source, returning the sink

	"""

	commit = source

	for _ in range(number):

		commit = create_commit(repo, commit)

	return commit

def create_commit(repo: Repo, *parents: Sha) -> Sha:

	"""

	Create a new commit object, with any given parents

	"""

	tree = create_tree(

		repo,

		(create_blob(repo), "sample.dat"),

	)

	return repo.create("commit-tree", tree, *(f"-p{p}" for p in parents), input=b"A Commit")

def create_blob(repo: Repo) -> Sha:

	"""

	Create a new unique blob with arbitrary contents and return it's SHA ID

	"""

	contents = random.randbytes(20)

	return repo.create("hash-object", "-w", "--stdin", input=contents)

def create_tree(repo: Repo, *objects: tuple[Sha, str]) -> Sha:

	"""

	Create a new tree object containing the given blobs or sub-trees

	"""

	objinfo = (f"100644,{sha},{name}" for sha, name in objects)

	repo.command("update-index", "--add", "--cacheinfo", *objinfo)

	return repo.create("write-tree")