stgit/lib/stack.py

   1 """A Python class hierarchy wrapping the StGit on-disk metadata."""
   2 import re
   3
   4 from stgit.config import config
   5 from stgit.exception import StackException
   6 from stgit.lib import log, stackupgrade
   7 from stgit.lib.git import Repository
   8 from stgit.lib.git.branch import Branch, BranchException
   9
  10
  11 def _stack_state_ref(stack_name):
  12     """Reference to stack state metadata. A.k.a. the stack's "log"."""
  13     return 'refs/heads/%s.stgit' % (stack_name,)
  14
  15
  16 def _patch_ref(stack_name, patch_name):
  17     """Reference to a named patch's commit."""
  18     return 'refs/patches/%s/%s' % (stack_name, patch_name)
  19
  20
  21 def _patch_ref_prefix(stack_name):
  22     return _patch_ref(stack_name, '')
  23
  24
  25 class PatchOrder:
  26     """Keeps track of patch order, and which patches are applied.
  27
  28     Works with patch names, not actual patches.
  29
  30     """
  31
  32     def __init__(self, state):
  33         self._applied = tuple(state.applied)
  34         self._unapplied = tuple(state.unapplied)
  35         self._hidden = tuple(state.hidden)
  36
  37     @property
  38     def applied(self):
  39         return self._applied
  40
  41     @property
  42     def unapplied(self):
  43         return self._unapplied
  44
  45     @property
  46     def hidden(self):
  47         return self._hidden
  48
  49     @property
  50     def all(self):
  51         return self.applied + self.unapplied + self.hidden
  52
  53     @property
  54     def all_visible(self):
  55         return self.applied + self.unapplied
  56
  57     def set_order(self, applied, unapplied, hidden):
  58         self._applied = tuple(applied)
  59         self._unapplied = tuple(unapplied)
  60         self._hidden = tuple(hidden)
  61
  62     def rename_patch(self, old_name, new_name):
  63         for attr in ['_applied', '_unapplied', '_hidden']:
  64             patch_list = list(getattr(self, attr))
  65             try:
  66                 index = patch_list.index(old_name)
  67             except ValueError:
  68                 continue
  69             else:
  70                 patch_list[index] = new_name
  71                 setattr(self, attr, tuple(patch_list))
  72                 break
  73         else:
  74             raise AssertionError('"%s" not found in patchorder' % old_name)
  75
  76
  77 class Patches:
  78     """Interface for managing patch references."""
  79
  80     def __init__(self, stack, state):
  81         self._stack = stack
  82
  83         # Ensure patch refs in repository match those from stack state.
  84         repository = stack.repository
  85         patch_ref_prefix = _patch_ref_prefix(stack.name)
  86
  87         state_patch_ref_map = {
  88             _patch_ref(stack.name, pn): commit for pn, commit in state.patches.items()
  89         }
  90
  91         state_patch_refs = set(state_patch_ref_map)
  92         repo_patch_refs = {
  93             ref for ref in repository.refs if ref.startswith(patch_ref_prefix)
  94         }
  95
  96         delete_patch_refs = repo_patch_refs - state_patch_refs
  97         create_patch_refs = state_patch_refs - repo_patch_refs
  98         update_patch_refs = {
  99             ref
 100             for ref in state_patch_refs - create_patch_refs
 101             if state_patch_ref_map[ref].sha1 != repository.refs.get(ref).sha1
 102         }
 103
 104         if create_patch_refs or update_patch_refs or delete_patch_refs:
 105             repository.refs.batch_update(
 106                 msg='restore from stack state',
 107                 create=[(ref, state_patch_ref_map[ref]) for ref in create_patch_refs],
 108                 update=[(ref, state_patch_ref_map[ref]) for ref in update_patch_refs],
 109                 delete=delete_patch_refs,
 110             )
 111
 112     def _patch_ref(self, name):
 113         return _patch_ref(self._stack.name, name)
 114
 115     def __contains__(self, name):
 116         return self._stack.repository.refs.exists(self._patch_ref(name))
 117
 118     def __getitem__(self, name):
 119         return self._stack.repository.refs.get(self._patch_ref(name))
 120
 121     def __iter__(self):
 122         patch_ref_prefix = _patch_ref_prefix(self._stack.name)
 123         for ref in self._stack.repository.refs:
 124             if ref.startswith(patch_ref_prefix):
 125                 yield ref[len(patch_ref_prefix) :]
 126
 127     def name_from_sha1(self, partial_sha1):
 128         for pn in self:
 129             if self[pn].sha1.startswith(partial_sha1.lower()):
 130                 return pn
 131         else:
 132             return None
 133
 134     def is_name_valid(self, name):
 135         if '/' in name:
 136             # TODO slashes in patch names could be made to be okay
 137             return False
 138         ref = _patch_ref(self._stack.name, name)
 139         p = self._stack.repository.run(['git', 'check-ref-format', ref])
 140         p.returns([0, 1]).discard_stderr().discard_output()
 141         return p.exitcode == 0
 142
 143     def new(self, name, commit, msg):
 144         assert name not in self
 145         assert self.is_name_valid(name)
 146         self._stack.repository.refs.set(self._patch_ref(name), commit, msg)
 147
 148     def update(self, name, commit, msg):
 149         old_sha1 = self[name].sha1
 150         if old_sha1 != commit.sha1:
 151             self._stack.repository.refs.set(self._patch_ref(name), commit, msg)
 152             self._stack.repository.copy_notes(old_sha1, commit.sha1)
 153
 154     def rename(self, old_name, new_name, msg):
 155         commit = self[old_name]
 156         self._stack.repository.refs.delete(self._patch_ref(old_name))
 157         self._stack.repository.refs.set(self._patch_ref(new_name), commit, msg)
 158
 159     def delete(self, name):
 160         self._stack.repository.refs.delete(self._patch_ref(name))
 161
 162     def make_name(self, raw, unique=True, lower=True, allow=(), disallow=()):
 163         """Make a unique and valid patch name from provided raw name.
 164
 165         The raw name may come from a filename, commit message, or email subject line.
 166
 167         The generated patch name will meet the rules of `git check-ref-format` along
 168         with some additional StGit patch name rules.
 169
 170         """
 171         default_name = 'patch'
 172
 173         for line in raw.split('\n'):
 174             if line:
 175                 break
 176
 177         if not line:
 178             line = default_name
 179
 180         if lower:
 181             line = line.lower()
 182
 183         parts = []
 184         for part in line.split('/'):
 185             # fmt: off
 186             part = re.sub(r'\.lock$', '', part)    # Disallowed in Git refs
 187             part = re.sub(r'^\.+|\.+$', '', part)  # Cannot start or end with '.'
 188             part = re.sub(r'\.+', '.', part)       # No consecutive '.'
 189             part = re.sub(r'[^\w.]+', '-', part)   # Non-word and whitespace to dashes
 190             part = re.sub(r'-+', '-', part)        # Squash consecutive dashes
 191             part = re.sub(r'^-+|-+$', '', part)    # Remove leading and trailing dashes
 192             # fmt: on
 193             if part:
 194                 parts.append(part)
 195
 196         long_name = '/'.join(parts)
 197
 198         # TODO: slashes could be allowed in the future.
 199         long_name = long_name.replace('/', '-')
 200
 201         if not long_name:
 202             long_name = default_name
 203
 204         assert self.is_name_valid(long_name)
 205
 206         name_len = config.getint('stgit.namelength')
 207
 208         words = long_name.split('-')
 209         short_name = words[0]
 210         for word in words[1:]:
 211             new_name = '%s-%s' % (short_name, word)
 212             if name_len <= 0 or len(new_name) <= name_len:
 213                 short_name = new_name
 214             else:
 215                 break
 216         assert self.is_name_valid(short_name)
 217
 218         if not unique:
 219             return short_name
 220
 221         unique_name = short_name
 222         while unique_name not in allow and (
 223             unique_name in self or unique_name in disallow
 224         ):
 225             m = re.match(r'(.*?)(-)?(\d+)$', unique_name)
 226             if m:
 227                 base, sep, n_str = m.groups()
 228                 n = int(n_str) + 1
 229                 if sep:
 230                     unique_name = '%s%s%d' % (base, sep, n)
 231                 else:
 232                     unique_name = '%s%d' % (base, n)
 233             else:
 234                 unique_name = '%s-1' % unique_name
 235
 236         assert self.is_name_valid(unique_name)
 237         return unique_name
 238
 239
 240 class Stack(Branch):
 241     """Represents a StGit stack.
 242
 243     A StGit stack is a Git branch with extra metadata for patch stack state.
 244
 245     """
 246
 247     def __init__(self, repository, name):
 248         super().__init__(repository, name)
 249         if not stackupgrade.update_to_current_format_version(repository, name):
 250             raise StackException('%s: branch not initialized' % name)
 251         state = log.get_stack_state(self.repository, self.state_ref)
 252         self.patchorder = PatchOrder(state)
 253         self.patches = Patches(self, state)
 254
 255     @property
 256     def base(self):
 257         if self.patchorder.applied:
 258             return self.patches[self.patchorder.applied[0]].data.parent
 259         else:
 260             return self.head
 261
 262     @property
 263     def top(self):
 264         """Commit of the topmost patch, or the stack base if no patches are applied."""
 265         if self.patchorder.applied:
 266             return self.patches[self.patchorder.applied[-1]]
 267         else:
 268             # When no patches are applied, base == head.
 269             return self.head
 270
 271     def set_parents(self, remote, branch):
 272         if remote:
 273             self.set_parent_remote(remote)
 274         if branch:
 275             self.set_parent_branch(branch)
 276             config.set('branch.%s.stgit.parentbranch' % self.name, branch)
 277
 278     @property
 279     def protected(self):
 280         return config.getbool('branch.%s.stgit.protect' % self.name)
 281
 282     @protected.setter
 283     def protected(self, protect):
 284         protect_key = 'branch.%s.stgit.protect' % self.name
 285         if protect:
 286             config.set(protect_key, 'true')
 287         elif self.protected:
 288             config.unset(protect_key)
 289
 290     @property
 291     def state_ref(self):
 292         return _stack_state_ref(self.name)
 293
 294     def cleanup(self):
 295         assert not self.protected, 'attempt to delete protected stack'
 296         for pn in self.patchorder.all:
 297             self.patches.delete(pn)
 298         self.repository.refs.delete(self.state_ref)
 299         config.remove_section('branch.%s.stgit' % self.name)
 300
 301     def clear_log(self, msg='clear log'):
 302         stack_state = log.StackState.from_stack(prev=None, stack=self)
 303         state_commit = stack_state.commit_state(self.repository, msg)
 304         self.repository.refs.set(self.state_ref, state_commit, msg=msg)
 305
 306     def rename(self, new_name):
 307         old_name = self.name
 308         patch_names = self.patchorder.all
 309         super().rename(new_name)
 310         renames = []
 311         for pn in patch_names:
 312             renames.append((_patch_ref(old_name, pn), _patch_ref(new_name, pn)))
 313         renames.append((_stack_state_ref(old_name), _stack_state_ref(new_name)))
 314
 315         self.repository.refs.rename('rename %s to %s' % (old_name, new_name), *renames)
 316
 317         config.rename_section(
 318             'branch.%s.stgit' % old_name,
 319             'branch.%s.stgit' % new_name,
 320         )
 321
 322     def rename_patch(self, old_name, new_name, msg='rename'):
 323         if new_name == old_name:
 324             raise StackException('New patch name same as old: "%s"' % new_name)
 325         elif new_name in self.patches:
 326             raise StackException('Patch already exists: "%s"' % new_name)
 327         elif not self.patches.is_name_valid(new_name):
 328             raise StackException('Invalid patch name: "%s"' % new_name)
 329         elif old_name not in self.patches:
 330             raise StackException('Unknown patch name: "%s"' % old_name)
 331         self.patchorder.rename_patch(old_name, new_name)
 332         self.patches.rename(old_name, new_name, msg)
 333
 334     def clone(self, clone_name, msg):
 335         clone = self.create(
 336             self.repository,
 337             name=clone_name,
 338             msg=msg,
 339             create_at=self.base,
 340             parent_remote=self.parent_remote,
 341             parent_branch=self.name,
 342         )
 343
 344         for pn in self.patchorder.all_visible:
 345             clone.patches.new(pn, self.patches[pn], 'clone from %s' % self.name)
 346
 347         clone.patchorder.set_order(
 348             applied=[],
 349             unapplied=self.patchorder.all_visible,
 350             hidden=[],
 351         )
 352
 353         prefix = 'branch.%s.' % self.name
 354         clone_prefix = 'branch.%s.' % clone_name
 355         for k, v in list(config.getstartswith(prefix)):
 356             clone_key = k.replace(prefix, clone_prefix, 1)
 357             config.set(clone_key, v)
 358
 359         self.repository.refs.set(
 360             clone.state_ref,
 361             self.repository.refs.get(self.state_ref),
 362             msg=msg,
 363         )
 364
 365         return clone
 366
 367     @classmethod
 368     def initialise(cls, repository, name=None, msg='initialise', switch_to=False):
 369         """Initialise a Git branch to handle patch stack.
 370
 371         :param repository: :class:`Repository` where the :class:`Stack` will be created
 372         :param name: the name of the :class:`Stack`
 373
 374         """
 375         if not name:
 376             name = repository.current_branch_name
 377         # make sure that the corresponding Git branch exists
 378         branch = Branch(repository, name)
 379
 380         stack_state_ref = _stack_state_ref(name)
 381         if repository.refs.exists(stack_state_ref):
 382             raise StackException('%s: stack already initialized' % name)
 383
 384         if switch_to:
 385             branch.switch_to()
 386
 387         stack_state = log.StackState.new_empty(branch.head)
 388         state_commit = stack_state.commit_state(repository, msg)
 389         repository.refs.set(stack_state_ref, state_commit, msg)
 390
 391         return repository.get_stack(name)
 392
 393     @classmethod
 394     def create(
 395         cls,
 396         repository,
 397         name,
 398         msg,
 399         create_at=None,
 400         parent_remote=None,
 401         parent_branch=None,
 402         switch_to=False,
 403     ):
 404         """Create and initialise a Git branch returning the :class:`Stack` object.
 405
 406         :param repository: :class:`Repository` where the :class:`Stack` will be created
 407         :param name: name of the :class:`Stack`
 408         :param msg: message to use in newly created log
 409         :param create_at: Git id used as the base for the newly created Git branch
 410         :param parent_remote: name of the parent remote Git branch
 411         :param parent_branch: name of the parent Git branch
 412
 413         """
 414         branch = Branch.create(repository, name, create_at=create_at)
 415         try:
 416             stack = cls.initialise(repository, name, msg, switch_to=switch_to)
 417         except (BranchException, StackException):
 418             branch.delete()
 419             raise
 420         stack.set_parents(parent_remote, parent_branch)
 421         return stack
 422
 423
 424 class StackRepository(Repository):
 425     """A Git :class:`Repository` with some added StGit-specific operations."""
 426
 427     def __init__(self, directory):
 428         super().__init__(directory)
 429         self._stacks = {}  # name -> Stack
 430
 431     @property
 432     def current_stack(self):
 433         return self.get_stack()
 434
 435     def get_stack(self, name=None):
 436         if not name:
 437             name = self.current_branch_name
 438         if name not in self._stacks:
 439             self._stacks[name] = Stack(self, name)
 440         return self._stacks[name]