[Zope3-checkins] CVS: Zope3/src/zope/fssync - bundle.py:1.1 fsbundle.py:1.1

[Fred L. Drake, Jr.]

Update of /cvs-repository/Zope3/src/zope/fssync
In directory cvs.zope.org:/tmp/cvs-serv11471/src/zope/fssync

Added Files:
	bundle.py fsbundle.py 
Log Message:
Initial bundle management tool.  All this can do is create a new bundle in an fssync
checkout with enough information for zsync to handle it (including commit).
But it means we really have bundles.


=== Added File Zope3/src/zope/fssync/bundle.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Zope 3 bundle management utility.

Command line syntax summary:

%(program)s create PATH ...

``%(program)s help'' prints the global help (this message)
``%(program)s help command'' prints the local help for the command
"""
"""
$Id: bundle.py,v 1.1 2003/08/12 22:08:34 fdrake Exp $
"""

import os

from zope.fssync.command import Command, Usage
from zope.fssync.fsbundle import FSBundle


def main():
    """Main program.

    The return value is the suggested sys.exit() status code:
    0 or None for success
    2 for command line syntax errors
    1 or other for later errors
    """
    cmd = Command(usage=__doc__)
    for func, aliases, short, long in command_table:
        cmd.addCommand(func.__name__, func, short, long, aliases)

    return cmd.main()


def create(opts, args):
    """%(program)s create PATH ...

    """
    factory = None
    type = None
    for opt, arg in opts:
        if opt in ("-f", "--factory"):
            if factory:
                raise Usage("-f/--factory can only be given once")
            factory = arg
        elif opt in ("-t", "--type"):
            if type:
                raise Usage("-t/--type can only be given once")
            type = arg
    if len(args) != 1:
        raise Usage("create requires exactly one path")
    fs = FSBundle()
    fs.create(args[0], type, factory)


command_table = [
    # name is taken from the function name
    # function, aliases,  short opts,   long opts
    (create,    "",       "f:t:",       "factory= type="),
    ]


=== Added File Zope3/src/zope/fssync/fsbundle.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
# 
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
# 
##############################################################################
"""High-level class to support bundle management on an fssync checkout.

$Id: fsbundle.py,v 1.1 2003/08/12 22:08:34 fdrake Exp $
"""

import os
import posixpath

from zope.fssync.fssync import FSSync
from zope.fssync.fsutil import Error
from zope.fssync.metadata import Metadata


class FSBundle(object):

    def __init__(self):
        self.metadata = Metadata()
        self.sync = FSSync(metadata=self.metadata)

    # bundle operations

    def create(self, path, type, factory):
        if os.path.exists(path):
            raise Error("%r already exists", path)
        dir, name = os.path.split(path)
        self.check_name(name)
        self.check_parent_directory(dir)
        if factory is None and type is None:
            factory = type = "zope.app.services.bundle.Bundle"
        self.sync.mkdir(path)
        entry = self.metadata.getentry(path)
        assert entry.get("flag") == "added"
        if factory:
            entry["factory"] = factory
        if type:
            entry["type"] = type
        self.metadata.flush()

    # helper methods

    def check_parent_directory(self, dir):
        if dir:
            if not os.path.exists(dir):
                raise Error("%r does not exist", dir)
            if not os.path.isdir(dir):
                raise Error("%r is not a directory", dir)
        else:
            dir = os.curdir
        # XXX this might not be the right check
        entry = self.metadata.getentry(dir)
        if not entry:
            raise Error("nothing known about", dir)

    def check_name(self, name):
        if name.count("-") != 1:
            raise Error("%r is not a legal bundle name", name)
        basename, version = name.split("-")
        self.check_version(version)

    def check_version(self, s):
        self.parse_version(s)

    def parse_version(self, s):
        parts = s.split(".")
        if len(parts) not in (3, 4):
            raise Error("%r is not a valid bundle version", s)
        try:
            n0 = int(parts[0])
            n1 = int(parts[1])
            n2 = int(parts[2])
        except ValueError:
            raise Error("%r is not a valid bundle version", s)
        try:
            p3 = int(parts[3])
        except IndexError:
            p3 = None
        except ValueError:
            p3 = parts[3]
        return (n0, n1, n2, p3)