|
1 | 1 | """Various with-statement context utilities.""" |
2 | 2 |
|
3 | | -from contextlib import contextmanager |
| 3 | +from contextlib import AbstractContextManager, contextmanager |
4 | 4 | from importlib import import_module |
5 | 5 | from multiprocessing.connection import Pipe |
6 | 6 | import errno |
7 | 7 | import inspect |
8 | 8 | import os |
9 | 9 | import pickle |
10 | 10 | import signal |
| 11 | +import subprocess |
11 | 12 | import sys |
12 | 13 | import threading |
13 | 14 | import traceback |
14 | 15 |
|
| 16 | +from .cli.exceptions import UserException |
15 | 17 | from .process import namespaces |
| 18 | +from .sequences import predicate_split |
16 | 19 |
|
17 | 20 |
|
18 | 21 | # Ideas and code for SplitExec have been borrowed from withhacks |
@@ -272,6 +275,59 @@ def _child_setup(self): |
272 | 275 | namespaces.simple_unshare(hostname=self._hostname, **self._namespaces) |
273 | 276 |
|
274 | 277 |
|
| 278 | +class GitStash(AbstractContextManager): |
| 279 | + """Context manager for stashing untracked or modified/uncommitted files.""" |
| 280 | + |
| 281 | + def __init__(self, path, staged=False): |
| 282 | + self.path = path |
| 283 | + self._staged = ['--keep-index'] if staged else [] |
| 284 | + self._stashed = False |
| 285 | + |
| 286 | + def __enter__(self): |
| 287 | + """Stash all untracked or modified files in working tree.""" |
| 288 | + # check for untracked or modified/uncommitted files |
| 289 | + try: |
| 290 | + p = subprocess.run( |
| 291 | + ['git', 'status', '--porcelain=1', '-u'], |
| 292 | + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, |
| 293 | + cwd=self.path, encoding='utf8', check=True) |
| 294 | + except subprocess.CalledProcessError: |
| 295 | + raise ValueError(f'not a git repo: {self.path}') |
| 296 | + |
| 297 | + # split file changes into unstaged vs staged |
| 298 | + unstaged, staged = predicate_split(lambda x: x[1] == ' ', p.stdout.splitlines()) |
| 299 | + |
| 300 | + # don't stash when no relevant changes exist |
| 301 | + if self._staged: |
| 302 | + if not unstaged: |
| 303 | + return |
| 304 | + elif not p.stdout: |
| 305 | + return |
| 306 | + |
| 307 | + # stash all existing untracked or modified/uncommitted files |
| 308 | + try: |
| 309 | + subprocess.run( |
| 310 | + ['git', 'stash', 'push', '-u', '-m', 'pkgcheck scan --commits'] + self._staged, |
| 311 | + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, |
| 312 | + cwd=self.path, check=True, encoding='utf8') |
| 313 | + except subprocess.CalledProcessError as e: |
| 314 | + error = e.stderr.splitlines()[0] |
| 315 | + raise UserException(f'git failed stashing files: {error}') |
| 316 | + self._stashed = True |
| 317 | + |
| 318 | + def __exit__(self, _exc_type, _exc_value, _traceback): |
| 319 | + """Apply any previously stashed files back to the working tree.""" |
| 320 | + if self._stashed: |
| 321 | + try: |
| 322 | + subprocess.run( |
| 323 | + ['git', 'stash', 'pop'], |
| 324 | + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, |
| 325 | + cwd=self.path, check=True, encoding='utf8') |
| 326 | + except subprocess.CalledProcessError as e: |
| 327 | + error = e.stderr.splitlines()[0] |
| 328 | + raise UserException(f'git failed applying stash: {error}') |
| 329 | + |
| 330 | + |
275 | 331 | @contextmanager |
276 | 332 | def chdir(path): |
277 | 333 | """Context manager that changes the current working directory. |
|
0 commit comments