blob: a0047e7107a720cc70ff6eae93a91f9835dc56bd [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
This module loads protocol metadata into python objects. It provides
access to spec metadata via a python object model, and can also
dynamically creating python methods, classes, and modules based on the
spec metadata. All the generated methods have proper signatures and
doc strings based on the spec metadata so the python help system can
be used to browse the spec documentation. The generated methods all
dispatch to the self.invoke(meth, args) callback of the containing
class so that the generated code can be reused in a variety of
import re, new, mllib, qpid
from util import fill
class SpecContainer:
def __init__(self):
self.items = []
self.byname = {}
self.byid = {}
self.indexes = {}
def add(self, item):
if self.byname.has_key(
raise ValueError("duplicate name: %s" % item)
if == None: = len(self)
elif self.byid.has_key(
raise ValueError("duplicate id: %s" % item)
self.indexes[item] = len(self.items)
self.byname[] = item
self.byid[] = item
def index(self, item):
return self.indexes[item]
except KeyError:
raise ValueError(item)
def __iter__(self):
return iter(self.items)
def __len__(self):
return len(self.items)
class Metadata:
PRINT = []
def __init__(self):
def __str__(self):
args = map(lambda f: "%s=%s" % (f, getattr(self, f)), self.PRINT)
return "%s(%s)" % (self.__class__.__name__, ", ".join(args))
def __repr__(self):
return str(self)
class Spec(Metadata):
PRINT=["major", "minor", "file"]
def __init__(self, major, minor, file):
self.major = major
self.minor = minor
self.file = file
self.constants = SpecContainer() = SpecContainer()
self.classes = SpecContainer()
# methods indexed by classname_methname
self.methods = {}
# structs by type code
self.structs = {}
def post_load(self):
self.module = self.define_module("amqp%s%s" % (self.major, self.minor))
self.klass = self.define_class("Amqp%s%s" % (self.major, self.minor))
def method(self, name):
if not self.methods.has_key(name):
for cls in self.classes:
clen = len(
if name.startswith( and name[clen] == "_":
end = name[clen + 1:]
if cls.methods.byname.has_key(end):
self.methods[name] = cls.methods.byname[end]
return self.methods.get(name)
def parse_method(self, name):
parts = re.split(r"\s*\.\s*", name)
if len(parts) != 2:
raise ValueError(name)
klass, meth = parts
return self.classes.byname[klass].methods.byname[meth]
def struct(self, name, *args, **kwargs):
type =[name].type
return qpid.Struct(type, *args, **kwargs)
def define_module(self, name, doc = None):
module = new.module(name, doc)
module.__file__ = self.file
for c in self.classes:
cls = c.define_class(
cls.__module__ = module.__name__
setattr(module,, cls)
return module
def define_class(self, name):
methods = {}
for c in self.classes:
for m in c.methods:
meth = + "_" +
methods[meth] = m.define_method(meth)
return type(name, (), methods)
class Constant(Metadata):
PRINT=["name", "id"]
def __init__(self, spec, name, id, klass, docs):
self.spec = spec = name = id
self.klass = klass = docs
class Domain(Metadata):
PRINT=["name", "type"]
def __init__(self, spec, name, type, description, docs):
self.spec = spec = None = name
self.type = type
self.description = description = docs
class Struct(Metadata):
PRINT=["size", "type", "pack"]
def __init__(self, size, type, pack):
self.size = size
self.type = type
self.pack = pack
self.fields = SpecContainer()
class Class(Metadata):
PRINT=["name", "id"]
def __init__(self, spec, name, id, handler, docs):
self.spec = spec = name = id
self.handler = handler
self.fields = SpecContainer()
self.methods = SpecContainer() = docs
def define_class(self, name):
methods = {}
for m in self.methods:
methods[] = m.define_method(
return type(name, (), methods)
class Method(Metadata):
PRINT=["name", "id"]
def __init__(self, klass, name, id, content, responses, result, synchronous,
description, docs):
self.klass = klass = name = id
self.content = content
self.responses = responses
self.result = result
self.synchronous = synchronous
self.fields = SpecContainer()
self.description = description = docs
self.response = False
def is_l4_command(self):
return not in ["execution", "channel", "connection", "session"]
def arguments(self, *args, **kwargs):
nargs = len(args) + len(kwargs)
maxargs = len(self.fields)
if nargs > maxargs:
self._type_error("takes at most %s arguments (%s) given", maxargs, nargs)
result = []
for f in self.fields:
idx = self.fields.index(f)
if idx < len(args):
elif kwargs.has_key(
for key, value in kwargs.items():
if self.fields.byname.has_key(key):
self._type_error("got multiple values for keyword argument '%s'", key)
self._type_error("got an unexpected keyword argument '%s'", key)
return tuple(result)
def _type_error(self, msg, *args):
raise TypeError("%s %s" % (, msg % args))
def docstring(self):
s = "\n\n".join([fill(d, 2) for d in [self.description] +])
for f in self.fields:
s += "\n\n" + "\n\n".join([fill([0], 4,] +
[fill(d, 4) for d in[1:]])
if self.responses:
s += "\n\nValid responses: "
for r in self.responses:
s += + " "
return s
METHOD = "__method__"
DEFAULTS = {"bit": False,
"shortstr": "",
"longstr": "",
"table": {},
"array": [],
"octet": 0,
"short": 0,
"long": 0,
"longlong": 0,
"timestamp": 0,
"content": None,
"uuid": "",
"rfc1982_long": 0,
"rfc1982_long_set": [],
"long_struct": None}
def define_method(self, name):
g = {Method.METHOD: self}
l = {}
args = [(, Method.DEFAULTS[f.type]) for f in self.fields]
methargs = args[:]
if self.content:
args += [("content", None)]
code = "def %s(self, %s):\n" % \
(name, ", ".join(["%s = %r" % a for a in args]))
code += " %r\n" % self.docstring()
argnames = ", ".join([a[0] for a in methargs])
code += " return self.invoke(%s" % Method.METHOD
if argnames:
code += ", (%s,)" % argnames
code += ", ()"
if self.content:
code += ", content"
code += ")"
exec code in g, l
return l[name]
class Field(Metadata):
PRINT=["name", "id", "type"]
def __init__(self, name, id, type, domain, description, docs):
Metadata.__init__(self) = name = id
self.type = type
self.domain = domain
self.description = description = docs
def default(self):
if isinstance(self.type, Struct):
return None
return Method.DEFAULTS[self.type]
"octet": 1,
"short": 2,
"long": 4
def width(st, default=None):
if st in (None, "none", ""):
return default
return WIDTHS[st]
def get_result(nd, spec):
result = nd["result"]
if not result: return None
name = result["@domain"]
if name != None: return[name]
st_nd = result["struct"]
st = Struct(width(st_nd["@size"]), int(result.parent.parent["@index"])*256 +
int(st_nd["@type"]), width(st_nd["@pack"], 2))
spec.structs[st.type] = st
load_fields(st_nd, st.fields,
return st
def get_desc(nd):
label = nd["@label"]
if not label:
label = nd.text()
if label:
label = label.strip()
return label
def get_docs(nd):
return [n.text() for n in nd.query["doc"]]
def load_fields(nd, l, domains):
for f_nd in nd.query["field"]:
type = f_nd["@domain"]
if type == None:
type = f_nd["@type"]
type = pythonize(type)
domain = None
while domains.has_key(type) and domains[type].type != type:
domain = domains[type]
type = domain.type
l.add(Field(pythonize(f_nd["@name"]), f_nd.index(), type, domain,
get_desc(f_nd), get_docs(f_nd)))
def load(specfile, *errata):
doc = mllib.xml_parse(specfile)
spec_root = doc["amqp"]
spec = Spec(int(spec_root["@major"]), int(spec_root["@minor"]), specfile)
for root in [spec_root] + map(lambda x: mllib.xml_parse(x)["amqp"], errata):
# constants
for nd in root.query["constant"]:
val = nd["@value"]
if val.startswith("0x"): val = int(val, 16)
else: val = int(val)
const = Constant(spec, pythonize(nd["@name"]), val, nd["@class"],
except ValueError, e:
#print "Warning:", e
# domains are typedefs
structs = []
for nd in root.query["domain"]:
type = nd["@type"]
if type == None:
st_nd = nd["struct"]
code = st_nd["@type"]
if code not in (None, "", "none"):
code = int(code)
type = Struct(width(st_nd["@size"]), code, width(st_nd["@pack"], 2))
if type.type != None:
spec.structs[type.type] = type
structs.append((type, st_nd))
type = pythonize(type)
domain = Domain(spec, pythonize(nd["@name"]), type, get_desc(nd),
# structs
for st, st_nd in structs:
load_fields(st_nd, st.fields,
# classes
for c_nd in root.query["class"]:
cname = pythonize(c_nd["@name"])
if spec.classes.byname.has_key(cname):
klass = spec.classes.byname[cname]
klass = Class(spec, cname, int(c_nd["@index"]), c_nd["@handler"],
added_methods = []
load_fields(c_nd, klass.fields,
for m_nd in c_nd.query["method"]:
mname = pythonize(m_nd["@name"])
if klass.methods.byname.has_key(mname):
meth = klass.methods.byname[mname]
meth = Method(klass, mname,
m_nd["@content"] == "1",
[pythonize(nd["@name"]) for nd in m_nd.query["response"]],
get_result(m_nd, spec),
m_nd["@synchronous"] == "1",
load_fields(m_nd, meth.fields,
# resolve the responses
for m in added_methods:
m.responses = [klass.methods.byname[r] for r in m.responses]
for resp in m.responses:
resp.response = True
return spec
REPLACE = {" ": "_", "-": "_"}
KEYWORDS = {"global": "global_",
"return": "return_"}
def pythonize(name):
name = str(name)
for key, val in REPLACE.items():
name = name.replace(key, val)
name = KEYWORDS[name]
except KeyError:
return name
class Rule(Metadata):
PRINT = ["text", "implement", "tests"]
def __init__(self, text, implement, tests, path):
self.text = text
self.implement = implement
self.tests = tests
self.path = path
def find_rules(node, rules):
if == "rule":
rules.append(Rule(node.text, node.get("@implement"),
[ch.text for ch in node if == "test"],
if == "doc" and node.get("@name") == "rule":
tests = []
if node.has("@test"):
rules.append(Rule(node.text, None, tests, node.path()))
for child in node:
find_rules(child, rules)
def load_rules(specfile):
rules = []
find_rules(xmlutil.parse(specfile), rules)
return rules
def test_summary():
template = """
<html><head><title>AMQP Tests</title></head>
<table width="80%%" align="center">
rows = []
for rule in load_rules(""):
if rule.tests:
tests = ", ".join(rule.tests)
tests = "&nbsp;"
rows.append('<tr bgcolor="#EEEEEE"><td><b>Path:</b> %s</td>'
'<td><b>Implement:</b> %s</td>'
'<td><b>Tests:</b> %s</td></tr>' %
(rule.path[len("/root/amqp"):], rule.implement, tests))
rows.append('<tr><td colspan="3">%s</td></tr>' % rule.text)
rows.append('<tr><td colspan="3">&nbsp;</td></tr>')
print template % "\n".join(rows)