+++ /dev/null
-"""RCS interface module.\r
-\r
-Defines the class RCS, which represents a directory with rcs version\r
-files and (possibly) corresponding work files.\r
-\r
-"""\r
-\r
-\r
-import fnmatch\r
-import os\r
-import re\r
-import string\r
-import tempfile\r
-\r
-\r
-class RCS:\r
-\r
- """RCS interface class (local filesystem version).\r
-\r
- An instance of this class represents a directory with rcs version\r
- files and (possible) corresponding work files.\r
-\r
- Methods provide access to most rcs operations such as\r
- checkin/checkout, access to the rcs metadata (revisions, logs,\r
- branches etc.) as well as some filesystem operations such as\r
- listing all rcs version files.\r
-\r
- XXX BUGS / PROBLEMS\r
-\r
- - The instance always represents the current directory so it's not\r
- very useful to have more than one instance around simultaneously\r
-\r
- """\r
-\r
- # Characters allowed in work file names\r
- okchars = string.ascii_letters + string.digits + '-_=+'\r
-\r
- def __init__(self):\r
- """Constructor."""\r
- pass\r
-\r
- def __del__(self):\r
- """Destructor."""\r
- pass\r
-\r
- # --- Informational methods about a single file/revision ---\r
-\r
- def log(self, name_rev, otherflags = ''):\r
- """Return the full log text for NAME_REV as a string.\r
-\r
- Optional OTHERFLAGS are passed to rlog.\r
-\r
- """\r
- f = self._open(name_rev, 'rlog ' + otherflags)\r
- data = f.read()\r
- status = self._closepipe(f)\r
- if status:\r
- data = data + "%s: %s" % status\r
- elif data[-1] == '\n':\r
- data = data[:-1]\r
- return data\r
-\r
- def head(self, name_rev):\r
- """Return the head revision for NAME_REV"""\r
- dict = self.info(name_rev)\r
- return dict['head']\r
-\r
- def info(self, name_rev):\r
- """Return a dictionary of info (from rlog -h) for NAME_REV\r
-\r
- The dictionary's keys are the keywords that rlog prints\r
- (e.g. 'head' and its values are the corresponding data\r
- (e.g. '1.3').\r
-\r
- XXX symbolic names and locks are not returned\r
-\r
- """\r
- f = self._open(name_rev, 'rlog -h')\r
- dict = {}\r
- while 1:\r
- line = f.readline()\r
- if not line: break\r
- if line[0] == '\t':\r
- # XXX could be a lock or symbolic name\r
- # Anything else?\r
- continue\r
- i = string.find(line, ':')\r
- if i > 0:\r
- key, value = line[:i], string.strip(line[i+1:])\r
- dict[key] = value\r
- status = self._closepipe(f)\r
- if status:\r
- raise IOError, status\r
- return dict\r
-\r
- # --- Methods that change files ---\r
-\r
- def lock(self, name_rev):\r
- """Set an rcs lock on NAME_REV."""\r
- name, rev = self.checkfile(name_rev)\r
- cmd = "rcs -l%s %s" % (rev, name)\r
- return self._system(cmd)\r
-\r
- def unlock(self, name_rev):\r
- """Clear an rcs lock on NAME_REV."""\r
- name, rev = self.checkfile(name_rev)\r
- cmd = "rcs -u%s %s" % (rev, name)\r
- return self._system(cmd)\r
-\r
- def checkout(self, name_rev, withlock=0, otherflags=""):\r
- """Check out NAME_REV to its work file.\r
-\r
- If optional WITHLOCK is set, check out locked, else unlocked.\r
-\r
- The optional OTHERFLAGS is passed to co without\r
- interpretation.\r
-\r
- Any output from co goes to directly to stdout.\r
-\r
- """\r
- name, rev = self.checkfile(name_rev)\r
- if withlock: lockflag = "-l"\r
- else: lockflag = "-u"\r
- cmd = 'co %s%s %s %s' % (lockflag, rev, otherflags, name)\r
- return self._system(cmd)\r
-\r
- def checkin(self, name_rev, message=None, otherflags=""):\r
- """Check in NAME_REV from its work file.\r
-\r
- The optional MESSAGE argument becomes the checkin message\r
- (default "<none>" if None); or the file description if this is\r
- a new file.\r
-\r
- The optional OTHERFLAGS argument is passed to ci without\r
- interpretation.\r
-\r
- Any output from ci goes to directly to stdout.\r
-\r
- """\r
- name, rev = self._unmangle(name_rev)\r
- new = not self.isvalid(name)\r
- if not message: message = "<none>"\r
- if message and message[-1] != '\n':\r
- message = message + '\n'\r
- lockflag = "-u"\r
- if new:\r
- f = tempfile.NamedTemporaryFile()\r
- f.write(message)\r
- f.flush()\r
- cmd = 'ci %s%s -t%s %s %s' % \\r
- (lockflag, rev, f.name, otherflags, name)\r
- else:\r
- message = re.sub(r'([\"$`])', r'\\\1', message)\r
- cmd = 'ci %s%s -m"%s" %s %s' % \\r
- (lockflag, rev, message, otherflags, name)\r
- return self._system(cmd)\r
-\r
- # --- Exported support methods ---\r
-\r
- def listfiles(self, pat = None):\r
- """Return a list of all version files matching optional PATTERN."""\r
- files = os.listdir(os.curdir)\r
- files = filter(self._isrcs, files)\r
- if os.path.isdir('RCS'):\r
- files2 = os.listdir('RCS')\r
- files2 = filter(self._isrcs, files2)\r
- files = files + files2\r
- files = map(self.realname, files)\r
- return self._filter(files, pat)\r
-\r
- def isvalid(self, name):\r
- """Test whether NAME has a version file associated."""\r
- namev = self.rcsname(name)\r
- return (os.path.isfile(namev) or\r
- os.path.isfile(os.path.join('RCS', namev)))\r
-\r
- def rcsname(self, name):\r
- """Return the pathname of the version file for NAME.\r
-\r
- The argument can be a work file name or a version file name.\r
- If the version file does not exist, the name of the version\r
- file that would be created by "ci" is returned.\r
-\r
- """\r
- if self._isrcs(name): namev = name\r
- else: namev = name + ',v'\r
- if os.path.isfile(namev): return namev\r
- namev = os.path.join('RCS', os.path.basename(namev))\r
- if os.path.isfile(namev): return namev\r
- if os.path.isdir('RCS'):\r
- return os.path.join('RCS', namev)\r
- else:\r
- return namev\r
-\r
- def realname(self, namev):\r
- """Return the pathname of the work file for NAME.\r
-\r
- The argument can be a work file name or a version file name.\r
- If the work file does not exist, the name of the work file\r
- that would be created by "co" is returned.\r
-\r
- """\r
- if self._isrcs(namev): name = namev[:-2]\r
- else: name = namev\r
- if os.path.isfile(name): return name\r
- name = os.path.basename(name)\r
- return name\r
-\r
- def islocked(self, name_rev):\r
- """Test whether FILE (which must have a version file) is locked.\r
-\r
- XXX This does not tell you which revision number is locked and\r
- ignores any revision you may pass in (by virtue of using rlog\r
- -L -R).\r
-\r
- """\r
- f = self._open(name_rev, 'rlog -L -R')\r
- line = f.readline()\r
- status = self._closepipe(f)\r
- if status:\r
- raise IOError, status\r
- if not line: return None\r
- if line[-1] == '\n':\r
- line = line[:-1]\r
- return self.realname(name_rev) == self.realname(line)\r
-\r
- def checkfile(self, name_rev):\r
- """Normalize NAME_REV into a (NAME, REV) tuple.\r
-\r
- Raise an exception if there is no corresponding version file.\r
-\r
- """\r
- name, rev = self._unmangle(name_rev)\r
- if not self.isvalid(name):\r
- raise os.error, 'not an rcs file %r' % (name,)\r
- return name, rev\r
-\r
- # --- Internal methods ---\r
-\r
- def _open(self, name_rev, cmd = 'co -p', rflag = '-r'):\r
- """INTERNAL: open a read pipe to NAME_REV using optional COMMAND.\r
-\r
- Optional FLAG is used to indicate the revision (default -r).\r
-\r
- Default COMMAND is "co -p".\r
-\r
- Return a file object connected by a pipe to the command's\r
- output.\r
-\r
- """\r
- name, rev = self.checkfile(name_rev)\r
- namev = self.rcsname(name)\r
- if rev:\r
- cmd = cmd + ' ' + rflag + rev\r
- return os.popen("%s %r" % (cmd, namev))\r
-\r
- def _unmangle(self, name_rev):\r
- """INTERNAL: Normalize NAME_REV argument to (NAME, REV) tuple.\r
-\r
- Raise an exception if NAME contains invalid characters.\r
-\r
- A NAME_REV argument is either NAME string (implying REV='') or\r
- a tuple of the form (NAME, REV).\r
-\r
- """\r
- if type(name_rev) == type(''):\r
- name_rev = name, rev = name_rev, ''\r
- else:\r
- name, rev = name_rev\r
- for c in rev:\r
- if c not in self.okchars:\r
- raise ValueError, "bad char in rev"\r
- return name_rev\r
-\r
- def _closepipe(self, f):\r
- """INTERNAL: Close PIPE and print its exit status if nonzero."""\r
- sts = f.close()\r
- if not sts: return None\r
- detail, reason = divmod(sts, 256)\r
- if reason == 0: return 'exit', detail # Exit status\r
- signal = reason&0x7F\r
- if signal == 0x7F:\r
- code = 'stopped'\r
- signal = detail\r
- else:\r
- code = 'killed'\r
- if reason&0x80:\r
- code = code + '(coredump)'\r
- return code, signal\r
-\r
- def _system(self, cmd):\r
- """INTERNAL: run COMMAND in a subshell.\r
-\r
- Standard input for the command is taken from /dev/null.\r
-\r
- Raise IOError when the exit status is not zero.\r
-\r
- Return whatever the calling method should return; normally\r
- None.\r
-\r
- A derived class may override this method and redefine it to\r
- capture stdout/stderr of the command and return it.\r
-\r
- """\r
- cmd = cmd + " </dev/null"\r
- sts = os.system(cmd)\r
- if sts: raise IOError, "command exit status %d" % sts\r
-\r
- def _filter(self, files, pat = None):\r
- """INTERNAL: Return a sorted copy of the given list of FILES.\r
-\r
- If a second PATTERN argument is given, only files matching it\r
- are kept. No check for valid filenames is made.\r
-\r
- """\r
- if pat:\r
- def keep(name, pat = pat):\r
- return fnmatch.fnmatch(name, pat)\r
- files = filter(keep, files)\r
- else:\r
- files = files[:]\r
- files.sort()\r
- return files\r
-\r
- def _remove(self, fn):\r
- """INTERNAL: remove FILE without complaints."""\r
- try:\r
- os.unlink(fn)\r
- except os.error:\r
- pass\r
-\r
- def _isrcs(self, name):\r
- """INTERNAL: Test whether NAME ends in ',v'."""\r
- return name[-2:] == ',v'\r
projects / mirror_edk2.git / blobdiff