Mercurial > treepkg > treepkg
view treepkg/status.py @ 557:9824e409388b
Refactor git branching
If a checkout is already available and the branch is changed in
the config git command would always fail because it doesn't know
the branch to track. Therefore always check if the branch is
locally available and if not checkout the remote branch
author | Bjoern Ricks <bricks@intevation.de> |
---|---|
date | Fri, 02 Sep 2011 08:45:28 +0000 |
parents | 5fda6768bef6 |
children |
line wrap: on
line source
# Copyright (C) 2007, 2008, 2009 by Intevation GmbH # Authors: # Bernhard Herzog <bh@intevation.de> # # This program is free software under the GPL (>=v2) # Read the file COPYING coming with the software for details. import os import datetime import time import util # special object to indicate no default value nodefault = object() class FieldDesc(object): def __init__(self, default=nodefault): self.default = default def has_default(self): return self.default is not nodefault def set_default(self, value): self.default = value def serialize(self, value): raise NotImplementedError def deserialize(self, string): raise NotImplementedError class StringFieldDesc(FieldDesc): def serialize(self, value): return str(value) def deserialize(self, value): return value.strip() class DateFieldDesc(FieldDesc): date_format = "%Y-%m-%d %H:%M:%S" def serialize(self, value): return value.strftime(self.date_format) def deserialize(self, string): return datetime.datetime(*time.strptime(string.strip(), self.date_format)[:6]) class EnumValue(object): def __init__(self, name, description, finished=False, error=False): self.name = name self.description = description self.finished = finished self.error = error class EnumFieldDesc(FieldDesc): def __init__(self, *args, **kw): super(EnumFieldDesc, self).__init__(*args, **kw) self.values = {} def add(self, name, description, default=False, **kw): enum = EnumValue(name, description, **kw) self.values[enum.name] = enum if default: self.set_default(enum) def __iter__(self): return self.values.itervalues() def serialize(self, value): assert value.name is not None return value.name def deserialize(self, string): return self.values[string.strip()] def make_setter(fieldname, enum): def setter(self): setattr(self, fieldname, enum) setter.__name__ = enum.name return setter class StatusMetaClass(type): def __new__(cls, name, bases, clsdict): # Generate the _fields class variable from the field descriptors # in clsdict and remove the descriptors themselves. Also, add # one setter method for each enum. fields = dict() for key, value in clsdict.items(): if isinstance(value, FieldDesc): fields[key] = value del clsdict[key] if isinstance(value, EnumFieldDesc): for enum in value: clsdict[enum.name] = make_setter(key, enum) clsdict["_fields"] = fields return type.__new__(cls, name, bases, clsdict) class Status(object): __metaclass__ = StatusMetaClass # Overwrite in derived classes with a different magic string _magic = "Status 0.0\n" # Derived classes may extend a copy of this set with more instance # variables. _attrs = set(["_filename", "_values", "_after_setattr"]) def __init__(self, filename, after_setattr=None): self._values = dict() self._filename = filename self._after_setattr = after_setattr if self._filename is not None: assert os.path.isabs(self._filename) self.read() def deserialize(self, iter_lines, filename=None): magic = iter_lines.next() if magic != self._magic: message = ("Wrong magic: found %r but expected %r" % (magic, self._magic)) if filename is not None: message += " (read from %r)" % filename raise ValueError(message) values = dict() for line in iter_lines: field, value = line.split(":", 1) values[field] = self._fields[field].deserialize(value) self._values = values def serialize(self): lines = [self._magic] for field, desc in self._fields.items(): if field in self._values: lines.append("%s: %s\n" % (field, desc.serialize(self._values[field]))) return lines def read(self): if not os.path.exists(self._filename): return f = open(self._filename) try: self.deserialize(f, self._filename) finally: f.close() def write(self): if self._filename is not None: util.writefile(self._filename, "".join(self.serialize()), 0644) def __getattr__(self, attr): desc = self._fields.get(attr) if desc is not None: if attr in self._values: return self._values[attr] elif desc.has_default(): return desc.default raise AttributeError(attr) def __setattr__(self, attr, value): if attr in self._fields: self._values[attr] = value self.write() if self._after_setattr is not None: self._after_setattr(self, attr) elif attr in self._attrs: self.__dict__[attr] = value else: raise AttributeError(attr) class RevisionStatus(Status): _magic = "TreePackagerStatus 0.0\n" status = EnumFieldDesc() status.add("creating_source_package", "creating source package") status.add("source_package_created", "source package created") status.add("creating_binary_package", "building binary packages") status.add("binary_package_created", "build successful", finished=True) status.add("error", "error", error=True) status.add("unknown", "unknown", default=True) start = DateFieldDesc(default=None) stop = DateFieldDesc(default=None) tags = StringFieldDesc(default="") notification_mail = EnumFieldDesc() notification_mail.add("notification_sent", "notification mail has been sent", default=True) notification_mail.add("notification_pending", "notification mail still needs to be sent")