Mercurial > treepkg > treepkg
view treepkg/status.py @ 99:7888fe374e11
Add support for notification mails in case of build errors
This involves a new status field notification_mail to keep track of
whether a notification has been sent for a particular build attempt and
two programs to list the pending notifications and to send the pending
notifications (similar to how the static web pages are published) as
well as the corresponding configuration files.
author | Bernhard Herzog <bh@intevation.de> |
---|---|
date | Tue, 19 Feb 2008 19:19:23 +0000 |
parents | 39b2deea8481 |
children | b3f9cc956acc |
line wrap: on
line source
# Copyright (C) 2007, 2008 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"]) def __init__(self, filename): assert os.path.isabs(filename) self._filename = filename self.read() def _init_values(self): self._values = {} def read(self): self._init_values() if not os.path.exists(self._filename): return f = open(self._filename) try: magic = f.next() if magic != self._magic: raise ValueError("File %r has wrong magic" % self._filename) for line in f: field, value = line.split(":", 1) self._values[field] = self._fields[field].deserialize(value) finally: f.close() def write(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]))) util.writefile(self._filename, "".join(lines), 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() 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) 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")