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")
This site is hosted by Intevation GmbH (Datenschutzerklärung und Impressum | Privacy Policy and Imprint)