#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0-only # SPDX-FileCopyrightText: Copyright (C) 2024 Stefan Lengfeld import os import sys import shutil import time import stat import subprocess from enum import Enum from argparse import ArgumentParser from subprocess import Popen, PIPE, DEVNULL from os.path import join, abspath from os import mkdir from dataclasses import dataclass from typing import Any import contextlib # See https://peps.python.org/pep-0440/ for details about the version format. # e.g. dashes "-" are not allowed and 'a' stands for 'alpha' release. __version__ = "0.1a4" # It's the SPDX identifier. See https://spdx.org/licenses/GPL-2.0-only.html __LICENSE__ = "GPL-2.0-only" # TODO refactor. copyied from helpers @contextlib.contextmanager def cwd(path, create=False): if create: os.makedirs(path) old_path = os.getcwd() try: os.chdir(path) yield finally: os.chdir(old_path) # "Section names are case-insensitive. Only alphanumeric characters, - and . # are allowed in section names" def is_section_name(b): raise NotImplementedError() # Split with terminator sematics def split_with_ts(s): len_s = len(s) pos = 0 while pos < len_s: new_pos = s.find("\n", pos) if new_pos == -1: # newline character not found anymore new_pos = len_s - 1 yield s[pos:new_pos + 1] pos = new_pos + 1 # TODO find naming convention! # # Parse git-style config file. # See https://git-scm.com/docs/git-config # The syntax is described here: # https://git-scm.com/docs/git-config/2.12.5#_syntax # Goal: The parsed result can be converted back to the lines object with a 1to1 mapping # NOTES: # - not supporting "[section.subsection]" syntax # - not supporting continouation lines yet # - not supporting comments at end of line yet. # - every line in lines, ends with a "\n" character # Only the last line may not have a trailing "\n" character. # TODO add errors on invalid syntax # TODO lines should be byte objects, not strings def config_parse(lines): # result items # (,
, , , , ) def get_first(b): if len(b) == 0: return None return b[0] for line in lines: first_char = get_first(line.lstrip()) if first_char is None: # It's an empty line yield 1, line else: if first_char in ('#', ';'): yield 1, line elif first_char == '[': # section start, like: "[section]\n" # Parse section name # TODO Check for valid section characters inner_part = line.split('[', 1)[1].split(']')[0] if '"' in inner_part: # There is a subsection: # [section "subsection"] section_name = inner_part.split('"')[0].strip() subsection_name = inner_part.split('"', 2)[1] else: section_name = inner_part subsection_name = None yield 2, line, section_name, subsection_name else: # This is mostly a variable line # name = value parts = line.split("=", 1) name = parts[0].strip() value = parts[1].strip() yield 3, line, name, value def config_unparse(parts_list): s = "" for parts in parts_list: s += parts[1] return s # NOTE: the section_lines is a list of the internal parsing type. This is a bit # strange to expose the internal parsing data structure to the caller. def config_add_section(parts_list, section_name, subsection_name, section_lines): # HACK: Make the variable a "box" value to access it from inside the inner # function. was_emit = [False] def emit(): if was_emit[0]: return # TODO the trailing "\n" is strange here. All code is agnostice, e.g. # works without newlines But here is the assumption that the config # file is properly newlined. yield 2, "[%s \"%s\"]\n" % (section_name, subsection_name), section_name, subsection_name for parts in section_lines: yield parts was_emit[0] = True for parts in parts_list: if parts[0] == 2: if parts[2] > section_name: yield from emit() elif parts[2] == section_name: if parts[3] is None: yield from emit() elif parts[3] > subsection_name: yield from emit() yield parts yield from emit() def config_remove_section(parts_list, section_name, subsection_name): in_section = False for parts in parts_list: if parts[0] == 2: in_section = False if parts[2] == section_name and parts[3] == subsection_name: in_section = True if not in_section: yield parts @dataclass class Subproject: # TODO Document with fields can be None path: str url: str | None = None revision: str | None = None # Read the config file into Subproject objects # TODO is it "_" or "_"? # "subprojects_parse" vs "parse_subprojects" # ... -> then should should be "parse_config_parts_to_subprojects # NOTE # - this will not filter out duplicate entries # - it will emit elements in the order as they appear in the config file! def subprojects_parse(config_parts): subproject = None for line in config_parts: # TODO The type encoding into numbers 1, 2, 3 must go away!!! if line[0] == 1: pass # ignore empty lines elif line[0] == 2: if subproject is not None: yield subproject subproject = None if line[2] == "subpatch": subproject = Subproject(line[3]) elif line[0] == 3: if subproject is not None: name = line[2] if name == "url": subproject.url = line[3] elif name == "revision": subproject.revision = line[3] # TODO add warning for unknown value else: raise Exception("here") if subproject is not None: yield subproject # TODO Write blogpost about common error categories, e.g. in HTTP and errno # E.g. there is also # * invalid argument/Bad request # * (generic) runtime error (maybe the same as IO error) # * permission denied # * NotImplemented/Does not exists class ErrorCode(Enum): UNKNOWN = 1 # TODO distinguish between not implemented and not implemented yet! # Not implemented should mostly be a invalid argument then NOT_IMPLEMENTED_YET = 2 WORKS_ONLY_IN_A_GIT_REPO = 3 SUPERPROJECT_NOT_CONFIGURED = 4 # The user has given an invalid argument on the command line INVALID_ARGUMENT = 5 # TODO remove this type. Every ErrorCode should support a message. CUSTOM = 6 class AppException(Exception): def __init__(self, code, msg=None): self._code = code if msg is not None: super().__init__(msg) else: super().__init__() class URLTypes(Enum): LOCAL_RELATIVE = 1 LOCAL_ABSOLUTE = 2 REMOTE = 3 def get_url_type(url): if len(url) == 0: raise ValueError("The URL is empty!") # TODO mabye using url parsing library? # TODO Implemente "file://" prefix if url.startswith("http"): return URLTypes.REMOTE elif url.startswith("git"): return URLTypes.REMOTE elif url.startswith("ssh"): return URLTypes.REMOTE if "://" in url: raise NotImplementedError("The URL '%s' is not implemented yet" % (url,)) # Is mostly just a local path if url[0] == "/": return URLTypes.LOCAL_ABSOLUTE return URLTypes.LOCAL_RELATIVE def nocommand(args, parser): parser.print_help(file=sys.stderr) return 2 # TODO why 2? def cmd_help(args, parser): parser.print_help() return 0 class SCMType(Enum): GIT = b"git" # NOTE: In both cases it's the topdir path! @dataclass class FindSuperprojectData: super_path: bytes | None = None scm_type: SCMType | None = None scm_path: bytes | None = None # Based on the current work directory search for a subpatch project # and scm. # Returns: FindSuperprojectData # There can be four cases # - no ".subpatch" config and no scm # - ".subpatch" config and no scm # - no ".subpach" config and scm # - ".subpach" config and scm # For the last cases there are two sub-cases: # - super_path matches scm_path # - both paths do not match # # TODO support svn, mercurial and others in the future # TODO thinking about symlinks! def find_superproject(): abs_cwd = abspath(os.getcwdb()) # TODO refactory this self-made and hacky approach of a directory walk # The first element is empty, because of the absolute path. But using a # empty element in "join" does not work. It's the reverse operation! assert abs_cwd[0] == ord("/") abs_cwd_parts = os.path.normpath(abs_cwd).split(b"/") assert abs_cwd_parts[0] == b"" abs_cwd_parts.pop(0) statinfo = os.stat(abs_cwd) assert stat.S_ISDIR(statinfo.st_mode) cwd_st_dev = statinfo.st_dev data = FindSuperprojectData() while True: abs_cur_path = join(b"/", *abs_cwd_parts) statinfo = os.stat(abs_cur_path) if statinfo.st_dev != cwd_st_dev: # The directory walk leaves the current filesystem. This is a stop # condition for now. # TODO Maybe implement 'GIT_DISCOVERY_ACROSS_FILESYSTEM' # See https://git-scm.com/docs/git.html#Documentation/git.txt-codeGITDISCOVERYACROSSFILESYSTEMcode break if data.super_path is None: if os.path.exists(join(abs_cur_path, b".subpatch")): # Configuration file found data.super_path = abs_cur_path if data.scm_type is None: # NOTE Simple implementation. Check whether there is a ".git" # folder next to it. # Does not work for worktrees! ... really? TODO check it! if os.path.exists(join(abs_cur_path, b".git")): data.scm_type = SCMType.GIT data.scm_path = abs_cur_path if data.super_path is not None and data.scm_type is not None: # If both is already found, stop the directory walk break if len(abs_cwd_parts) == 0: break abs_cwd_parts.pop() return data # :: void -> None or byte object (or raises an exception) # TODO currently unused. Maybe remove this function def git_get_toplevel(): p = Popen(["git", "rev-parse", "--show-toplevel"], stdout=PIPE, stderr=DEVNULL) stdout, _ = p.communicate() if p.returncode == 0: return stdout.rstrip(b"\n") elif p.returncode == 128: # fatal: not a git repository (or any of the parent directories): .git return None else: # TODO Create a generic git error exception raise Exception("git failure") # NOTE This is cwd aware def git_clone(url, directory): p = Popen(["git", "clone", "-q", url, directory]) p.communicate() if p.returncode != 0: raise Exception("git failure") def git_reset_hard(sha1): p = Popen(["git", "reset", "-q", "--hard", sha1]) p.communicate() if p.returncode != 0: raise Exception("git failure") def is_sha1(sha1: bytes): assert isinstance(sha1, bytes) if len(sha1) != 40: return False return all(0x30 <= c <= 0x39 or 0x61 <= c <= 0x66 for c in sha1) # TODO Split this command. It seems like a ugly combination def git_init_and_fetch(url, ref): p = Popen(["git", "init", "-q"]) p.communicate() if p.returncode != 0: raise Exception("git failure") cmd = ["git", "fetch", "-q", url, ref] # NOTE: Http transport which is currently used in the tests! And for dump # the "--depth" argument does not work. So we need this hack for the tests. # The git error message is # fatal: dumb http transport does not support shallow capabilities # TODO implement this hack nicly! # TODO No need for an env variable or argument. This code should # automatically detect whether shallow cloes are working or not! if os.environ.get("HACK_DISABLE_DEPTH_OPTIMIZATION", "0").strip() != "1": cmd += ["--depth", "1"] p = Popen(cmd, stderr=DEVNULL) # NOTE If stderr==DEVNULL(no-tty) no progress is showing on the commandline # Not getting the error is bad! p.communicate() if p.returncode != 0: # TODO think about error handling!! # Maybe every subcommand should be able and allows to write to stderr # directly! raise Exception("git failure: %d" % (p.returncode,)) # get SHA1 of fetched object with open(".git/FETCH_HEAD", "br") as f: sha1 = f.read().split(b"\t", 1)[0] assert is_sha1(sha1) return sha1 def git_add(args): assert len(args) >= 1 p = Popen(["git", "add", "-f"] + args) p.communicate() if p.returncode != 0: raise Exception("git failure") # TODO clarify naming: path, filename, url # TODO clairfy name for remote git name and path/url # Naming: # Git naming # - object: something in the object store that has a SHA1 # - object type: type of the object. Can be blob, tree, commit or tag. # - object id: the hash of the object (taken from the git source code) # - ref: short for "reference", like branches and tags. # Everything in the 'refs' folder/namespace # - rev: short for "revision" # "A revision parameter typically, but not necessarily, names a # commit object." See 'man gitrevisions' # E.g. a , "HEAD", "x..y" or "main:file" # # Variable names: # - url: The URL for a git/svn/.. repository or tarball. # Can be a http, git or file URL. Or just a local path. # Variant: # sub_url: The subproject's URL, e.g. to clone/download # # Subpatch naming # - "config" or "subpatch config". Is the name of the config in the # ".subpatch" file at the toplevel of the superproject. # TODO add to glossary # - "subproject": Idea: Is only the local directory and config # Not the remote repository # As of git v2.9.0 these are all valid git objects OBJECT_TYPES = (b"blob", b"tree", b"commit", b"tag") class ObjectType(Enum): BLOB = b"blob" TREE = b"tree" COMMIT = b"commit" TAG = b"tag" # Returns the type of the git object # TODO not used yet. Implement sanity check # TODO naming 'rev' is incorrect. 'rev' is only for commit objects. def git_get_object_type(rev): p = Popen(["git", "cat-file", "-t", rev], stdout=PIPE, stderr=DEVNULL) stdout, _ = p.communicate() if p.returncode != 0: # TODO Create a generic git error exception raise Exception("failed here") stdout = stdout.rstrip(b"\n") assert stdout in OBJECT_TYPES if stdout == b"blob": return ObjectType.BLOB elif stdout == b"tree": return ObjectType.TREE elif stdout == b"commit": return ObjectType.COMMIT return ObjectType.TAG # Convert URLs. Examples: # …bla/foo/.git/ -> "foo" # …bla/foo.git/ -> "foo" # …bla/foo -> "foo" def get_name_from_repository_url(url): if len(url) == 0: raise ValueError("The URL is empty!") url = url.rstrip("/") if url.endswith(".git"): url = url[:-4] url = url.rstrip("/") url = url.split("/")[-1] return url # Check whether the 'rev' can be resolved to a valid object in the repository # NOTE: # - If this command is not exectued in a git repo, it raises an exception. def git_verify(rev): p = Popen(["git", "rev-parse", "--quiet", "--verify", rev + "^{object}"], stdout=DEVNULL, stderr=DEVNULL) p.communicate() if p.returncode == 1: return False if p.returncode != 0: # Example stderr output: # fatal: not a git repository (or any of the parent directories): .git # TODO replace with git specific exception raise Exception("todo") return True def is_valid_revision(revision): # TODO add more invalid and strange chars to this list # TODO eval svn, git, ... for allow characters. What about umlauts? strange_chars = ("\t", "\n", "\b") for char in strange_chars: if char in revision: return False return True # parse_sha1_names :: bytes -> dict # TODO add example of input # TODO add more tests, e.g. with syntax errors or duplicated refs. def parse_sha1_names(lines, sep=b' '): # remove the last new line character # Should only be one character lines = lines.rstrip(b'\n') # Ignore empty lines lines = [line.split(sep) for line in lines.split(b'\n') if len(line) != 0] for line in lines: if len(line) != 2: raise Exception("Parsing error in line: sep = %s parts = %s" % (sep, line)) lines = [(name, sha1) for sha1, name in lines] # checks for _, sha1 in lines: if not is_sha1(sha1): raise ValueError("String is not a SHA1 sum: %s" % (sha1,)) return dict(lines) # git_ls_remote :: string(url) -> dict # Output contains branches, tags, tag-commitisch "^{}" and the HEAD. def git_ls_remote(url): # NOTE Subpress stderr output of 'ls-remote'. In case of a fetch failure # stuff is written to stderr. p = Popen(["git", "ls-remote", url], stdout=PIPE) stdout, _ = p.communicate() if p.returncode != 0: raise Exception("git ls-remote failed") # Parse output # The output of ls-remote uses the tab character as the separator. The # show-ref command uses spaces. return parse_sha1_names(stdout, sep=b'\t') # Query the remote git repo and try to resolve the 'ref'. # E.g. # - "main" -> "refs/heads/main" # - "v1" -> "refs/tags/v1" # Notes: # - function accepts ref as string, but returns a byte Object! # Returns # - None if ref could not be resolved # TODO also return the matched object def git_ls_remote_guess_ref(url, ref): # TODO "git ls-remote" also allows to query a single ref or a pattern of # refs! refs_sha1 = git_ls_remote(url) ref = ref.encode("utf8") if ref in refs_sha1: return ref # Direct match ref_as_tag = b"refs/tags/" + ref if ref_as_tag in refs_sha1: return ref_as_tag ref_as_branch = b"refs/heads/" + ref if ref_as_branch in refs_sha1: return ref_as_branch return None @dataclass(frozen=True) class DownloadConfig: url: Any revision: Any = None @dataclass(frozen=True) class CloneConfig: full_clone: bool object_id: str | None = None ref: str | None = None # 'revision' can be None def git_resolve_to_clone_config(url, revision): # Some heuristics to sort out branch, commit/tag id or tag if revision is not None: if is_sha1(revision.encode("utf8")): # NOTE Commit ids are bad! You cannot download just a single # commit id. You can only clone everything and hope that the commit id # is in there! # TODO Handle tag ids special. They can be looked up by "git # ls-remote". return CloneConfig(full_clone=True, object_id=revision) else: # It looks like a ref to a branch or tag. Try to resolve it ref_resolved = git_ls_remote_guess_ref(url, revision) if ref_resolved is None: raise AppException(ErrorCode.INVALID_ARGUMENT, "The reference '%s' cannot be resolved to a branch or tag!" % (revision,)) return CloneConfig(full_clone=False, ref=ref_resolved) else: # Use HEAD of remote repository # TODO optimize this by looking at the output of "git ls-remote". return CloneConfig(full_clone=True) # TODO finalize of naming convention for helper code for subprojects and # superprojects. class SubHelperGit: def get_revision_as_str(self, revision): # The revision is of type Maybe. It's either None or a str. # Convert to a printable string for stdout if revision is None: return "HEAD" return revision # Download remote repository, checkout the request revision, remove # repository metadata and leave the plan files in the dir 'local_tmp_name'. def download(self, download_config: DownloadConfig, folder): # NOTE "git submodule init" creates a bare repository in ".git/modules" # TODO maybe also use that folder as a scrat pad. # TODO clone only a single branch and maybe use --depth 1 # - that is already partially implemented # TODO handle potential submodules in the subproject correctly url = download_config.url revision = download_config.revision # Can be None clone_config = git_resolve_to_clone_config(url, revision) # TODO optimization. If it's a local git repo, use --reference to share # object store. # There are three cases: # - full clone + with object id # - full clone + without object id # - fetch + with ref if clone_config.full_clone: git_clone(url, folder) if clone_config.object_id is not None: with cwd(folder): if not git_verify(clone_config.object_id): # TODO use context wrapper for cleanup! shutil.rmtree(b"../" + folder) raise AppException(ErrorCode.INVALID_ARGUMENT, "Object id '%s' does not point to a valid object!" % (clone_config.object_id,)) object_type = git_get_object_type(clone_config.object_id) if object_type not in (ObjectType.COMMIT, ObjectType.TAG): # TODO use context wrapper for cleanup! shutil.rmtree(b"../" + folder) raise AppException(ErrorCode.INVALID_ARGUMENT, "Object id '%s' does not point to a commit or tag object!" % (clone_config.object_id,)) git_reset_hard(clone_config.object_id) else: # TODO This is really ugly! assert not os.path.isabs(folder) assert b"//" not in folder # sanitze path folder_count = folder.rstrip(b"/").count(b"/") + 1 with cwd(folder, create=True): # TODO Not work with relative paths, because a subdir is used! if get_url_type(url) == URLTypes.LOCAL_RELATIVE: # This is ugly! # TODO fix str bs byte missmatch. encode should not be needed here! url_tmp = b"../" * folder_count + url.encode("utf8") else: url_tmp = url object_id = git_init_and_fetch(url_tmp, clone_config.ref) git_reset_hard(object_id) # Remove ".git" folder in this repo local_repo_git_dir = join(folder, b".git") assert os.path.isdir(local_repo_git_dir) shutil.rmtree(local_repo_git_dir) class SuperHelperPlain: pass class SuperHelperGit: # Add the file in 'path' to the index # TODO not all version control systems have the notion of a index! def add(self, paths): git_add(paths) def print_add_success(self): print("- To inspect the changes, use `git status` and `git diff --staged`.") print("- If you want to keep the changes, commit them with `git commit`.") print("- If you want to revert the changes, execute `git reset --merge`.") def is_cwd_toplevel_directory(paths): return paths.super_to_cwd_relpath == b"" # TODO also take cwd into account def get_subproject(super_to_sub_relpath, subprojects): assert isinstance(super_to_sub_relpath, bytes) # TODO convert subprojects to bytes types x = os.fsdecode(super_to_sub_relpath) for subproject in subprojects: if subproject.path == x: return subproject raise AppException(ErrorCode.INVALID_ARGUMENT, "Path '%s' does not point to a subproject" % (x,)) def cmd_update(args, parser): if args.path is None: # TODO should also work when cwd is inside the subproject raise AppException(ErrorCode.NOT_IMPLEMENTED_YET, "Must give path to subproject") data = find_superproject() superx = check_and_get_superproject_from_data(data) ensure_superproject_is_configured(superx) ensure_superproject_is_git(superx) sub_helper = SubHelperGit() config_abspath = join(superx.path, b".subpatch") with open(config_abspath) as f: config_parts = config_parse(split_with_ts(f.read())) subprojects = subprojects_parse(config_parts) # Force the generator before closing the file subprojects = list(subprojects) cwd_to_sub_relpath = os.fsencode(args.path) paths = do_paths(superx.path, cwd_to_sub_relpath) subproject = get_subproject(paths.super_to_sub_relpath, subprojects) # TODO if some of the patches are deapply, it's ok to have changes in the # index! How to handle that here? In that case if git_diff_in_dir(superx.path, paths.super_to_sub_relpath): # TODO add more explanations and commands to fix! raise AppException(ErrorCode.INVALID_ARGUMENT, "There are unstaged changes in the subproject.") if git_diff_in_dir(superx.path, paths.super_to_sub_relpath, staged=True): # TODO add more explanations and commands to fix! raise AppException(ErrorCode.INVALID_ARGUMENT, "There are staged changes in the subproject.") # TODO deapply all patches # TODO check that the subproject is in a clean state # TODO Download new version if args.url is not None: # TODO verify URL sub_url = args.url else: sub_url = subproject.url if args.revision is not None: # TODO verify revision revision = args.revision else: revision = subproject.revision # Can also be None! print("Updating subproject '%s' from URL '%s' to revision '%s'..." % (paths.cwd_to_sub_relpath.decode("utf8"), sub_url, sub_helper.get_revision_as_str(revision)), end="") sys.stdout.flush() # TODO if git, and revision is a commit or tag id, check that the new sha1/id # is the same as the already integrated (not yet saved to config). If they are # the same, no need to reintegrated! # - Note: if there are subtree or exclude changes, update must still be done! # TODO combine with cmd_add(). It also does the same! # TODO implement git dir optimization # - use a bare repository! And ad-hoc checkouts # - And at best do not checkout, just import the git objects into the # current git and let git of the superproject check it out. local_tmp_name = paths.cwd_to_sub_relpath + b"-tmp" # TODO optimize with ls-remote, If the commit/tag hash are equal, don't # download! # subpatch download download_config = DownloadConfig(url=sub_url, revision=revision) # Fetch remote repository/tarball into local folder and checkout the # requested revision. # NOTE: The ".git" folder is already removed! Just the plain tree sub_helper.download(download_config, local_tmp_name) # subpatch unpack # TODO Just hacky local_tmp_abspath = os.path.abspath(join(os.getcwdb(), local_tmp_name)) # Quick and dirty performance improvement. Batch multiple paths together class GitCommandBachter: BATCH_COUNT = 5000 def __init__(self, cmd): self._args = [] self._cmd = cmd def add_and_maybe_exec(self, arg): self._args.append(arg) if len(self._args) >= self.BATCH_COUNT: self.exec_force() def exec_force(self): if len(self._args) == 0: return cmd = self._cmd + self._args subprocess.run(cmd, check=True, cwd=paths.super_abspath) self._args.clear() # Just quick and try remove and copy! # TODO convert this code to "superhelper" implementation with cwd(superx.path): git_rm = GitCommandBachter(["git", "rm", "-q"]) for path in git_ls_tree_in_dir(paths.super_to_sub_relpath): git_rm.add_and_maybe_exec(path) git_rm.exec_force() # and copy # TODO convert this code to "superhelper" implementation os.makedirs(paths.cwd_to_sub_relpath, exist_ok=True) with cwd(local_tmp_abspath): git_add = GitCommandBachter(["git", "add", "-f"]) for root, dirnames, files in os.walk(b"."): for dirname in dirnames: super_to_dest_relpath = join(paths.super_to_sub_relpath, root, dirname) dest_path = join(paths.super_abspath, super_to_dest_relpath) os.makedirs(dest_path, exist_ok=True) for filename in files: src_path = join(root, filename) super_to_dest_relpath = join(paths.super_to_sub_relpath, root, filename) dest_path = join(paths.super_abspath, super_to_dest_relpath) os.rename(src_path, dest_path) super_to_dest_relpath = join(paths.super_to_sub_relpath, root, filename) git_add.add_and_maybe_exec(super_to_dest_relpath) git_add.exec_force() # Remove empty directories in download/cache dir for root, dirnames, files in os.walk(local_tmp_abspath, topdown=False): assert len(files) == 0 for dirname in dirnames: os.rmdir(join(root, dirname)) assert len(os.listdir(local_tmp_abspath)) == 0 os.rmdir(local_tmp_abspath) # TODO update tree checksum in the config # TODO move config_abspath in paths structure with open(config_abspath) as f: config_parts = config_parse(split_with_ts(f.read())) # TODO Code was copied! refactor subproject_config_parts = [(3, "\turl = %s\n" % (sub_url), "url", sub_url)] if revision is not None: subproject_config_parts.append((3, "\trevision = %s\n" % (revision,), "revision", revision)) super_to_sub_relpath_as_str = paths.super_to_sub_relpath.decode("utf8") config_parts = config_remove_section(list(config_parts), "subpatch", super_to_sub_relpath_as_str) config_parts_new = config_add_section(config_parts, "subpatch", super_to_sub_relpath_as_str, subproject_config_parts) with open(config_abspath, "w") as f: f.write(config_unparse(config_parts_new)) # NOTE "add" assumes cwd is toplevel dir of superproject for now! # TODO refactor cleanup this! with cwd(superx.path): superx.helper.add([b".subpatch"]) # TODO reapply patches print(" Done.") # TODO think about this case. Maybe add an argument to make this an error. if not git_diff_in_dir(superx.path, paths.super_to_sub_relpath, staged=True): # There are no changes in the subproject directory and in the metadata. # Nothing has changed! And so there is nothing to commit! # NOTE: So also don't show the commands to commit and inspect the # changes. print("Note: There are no changes in the subproject. Nothing to commit!") return 0 # TODO show diff state # TODO Idea "subpatch status" should print the info/help text. The status # command should be command to get help if a user is lost. superx.helper.print_add_success() return 0 def cmd_add(args, parser): if args.url is None: # TODO make error message nicer raise Exception("Not the correct amount of args") if args.path is not None: # The optional path argument was given. Make some checks if len(args.path) == 0: raise AppException(ErrorCode.INVALID_ARGUMENT, "path is empty") # TODO But something like "/./././" is also invalid" # TODO disallow absolute paths # TODO think about relative paths, whether the must be relative to the # remote origin url! Repo and git-submodule do that # TODO "Can be None" comment is bad. It should be encoded into the variable name # and/or in the type. revision = args.revision # Can be None sub_url = args.url if revision is not None: if not is_valid_revision(revision): # TODO add reason why it's invalid raise AppException(ErrorCode.INVALID_ARGUMENT, "revision '%s' is invalid" % (revision,)) # TODO check with ls-remote that remote is accessible if args.path is None: cwd_to_sub_relpath = get_name_from_repository_url(sub_url) else: cwd_to_sub_relpath = args.path # TODO split into path and name component # sanitize path. Remove double slashes trailing slash # and paths that are outside of the repository # - For now just remove trailing slashes cwd_to_sub_relpath = cwd_to_sub_relpath.rstrip("/") data = find_superproject() superx = check_and_get_superproject_from_data(data) if not superx.configured and superx.typex != SuperprojectType.GIT: # NOTE: It's ok if the superproject is *not* configured. # The 'add' command will auto-configure the superproject if a SCM tool exists! raise AppException(ErrorCode.NOT_IMPLEMENTED_YET, "tbd") url_type = get_url_type(sub_url) # TODO remove additional encode() here. Argument should be bytes object! paths = do_paths(superx.path, cwd_to_sub_relpath.encode("utf8")) if url_type == URLTypes.LOCAL_RELATIVE: # Check the current work directory of the call is the top level # directory of the repository. If not, a relative path on the # command line is not relative to the top level directory and must be # converted. This conversion is a bit complex and error prone. So # just enforce (like also git does) that the current work directory # is the top level repository in that case. # TODO This is a invalid argument execption if not is_cwd_toplevel_directory(paths): raise AppException(ErrorCode.CUSTOM, "When using relative repository URLs, you current work directory must be the toplevel folder of the superproject!") elif url_type == URLTypes.LOCAL_ABSOLUTE: # TODO add reason why it's not supported raise AppException(ErrorCode.CUSTOM, "Absolute local paths to a remote repository are not supported!") # TODO check that git repo has nothing in the index! # But some changes are ok: # TODO add tests that adds a subproject after a "git rm -r" to the same folder. # So the changes are in the filesystem and in the index, but not committed yet. # This should work # For now just ensure that the superproject is a git repo ensure_superproject_is_git(superx) sub_helper = SubHelperGit() # TODO Also check the config file whether a entry already exists if os.path.exists(paths.cwd_to_sub_relpath): # TODO explain what can be done: Either remove the dir or use another name! # TODO fix decode utf8. Should the the system default encoding! # TODO add tests for this! raise AppException(ErrorCode.CUSTOM, "File '%s' alreay exists. Cannot add subproject!" % (paths.cwd_to_sub_relpath.decode("utf8"),)) # TODO maybe add pid to avoid race conditions local_tmp_name = paths.cwd_to_sub_relpath + b"-tmp" # NOTE: Design decision: The output is relative to the current working dir. # The content of '%s' is the remote git name or the path relative to the # current working dir. It's not relative to the top level dir of the git repo. # TODO move this design decision to the website print("Adding subproject '%s' from URL '%s' at revision '%s'..." % (paths.cwd_to_sub_relpath.decode("utf8"), sub_url, sub_helper.get_revision_as_str(revision)), end="") sys.stdout.flush() download_config = DownloadConfig(url=sub_url, revision=revision) # Fetch remote repository/tarball into local folder and checkout the # requested revision. sub_helper.download(download_config, local_tmp_name) config_abspath = join(superx.path, b".subpatch") try: # TODO read as bytes instead of str type with open(config_abspath, "r") as f: config_parts = config_parse(split_with_ts(f.read())) except FileNotFoundError: config_parts = [] # TODO Check if section and subsection already exists! # TODO currently this always uses "\t" for indention. Try to use the style # that is already used in the subpatch file. E.g. four-spaces, two-spaces # or no-spaces. # NOTE: DDX/Design decision: The URL is taken as a verbatim copy # of the argument. If there is a trailing slash in the argument, then the # trailing slash is also in the config file. It's not sanitized. It's the # same behavior as 'git submodule' does. subproject_config_parts = [(3, "\turl = %s\n" % (sub_url), "url", sub_url)] if revision is not None: subproject_config_parts.append((3, "\trevision = %s\n" % (revision,), "revision", revision)) config_parts_new = config_add_section(config_parts, "subpatch", paths.super_to_sub_relpath.decode("utf8"), subproject_config_parts) with open(config_abspath, "w") as f: f.write(config_unparse(config_parts_new)) # Move files into place os.rename(local_tmp_name, paths.cwd_to_sub_relpath) # TODO in case of failure, remove download git dir! # Add files for committing # TODO git_add must use "-f" otherwise ignore files are used and not all files are added! superx.helper.add([paths.cwd_to_sub_relpath]) with cwd(superx.path): superx.helper.add([b".subpatch"]) print(" Done.") # TODO Idea "subpatch status" should print the info/help text. The status # command should be command to get help if a user is lost. superx.helper.print_add_success() # TODO Maybe prepare a commit/patch/change message return 0 def show_version(args): print("subpatch version %s" % (__version__,)) return 0 def show_info(args): print("homepage: https://subpatch.net") print("git repo: https://github.com/lengfeld/subpatch") print("license: %s" % (__LICENSE__,)) # TODO add GPL license text/note return 0 def ensure_superproject_is_configured(superx): # Design decision: # Having a empty ".subpatch" config file is something different than having # no config file. It means that subptach is not yet configured for the project. # TODO document this design decision somewhere else! if not superx.configured: raise AppException(ErrorCode.SUPERPROJECT_NOT_CONFIGURED) def ensure_superproject_is_git(superx): if superx.typex != SuperprojectType.GIT: raise AppException(ErrorCode.WORKS_ONLY_IN_A_GIT_REPO) class SuperprojectType(Enum): PLAIN = b"plain" GIT = b"git" @dataclass(frozen=True) class Superproject: path: bytes helper: object configured: bool typex: SuperprojectType def check_and_get_superproject_from_data(data): def get_helper_from_scm_type(scm_type): if scm_type == SCMType.GIT: return SuperprojectType.GIT, SuperHelperGit() else: raise AppException(ErrorCode.NOT_IMPLEMENTED_YET, "SCM tool '%s' not implemented yet" % (scm_type,)) if data.scm_type is not None: # There is a SCM tool if data.super_path is None: # And there is no subpatch config file super_type, helper = get_helper_from_scm_type(data.scm_type) path = data.scm_path return Superproject(path, helper, False, super_type) else: # And there is a subpatch config file! # Check if the paths match up: if data.super_path != data.scm_path: # TODO add text here! raise AppException(ErrorCode.NOT_IMPLEMENTED_YET, "subpatch config file is not at the root of the SCM repository!") super_type, helper = get_helper_from_scm_type(data.scm_type) path = data.super_path return Superproject(path, helper, True, super_type) else: # Current working directory is not in a SCM system if data.super_path is None: # And there is no subpatch config file raise AppException(ErrorCode.SUPERPROJECT_NOT_CONFIGURED) else: # And there is a subpatch config file helper = SuperHelperPlain() super_type = SuperprojectType.PLAIN path = data.super_path return Superproject(path, helper, True, super_type) # TODO Document own internal path library # - e.g. b"" is the same as b"." and the later is not used! # TODO add tests def is_relpath(path): if len(path) == 0: return True if path[0] == ord("/"): return False return True # NOTE: It's the inverse of is_relpath() # TODO add tests def is_abspath(path): if len(path) == 0: return False if path[0] != ord("/"): return False return True @dataclass(frozen=True) class Paths: super_abspath: bytes super_to_cwd_relpath: bytes super_to_sub_relpath: bytes cwd_to_sub_relpath: bytes # Can be ".." or "../.." if cwd is inside a subproject sub_name: bytes # Can be b"" when subproject is at the toplevel directory # Paths documentation and naming # # Example ../folder/superproject/dirA/dirB/subproject/ # ^^^^cwd # # super_abspath: ../folder/superproject # other name: "top level directory" TODO unify # config_abspath: ../folder/superproject/.subpatch # super_to_sub_relpath: dirA/dirB/subproject # super_to_cwd_relpath: dirA # cwd_to_sub_relpath: dirB/subproject # sub_name: subproject # # TODO finalized this naming # Note on path/filesystem nameing # - path: "dir/test", "/dir/test" # - file: a file in a filesystem! # - filename: can be a dir name or a regular file name # reduandent words: # - maybe pathname # - maybe dirname # # TODO test and fix all the edge cases # e.g. cwd_to_sub_relpath="../symbolic-link/../../dir" # - symbolic links # - ".." pointing outside of repo # TODO fix function naming def do_paths(super_abspath, cwd_to_sub_relpath): assert is_abspath(super_abspath) assert is_relpath(cwd_to_sub_relpath) # TODO test and ensure that paths are normalized!! # Sanitize input path if cwd_to_sub_relpath != b"": cwd_to_sub_relpath = os.path.relpath(cwd_to_sub_relpath) if cwd_to_sub_relpath == b".": cwd_to_sub_relpath = b"" # Assert that super_path matches, so stripping is ok cwd_abspath = os.path.abspath(os.getcwdb()) assert cwd_abspath.startswith(super_abspath) super_to_cwd_relpath = cwd_abspath[len(super_abspath) + 1:] super_to_sub_relpath = join(super_to_cwd_relpath, cwd_to_sub_relpath) # Remove trailing slash if there is any if super_to_cwd_relpath != b"": super_to_sub_relpath = os.path.relpath(super_to_sub_relpath) sub_name = os.path.basename(super_to_sub_relpath) def my_join(a, b): if a == b"": return b if b == b"": return a # If b contains ".." or "../..", strip path components, instead of # adding ".." return os.path.relpath(join(a, b)) assert super_to_sub_relpath == my_join(super_to_cwd_relpath, cwd_to_sub_relpath) return Paths(super_abspath, super_to_cwd_relpath, super_to_sub_relpath, cwd_to_sub_relpath, sub_name) # TODO add "-z" option to be safe againts any chars in the path # TODO add escpaing for "evil" chars in non "-z" output # TODO add not about plumbing command def cmd_list(args, parser): data = find_superproject() superx = check_and_get_superproject_from_data(data) ensure_superproject_is_configured(superx) config_abspath = join(data.super_path, b".subpatch") with open(config_abspath) as f: config_parts = config_parse(split_with_ts(f.read())) subprojects = subprojects_parse(config_parts) for subproject in subprojects: print("%s" % (subproject.path,)) return 0 # Parse the "-z" output of git commands # TODO rework as a generator def parse_z(b: bytes) -> list[bytes]: if b == b"": # Special case. b"".split(b"\0") is [b""], but # we want a list without elements here. return [] return b.rstrip(b"\0").split(b"\0") def git_diff_in_dir(top_dir, subdir, staged=False): # TODO verify that top_dir is the toplevel dir in the repo # -> Refactor to git object or class that checks the top_dir # TODO check that subdir is a relative and a sane path! cmd = ["git", "diff", "--name-only", "-z"] if staged: cmd += ["--staged"] cmd += ["--", subdir] p = Popen(cmd, stdout=PIPE, cwd=top_dir) stdout, _ = p.communicate() if p.returncode != 0: raise Exception("error here") return len(stdout) != 0 # NOTE: argument and output is not cwd aware. It's always relative to the # toplevel of the git repository. def git_ls_tree_in_dir(subdir): assert isinstance(subdir, bytes) if subdir == b"": subdir = b"." p = Popen(["git", "ls-tree", "--full-tree", "-r", "--name-only", "-z", "HEAD", subdir], stdout=PIPE) stdout, _ = p.communicate() if p.returncode != 0: raise Exception("TODO error here") return parse_z(stdout) def git_diff_name_only(staged=False): # NOTE: "git diff" does not depend on the cwd inside the repo cmd = ["git", "diff", "--name-only", "-z"] if staged: cmd += ["--staged"] p = Popen(cmd, stdout=PIPE) stdout, _ = p.communicate() if p.returncode != 0: raise Exception("error here") return parse_z(stdout) # NOTE: This depends on the cwd for now. git ls-files has no option to force # listing files from the top level directory. # TODO fix that def git_ls_files_untracked(): # TODO refactor common code # NOTE: # - Use "--full-name" to make the paths relative to the toplevel # directory, not the current work directory. # - Use "--no-empty-directory" to avoid printing dirs that contain only # ignored files. p = Popen(["git", "ls-files", "--exclude-standard", "-o", "--directory", "-z", "--full-name", "--no-empty-directory"], stdout=PIPE) stdout, _ = p.communicate() if p.returncode != 0: raise Exception("error here") return parse_z(stdout) # TODO add note that the output of "status" is not an API/plumbing. Should not be used in scripts def cmd_status(args, parser): data = find_superproject() superx = check_and_get_superproject_from_data(data) ensure_superproject_is_configured(superx) ensure_superproject_is_git(superx) # TODO add argument for a path to a single subproject and then only list # the status of this subproject # TODO allow the cwd to select the subproject paths = do_paths(superx.path, b"") if not is_cwd_toplevel_directory(paths): raise AppException(ErrorCode.NOT_IMPLEMENTED_YET, "Current work directory must be the toplevel directory of the repo for now!") # TODO add plumbing commands for scripts! # TODO add colorscheme for output, e.g. like git status does # TODO handle stdout fd is pipe/file and not tty. config_abspath = join(superx.path, b".subpatch") with open(config_abspath) as f: config_parts = config_parse(split_with_ts(f.read())) subprojects = subprojects_parse(config_parts) # Force the generator before closing the file subprojects = list(subprojects) # Get all paths to subprojects # These are relative paths: super_to_sub_relpath # TODO check paths in config file for sanity. Should not contain trialing slash or double slashes! # The paths must be canoncical for comparing subproject_paths = [subproject.path for subproject in subprojects] # TODO does the concept of staged and unstaged files als exists in other # cvs systems diff_files_not_staged = git_diff_name_only() diff_files_staged = git_diff_name_only(staged=True) # TODO 'git_ls_files_untracked' depends on the cwd for now! Cwd must be the # root of the directory for now. ls_files_untrack = git_ls_files_untracked() # TODO refactor this struct and the following code to a function! And make # to interface for other cvs @dataclass class Changes: untracked: int = 0 unstaged: int = 0 uncommitted: int = 0 subproject_changes = {} for path in subproject_paths: subproject_changes[path] = Changes() for subproject_path in subproject_paths: # TODO str vs bytes mismatch subproject_path_bytes = subproject_path.encode("utf8") changes = subproject_changes[subproject_path] for path in ls_files_untrack: if path.startswith(subproject_path_bytes): changes.untracked += 1 for path in diff_files_staged: if path.startswith(subproject_path_bytes): changes.uncommitted += 1 for path in diff_files_not_staged: if path.startswith(subproject_path_bytes): changes.unstaged += 1 if len(subprojects) == 0: # Early return. Nothing to print! return 0 print("NOTE: The format of the output is human-readable and unstable. Do not use in scripts!") print("NOTE: The format is markdown currently. Will mostly change in the future.") print("") for i, subproject in enumerate(subprojects): # TODO Idea: make it valid markdown output # TODO Idea: For every cvs superproject (superhelper) make the output # like the cvs styled of console output. Subpach should look like git # for git superprojects. And look like svn for svn superprojects. # NOTE: these two ideas are conflicting! # TODO Think about other cvs systems print("# subproject at '%s'" % (subproject.path,)) print("") if subproject.url is not None: print("* was integrated from URL: %s" % (subproject.url,)) if subproject.revision is not None: print("* has integrated revision: %s" % (subproject.revision,)) # TODO Maybe included whether it "tracks a branch" or it was a git tag p = subproject.path changes = subproject_changes[p] # NOTE Listing the modified files seems to much like "git status". # There is already "git status". The output just referneces the git # commands To inspect the changes. if changes.untracked > 0: print("""\ * There are n=%d untracked files and/or directories: - To see them execute: `git status %s` `git ls-files --exclude-standard -o %s` - Use `git add %s` to add all of them - Use `git add %s/` to just add some of them - Use `rm ` to remove them""" % (changes.untracked, p, p, p, p)) if changes.unstaged > 0: print("""\ * There are n=%d modified files not staged for commit: - To see them execute: `git status %s` or `git diff %s` - Use `git add %s` to update what will be committed - Use `git restore %s` to discard changes in working directory""" % (changes.unstaged, p, p, p, p)) if changes.uncommitted > 0: # TODO check git restore print("""\ * There are n=%d modified files that are staged, but not committed: - To see them execute: `git status %s` or `git diff --staged %s` - Use `git commit %s` to commit the changes - Use `git restore --staged %s` to unstage""" % (changes.uncommitted, p, p, p, p)) # TODO check restore staged # TODO add command "git restore --staged --worktree -- xxxpath/" # to remove changes staged in the index from the index and the # working tree But leave changes in the working tree/unstaged # untouched This helps because it does ot leave new files in the # working tree as "git restore --staged" will done. if i + 1 < len(subprojects): # Add a empty line between the subprojects print("") # TODO list amount of patches return 0 def main_wrapped(): # TODO maybe add 'epilog' again parser = ArgumentParser(description='Adding subprojects into a git repo, the superproject.') parser.add_argument("--version", "-v", dest="version", action="store_true", default=False, help="Show version of program") parser.add_argument("--info", dest="info", action="store_true", default=False, help="Show more information, like homepage, repo and license") subparsers = parser.add_subparsers() parser_add = subparsers.add_parser("add", help="Fetch and add a subproject") parser_add.set_defaults(func=cmd_add) parser_add.add_argument(dest="url", type=str, help="URL or path to git repo") parser_add.add_argument(dest="path", type=str, default=None, nargs='?', help="folder or path in the local repo") parser_add.add_argument("-r", "--revision", dest="revision", type=str, help="Specify the revision to integrate. Can be a branch name, tag name or commit id.") parser_update = subparsers.add_parser("update", help="Fetch and update a subproject") parser_update.set_defaults(func=cmd_update) parser_update.add_argument(dest="path", type=str, default=None, nargs='?', help="path to subproject") parser_update.add_argument("--url", dest="url", type=str, help="URL or path to the remote git repo") parser_update.add_argument("-r", "--revision", dest="revision", type=str, help="Specify the revision to integrate. Can be a branch name, tag name or commit id.") # TODO add argument to print the status of only a single subproject parser_status = subparsers.add_parser("status", help="Prints a summary of all subprojects") parser_status.set_defaults(func=cmd_status) parser_list = subparsers.add_parser("list", help="List all subprojects") parser_list.set_defaults(func=cmd_list) parser_help = subparsers.add_parser("help", help="Also shows the help message") parser_help.set_defaults(func=cmd_help) args = parser.parse_args() # Just for testing. A bit ugly to have this in the production code. if os.environ.get("HANG_FOR_TEST", "0") == "1": time.sleep(5) if args.version: ret = show_version(args) elif args.info: ret = show_info(args) else: # Workaround for help if hasattr(args, "func"): ret = args.func(args, parser) else: ret = nocommand(args, parser) return ret def main(): try: ret = main_wrapped() except AppException as e: # TODO allow to append a generic error message for all messages, not only for invalid argument. if e._code == ErrorCode.NOT_IMPLEMENTED_YET: # TODO the message should show the github issue url! print("Error: Feature not implemented yet: %s" % (e,), file=sys.stderr) elif e._code == ErrorCode.WORKS_ONLY_IN_A_GIT_REPO: print("Error: Feature works only inside a git repository!", file=sys.stderr) elif e._code == ErrorCode.SUPERPROJECT_NOT_CONFIGURED: # TODO add steps to resolve the issue. e.g. touching the file print("Error: subpatch not yet configured for superproject!", file=sys.stderr) elif e._code == ErrorCode.INVALID_ARGUMENT: # TODO change structure of errors. It contains two colons now. # Looks ugly. print("Error: Invalid argument: %s" % (e,), file=sys.stderr) elif e._code == ErrorCode.CUSTOM: print("Error: %s" % (e,), file=sys.stderr) else: assert e._code == ErrorCode.UNKNOWN # TODO find a better name for UNKNOWN # Don't print a message here. The caller has already written the # message. # TODO maybe it still better that the error is printed here and not # printed by the caller! pass ret = 4 except KeyboardInterrupt: print("Interrupted!", file=sys.stderr) # TODO What is the correct/best error code? ret = 3 sys.exit(ret) if __name__ == '__main__': main() raise Exception("Never reached!")