PK!҂kkis_git_repo_clean/__init__.pyfrom .meta import version from .check import check, checkSync __all__ = ["check", "checkSync", "version"] PK! 1Nis_git_repo_clean/check.py# # TODO: find a cross-platform way to rely only on the returncode for untracked # files. Currently it returns stdout/stderr which should be unnecessary. # # ------- # # Imports # # ------- # from textwrap import dedent from .utils import iif, resolveAll import asyncio import os import subprocess # ---- # # Init # # ---- # runAsync = asyncio.create_subprocess_exec run = subprocess.run headExistsCmd = ["git", "rev-parse", "HEAD"] isAGitRepoCmd = ["git", "rev-parse", "--is-inside-work-tree"] # Any way to remove the double negative? allFilesAreTrackedCmd = [ "git", "ls-files", "--other", "--directory", "--exclude-standard", "--no-empty-directory", ] emptyGitTree = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" # ---- # # Main # # ---- # async def check(cwd=None): cwd = cleanCwd(cwd) isAGitRepo_proc = await runAsync( *isAGitRepoCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ) isAGitRepo_result = await isAGitRepo_proc.wait() if isAGitRepo_result != 0: raise Exception(f"'{cwd}' is not a git repository") headExists_proc = await runAsync( *headExistsCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ) headExists_result = await headExists_proc.wait() headExists = headExists_result == 0 hasNoChangesCmd = getHasNoChangesCmd(headExists) checks = [ runAsync( *hasNoChangesCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ), runAsync( *allFilesAreTrackedCmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, cwd=cwd, ), ] noChanges_proc, allTracked_proc = await resolveAll(checks) results = await resolveAll( [noChanges_proc.wait(), allTracked_proc.communicate()] ) noChanges_result, allTracked_result = results if allTracked_proc.returncode != 0 or noChanges_result != 0: return False stdout, _stderr = allTracked_result return len(stdout.decode("utf-8")) == 0 def checkSync(cwd=None): cwd = cleanCwd(cwd) isAGitRepo = ( run( isAGitRepoCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ).returncode == 0 ) if not isAGitRepo: raise Exception(f"'{cwd} is not a git repository") headExists = ( run( headExistsCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ).returncode == 0 ) hasNoStagedChangesCmd = getHasNoChangesCmd(headExists, staged=True) hasNoUnstagedChangesCmd = getHasNoChangesCmd(headExists) allTracked_result = run( allFilesAreTrackedCmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=cwd, ) hasNoStagedChanges_result = run( hasNoStagedChangesCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ) hasNoUnstagedChanges_result = run( hasNoUnstagedChangesCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ) return ( hasNoStagedChanges_result.returncode == 0 and hasNoUnstagedChanges_result.returncode == 0 and allTracked_result.returncode == 0 and len(allTracked_result.stdout) == 0 ) # ------- # # Helpers # # ------- # def cleanCwd(cwd): if cwd is None: return os.getcwd() elif not isinstance(cwd, str): raise TypeError( dedent( f"""\ cwd must be an instance of str type(cwd): {type(cwd)} cwd: {cwd} """ ) ) return cwd def getHasNoChangesCmd(headExists, staged=False): cmd = [ "git", "diff-index", "--quiet", iif(headExists, "HEAD", emptyGitTree), "--", ] if staged: cmd.insert(2, "--cached") return cmd PK!@Bq q is_git_repo_clean/cli.py# ------- # # Imports # # ------- # from .meta import version from traceback import format_exc from textwrap import dedent from types import SimpleNamespace as o from .check import checkSync from .utils import iif import os # ---- # # Init # # ---- # arguments = set(["--dir", "--silent"]) helpOrVersion = set(["--help", "--version"]) twoLineSeps = os.linesep + os.linesep usage = dedent( f""" Usage is-git-repo-clean [options] is-git-repo-clean (--help | --version) Options --dir: path to the git repo to test. Defaults to `os.getcwd()` --silent: a flag which disables output Returns : 0: yes 1: no 2: 3: dir is not a git repository 4: unexpected error occurred """ ) # ---- # # Main # # ---- # def getIsGitRepoClean(args): result = o(stdout=None, stderr=None, code=None) numArgs = len(args) if numArgs == 1: if args[0] == "--help": result.stdout = usage result.code = 0 return result elif args[0] == "--version": result.stdout = version result.code = 0 return result argsObj = validateAndParseArgs(args, result) if argsObj is result: return result isSilent = argsObj.silent try: isClean = checkSync(argsObj.dir or None) if not isSilent: if isClean: result.stdout = "yes" else: result.stderr = "no" result.code = iif(isClean, 0, 1) return result except Exception as e: if "is not a git repository" in str(e): if not isSilent: result.stderr = "dir is not a git repository" result.code = 3 else: if not isSilent: result.stderr = ( f"unexpected error occurred{twoLineSeps}" + format_exc() ) result.code = 4 return result # ------- # # Helpers # # ------- # def validateAndParseArgs(args, result): i = 0 argsObj = o(dir=None, silent=False) while i < len(args): arg = args[i] if arg not in arguments: if arg in helpOrVersion: result.stderr = f"'{arg}' must be the only argument when passed" else: result.stderr = f"invalid option '{arg}'" result.stderr += os.linesep + usage result.code = 2 return result if arg == "--dir": if i == len(args) - 1: result.stderr = "'--dir' must be given a value" result.stderr += os.linesep + usage result.code = 2 return result argsObj.dir = args[i + 1] i += 1 else: argsObj.silent = True i += 1 return argsObj PK!SWis_git_repo_clean/meta.pyversion = "0.3.4" PK!J  is_git_repo_clean/script.pyfrom .cli import getIsGitRepoClean import sys def printErr(msg): print(msg, file=sys.stderr) def main(): result = getIsGitRepoClean(sys.argv[1:]) if result.stdout: print(result.stdout) if result.stderr: printErr(result.stderr) exit(result.code) PK!KKis_git_repo_clean/utils.pyfrom asyncio import gather # # I like to think the common use-case for 'gather' is to cancel all async # operations should any one of them fail. This is not how 'gather' was # implemented though which is why I wrote this function # async def resolveAll(awaitables): gatherAll = None try: gatherAll = gather(*awaitables) result = await gatherAll finally: if gatherAll: gatherAll.cancel() return result def iif(condition, whenTruthy, whenFalsey): if condition: return whenTruthy else: return whenFalsey PK!4 Everyone is permitted to copy and distribute verbatim or modified copies of this license document, and changing it is allowed as long as the name is changed. DO WHAT THE FUCK YOU WANT TO BUT IT'S NOT MY FAULT PUBLIC LICENSE TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION 0. You just DO WHAT THE FUCK YOU WANT TO. 1. Do not hold the author(s), creator(s), developer(s) or distributor(s) liable for anything that happens or goes wrong with your use of the work. PK!H[T.:C2is_git_repo_clean-0.3.4.dist-info/entry_points.txtN+I/N.,(),M,-J-MIM̳,ăD"zVy\\PK!HڽTU'is_git_repo_clean-0.3.4.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H *is_git_repo_clean-0.3.4.dist-info/METADATAV]o6}篸C4 *moF:;(bF-.T\R$޲%^^y.p 墴'>kM,JxstNOQDttR33Z % w"tvA- T'U:k ^$(x6}xRa2E:M_d:O/Gconƣ2 =GdD9Hn2UVi T{C 7 AY\31˯ƍqJZڄUA!t{%4S%#|74)|9QW]yU=u.+Vs:#cWI_4zиY$w9rtOu?G9J2V+ ]>Z(RgA6coYFᨥb"i끎tUvD ͷvSeGܹX9QM*(VN6 |tTx١xQB0>[C:K8ͮSPKI4eR BR4X-E {B|g5fXo}KCS' {h>yv[%djoGU_}aOɭ j1Ԭw+U1(:ԲSIz(2~in[D1--̍萹 (V>|[W;_PK!҂kkis_git_repo_clean/__init__.pyPK! 1Nis_git_repo_clean/check.pyPK!@Bq q kis_git_repo_clean/cli.pyPK!SWis_git_repo_clean/meta.pyPK!J  [is_git_repo_clean/script.pyPK!KKis_git_repo_clean/utils.pyPK!4