# for, say, SCCS.
#
-class RCS:
- "Class encapsulating RCS version-control operations."
- historyfile = ".HISTORY"
+class Generic:
+ "Generic file-oriented VCS - could be used for, say. SCCS as well as RCS."
def __init__(self):
self.roundtrip = False
if verbose:
- if not os.path.exists("RCS"):
+ if not os.path.exists(self.__class__.repodir):
try:
- os.makedirs("RCS")
+ os.makedirs(self.__class__.repodir)
except OSError:
raise FatalException("can't create RCS directory.")
- file(RCS.historyfile, "w").close()
- @staticmethod
- def master(filename):
+ def master(self, filename):
"Return the master file corresponding to a specified filename."
parts = list(os.path.split(filename))
- parts.insert(-1, "RCS")
+ parts.insert(-1, self.__class__.repodir)
return os.path.join(*parts) + ",v"
- @staticmethod
- def rcsfiles():
- "List all masters."
- res = []
- for root, dirs, files in os.walk(os.getcwd(), topdown=False):
- if root.endswith("/RCS"):
- for name in files:
- res.append(os.path.join(root, name))
- return res
- @staticmethod
- def gethdr(path, header):
- "Retrieve a required header from a master file"
- master = RCS.master(path)
- for line in file(master):
- if line.startswith(header):
- value = line.split()[1].strip()
- if value.endswith(";"):
- value = value[:-1]
- return value
- raise FatalException("missing '%s' in %s" % (header, master))
- @staticmethod
- def record(code, path=None):
- "Record an action."
- with file(RCS.historyfile, "a") as fp:
- fp.write(code)
- if path:
- fp.write(" ")
- fp.write(RCS.gethdr(path, "head\t") + " ")
- fp.write(path)
- fp.write("\n")
- @staticmethod
- def instantiate(path, copyfile=None):
- rcsdir = os.path.join(os.path.dirname(path), "RCS")
+ def rename(self, commit, source, target):
+ "Perform RCS renames, carrying history with the rename."
+ self.copy(commit, source, target, legend="Rename")
+ self.delete(commit, source, legend="Rename")
+ def instantiate(self, path, copyfile=None):
+ rcsdir = os.path.join(os.path.dirname(path), self.__class__.repodir)
if not os.path.exists(rcsdir):
try:
os.makedirs(rcsdir)
except OSError, e:
raise FatalException("refusing to step on %s: %s." % (path, `e`))
return rcsdir
+ def masterfiles(self):
+ "List all masters."
+ res = []
+ for root, dirs, files in os.walk(os.getcwd(), topdown=False):
+ if root.endswith(os.sep + self.__class__.repodir):
+ for name in files:
+ res.append(os.path.join(root, name))
+ return res
+ def precommit(self, event):
+ #self.record("commit")
+ pass
+ def delete(self, commit, path, legend="delete"):
+ "Mark a path deleted, but don't actually remove the master."
+ self.instantiate(path)
+ open(path, "w").close()
+ self.checkin(commit, path, legend="delete", state="Deleted")
+ #self.record("D", path)
+
+class RCS(Generic):
+ "Class encapsulating RCS version-control operations."
+ repodir = "RCS"
+ def gethdr(self, path, header):
+ "Retrieve a required header from a master file"
+ master = self.master(path)
+ for line in file(master):
+ if line.startswith(header):
+ value = line.split()[1].strip()
+ if value.endswith(";"):
+ value = value[:-1]
+ return value
+ raise FatalException("missing '%s' in %s" % (header, master))
def checkin(self, commit, path, legend, state=None):
"Check in a new revision - also used to create new masters."
if self.roundtrip:
command += " -s" + state
command += " -d'%s'" % event.committer.date.rfc822()
command += ' -m"%s"' % comment.replace('"', '\\"')
- command += " " + path + " " + RCS.master(path)
+ command += " " + path + " " + self.master(path)
do_or_die(command, legend)
loud_remove(path)
- def precommit(self, event):
- if verbose:
- RCS.record("commit")
def modify(self, event, path, mode, copyfile):
"Perform RCS file checkins and modifications,"
if mode == "160000":
raise FatalException("ignoring a submodule link %s", path)
else:
- RCS.instantiate(path, copyfile)
+ self.instantiate(path, copyfile)
self.checkin(event, path, "modify")
- RCS.record("M", path)
- def delete(self, commit, path, legend="delete"):
- "Mark a path deleted, but don't actually remove the master."
- RCS.instantiate(path)
- open(path, "w").close()
- self.checkin(commit, path, legend="delete", state="Deleted")
- RCS.record("D", path)
+ #self.record("M", path)
def copy(self, commit, source, target, legend="Copy"):
"Perform RCS copies. Start the copy with no history."
- if os.path.exists(RCS.master(target)):
+ if os.path.exists(self.master(target)):
raise FatalException("copy to existing file")
- do_or_die("co %s" % RCS.master(source), legend)
- RCS.instantiate(target)
+ do_or_die("co %s" % self.master(source), legend)
+ self.instantiate(target)
loud_rename(source, target)
self.checkin(commit, target, legend="copy")
- RCS.record("C", source)
- RCS.record("T", target)
- def rename(self, commit, source, target):
- "Perform RCS renames, carrying history with the rename."
- self.copy(commit, source, target, legend="Rename")
- self.delete(commit, source, legend="Rename")
+ #self.record("C", source)
+ #self.record("T", target)
def postcommit(self, event):
# Save annotated tags for round-tripping
for tag in event.taglist:
do_or_die("ci -l -m'Annotated tag %s' -d'%s' ANNOTATED-TAGS,v"
% (tag.name, tag.tagger.date.rfc822()))
os.remove("ANNOTATED-TAGS")
- do_or_die("rcs -n'%s': %s" % (tag.name, " ".join(RCS.rcsfiles())))
+ do_or_die("rcs -n'%s': %s" % (tag.name, " ".join(self.masterfiles())))
def checkout(self, switch):
"Post-conversion actions."
if switch != '-l':
- do_or_die("rcs -u %s" % (" ".join(RCS.rcsfiles()),))
+ do_or_die("rcs -u %s" % (" ".join(self.masterfiles()),))
if switch == '-u':
- do_or_die("co -u %s" % (" ".join(RCS.rcsfiles()),))
- if verbose:
- os.rename(RCS.historyfile, os.path.join("RCS", RCS.historyfile))
+ do_or_die("co -u %s" % (" ".join(self.masterfiles()),))
if __name__ == '__main__':
(options, arguments) = getopt.getopt(sys.argv[1:], "lnpuvV?")