Commit 04f14e86 authored by Tim Peters's avatar Tim Peters

The blob work requires zope.proxy.

parent 728f5faa
zope.interface
zope.testing
# Packaging information for zpkg.
header proxy.h
<extension _zope_proxy_proxy>
source _zope_proxy_proxy.c
depends-on proxy.h
</extension>
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
##############################################################################
"""More convenience functions for dealing with proxies.
$Id$
"""
from zope.interface import moduleProvides
from zope.proxy.interfaces import IProxyIntrospection
from types import ClassType
from zope.proxy._zope_proxy_proxy import *
from zope.proxy._zope_proxy_proxy import _CAPI
moduleProvides(IProxyIntrospection)
__all__ = tuple(IProxyIntrospection)
def ProxyIterator(p):
yield p
while isProxy(p):
p = getProxiedObject(p)
yield p
/*############################################################################
#
# Copyright (c) 2004 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
############################################################################*/
/*
* This file is also used as a really extensive macro in
* ../app/container/_zope_app_container_contained.c. If you need to
* change this file, you need to "svn copy" it to ../app/container/.
*
* This approach is taken to allow the sources for the two packages
* to be compilable when the relative locations of these aren't
* related in the same way as they are in a checkout.
*
* This will be revisited in the future, but works for now.
*/
#include "Python.h"
#include "modsupport.h"
#define PROXY_MODULE
#include "proxy.h"
static PyTypeObject ProxyType;
#define Proxy_Check(wrapper) (PyObject_TypeCheck((wrapper), &ProxyType))
static PyObject *
empty_tuple = NULL;
/*
* Slot methods.
*/
static PyObject *
wrap_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
{
PyObject *result = NULL;
PyObject *object;
if (PyArg_UnpackTuple(args, "__new__", 1, 1, &object)) {
if (kwds != NULL && PyDict_Size(kwds) != 0) {
PyErr_SetString(PyExc_TypeError,
"proxy.__new__ does not accept keyword args");
return NULL;
}
result = PyType_GenericNew(type, args, kwds);
if (result != NULL) {
ProxyObject *wrapper = (ProxyObject *) result;
Py_INCREF(object);
wrapper->proxy_object = object;
}
}
return result;
}
static int
wrap_init(PyObject *self, PyObject *args, PyObject *kwds)
{
int result = -1;
PyObject *object;
if (PyArg_UnpackTuple(args, "__init__", 1, 1, &object)) {
ProxyObject *wrapper = (ProxyObject *)self;
if (kwds != NULL && PyDict_Size(kwds) != 0) {
PyErr_SetString(PyExc_TypeError,
"proxy.__init__ does not accept keyword args");
return -1;
}
/* If the object in this proxy is not the one we
* received in args, replace it with the new one.
*/
if (wrapper->proxy_object != object) {
PyObject *temp = wrapper->proxy_object;
Py_INCREF(object);
wrapper->proxy_object = object;
Py_DECREF(temp);
}
result = 0;
}
return result;
}
static int
wrap_traverse(PyObject *self, visitproc visit, void *arg)
{
PyObject *ob = Proxy_GET_OBJECT(self);
if (ob != NULL)
return visit(ob, arg);
else
return 0;
}
static int
wrap_clear(PyObject *self)
{
ProxyObject *proxy = (ProxyObject *)self;
PyObject *temp = proxy->proxy_object;
if (temp != NULL) {
proxy->proxy_object = NULL;
Py_DECREF(temp);
}
return 0;
}
static PyObject *
wrap_richcompare(PyObject* self, PyObject* other, int op)
{
if (Proxy_Check(self)) {
self = Proxy_GET_OBJECT(self);
}
else {
other = Proxy_GET_OBJECT(other);
}
return PyObject_RichCompare(self, other, op);
}
static PyObject *
wrap_iter(PyObject *self)
{
return PyObject_GetIter(Proxy_GET_OBJECT(self));
}
static PyObject *
wrap_iternext(PyObject *self)
{
return PyIter_Next(Proxy_GET_OBJECT(self));
}
static void
wrap_dealloc(PyObject *self)
{
(void) wrap_clear(self);
self->ob_type->tp_free(self);
}
/* A variant of _PyType_Lookup that doesn't look in ProxyType.
*
* If argument search_wrappertype is nonzero, we can look in WrapperType.
*/
PyObject *
WrapperType_Lookup(PyTypeObject *type, PyObject *name)
{
int i, n;
PyObject *mro, *res, *base, *dict;
/* Look in tp_dict of types in MRO */
mro = type->tp_mro;
/* If mro is NULL, the type is either not yet initialized
by PyType_Ready(), or already cleared by type_clear().
Either way the safest thing to do is to return NULL. */
if (mro == NULL)
return NULL;
assert(PyTuple_Check(mro));
n = PyTuple_GET_SIZE(mro)
- 1; /* We don't want to look at the last item, which is object. */
for (i = 0; i < n; i++) {
base = PyTuple_GET_ITEM(mro, i);
if (((PyTypeObject *)base) != &ProxyType) {
if (PyClass_Check(base))
dict = ((PyClassObject *)base)->cl_dict;
else {
assert(PyType_Check(base));
dict = ((PyTypeObject *)base)->tp_dict;
}
assert(dict && PyDict_Check(dict));
res = PyDict_GetItem(dict, name);
if (res != NULL)
return res;
}
}
return NULL;
}
static PyObject *
wrap_getattro(PyObject *self, PyObject *name)
{
PyObject *wrapped;
PyObject *descriptor;
PyObject *res = NULL;
char *name_as_string;
int maybe_special_name;
#ifdef Py_USING_UNICODE
/* The Unicode to string conversion is done here because the
existing tp_getattro slots expect a string object as name
and we wouldn't want to break those. */
if (PyUnicode_Check(name)) {
name = PyUnicode_AsEncodedString(name, NULL, NULL);
if (name == NULL)
return NULL;
}
else
#endif
if (!PyString_Check(name)){
PyErr_SetString(PyExc_TypeError, "attribute name must be string");
return NULL;
}
else
Py_INCREF(name);
name_as_string = PyString_AS_STRING(name);
wrapped = Proxy_GET_OBJECT(self);
if (wrapped == NULL) {
PyErr_Format(PyExc_RuntimeError,
"object is NULL; requested to get attribute '%s'",
name_as_string);
goto finally;
}
maybe_special_name = name_as_string[0] == '_' && name_as_string[1] == '_';
if (!(maybe_special_name && strcmp(name_as_string, "__class__") == 0)) {
descriptor = WrapperType_Lookup(self->ob_type, name);
if (descriptor != NULL) {
if (PyType_HasFeature(descriptor->ob_type, Py_TPFLAGS_HAVE_CLASS)
&& descriptor->ob_type->tp_descr_get != NULL) {
res = descriptor->ob_type->tp_descr_get(
descriptor,
self,
(PyObject *)self->ob_type);
} else {
Py_INCREF(descriptor);
res = descriptor;
}
goto finally;
}
}
res = PyObject_GetAttr(wrapped, name);
finally:
Py_DECREF(name);
return res;
}
static int
wrap_setattro(PyObject *self, PyObject *name, PyObject *value)
{
PyObject *wrapped;
PyObject *descriptor;
int res = -1;
#ifdef Py_USING_UNICODE
/* The Unicode to string conversion is done here because the
existing tp_setattro slots expect a string object as name
and we wouldn't want to break those. */
if (PyUnicode_Check(name)) {
name = PyUnicode_AsEncodedString(name, NULL, NULL);
if (name == NULL)
return -1;
}
else
#endif
if (!PyString_Check(name)){
PyErr_SetString(PyExc_TypeError, "attribute name must be string");
return -1;
}
else
Py_INCREF(name);
descriptor = WrapperType_Lookup(self->ob_type, name);
if (descriptor != NULL) {
if (PyType_HasFeature(descriptor->ob_type, Py_TPFLAGS_HAVE_CLASS) &&
descriptor->ob_type->tp_descr_set != NULL) {
res = descriptor->ob_type->tp_descr_set(descriptor, self, value);
} else {
PyErr_Format(PyExc_TypeError,
"Tried to set attribute '%s' on wrapper, but it is not"
" a data descriptor", PyString_AS_STRING(name));
}
goto finally;
}
wrapped = Proxy_GET_OBJECT(self);
if (wrapped == NULL) {
PyErr_Format(PyExc_RuntimeError,
"object is NULL; requested to set attribute '%s'",
PyString_AS_STRING(name));
goto finally;
}
res = PyObject_SetAttr(wrapped, name, value);
finally:
Py_DECREF(name);
return res;
}
static int
wrap_print(PyObject *wrapper, FILE *fp, int flags)
{
return PyObject_Print(Proxy_GET_OBJECT(wrapper), fp, flags);
}
static PyObject *
wrap_str(PyObject *wrapper) {
return PyObject_Str(Proxy_GET_OBJECT(wrapper));
}
static PyObject *
wrap_repr(PyObject *wrapper)
{
return PyObject_Repr(Proxy_GET_OBJECT(wrapper));
}
static int
wrap_compare(PyObject *wrapper, PyObject *v)
{
return PyObject_Compare(Proxy_GET_OBJECT(wrapper), v);
}
static long
wrap_hash(PyObject *self)
{
return PyObject_Hash(Proxy_GET_OBJECT(self));
}
static PyObject *
wrap_call(PyObject *self, PyObject *args, PyObject *kw)
{
if (kw)
return PyEval_CallObjectWithKeywords(Proxy_GET_OBJECT(self),
args, kw);
else
return PyObject_CallObject(Proxy_GET_OBJECT(self), args);
}
/*
* Number methods
*/
/*
* Number methods.
*/
static PyObject *
call_int(PyObject *self)
{
PyNumberMethods *nb = self->ob_type->tp_as_number;
if (nb == NULL || nb->nb_int == NULL) {
PyErr_SetString(PyExc_TypeError,
"object can't be converted to int");
return NULL;
}
return nb->nb_int(self);
}
static PyObject *
call_long(PyObject *self)
{
PyNumberMethods *nb = self->ob_type->tp_as_number;
if (nb == NULL || nb->nb_long == NULL) {
PyErr_SetString(PyExc_TypeError,
"object can't be converted to long");
return NULL;
}
return nb->nb_long(self);
}
static PyObject *
call_float(PyObject *self)
{
PyNumberMethods *nb = self->ob_type->tp_as_number;
if (nb == NULL || nb->nb_float== NULL) {
PyErr_SetString(PyExc_TypeError,
"object can't be converted to float");
return NULL;
}
return nb->nb_float(self);
}
static PyObject *
call_oct(PyObject *self)
{
PyNumberMethods *nb = self->ob_type->tp_as_number;
if (nb == NULL || nb->nb_oct== NULL) {
PyErr_SetString(PyExc_TypeError,
"object can't be converted to oct");
return NULL;
}
return nb->nb_oct(self);
}
static PyObject *
call_hex(PyObject *self)
{
PyNumberMethods *nb = self->ob_type->tp_as_number;
if (nb == NULL || nb->nb_hex == NULL) {
PyErr_SetString(PyExc_TypeError,
"object can't be converted to hex");
return NULL;
}
return nb->nb_hex(self);
}
static PyObject *
call_ipow(PyObject *self, PyObject *other)
{
/* PyNumber_InPlacePower has three args. How silly. :-) */
return PyNumber_InPlacePower(self, other, Py_None);
}
typedef PyObject *(*function1)(PyObject *);
static PyObject *
check1(ProxyObject *self, char *opname, function1 operation)
{
PyObject *result = NULL;
result = operation(Proxy_GET_OBJECT(self));
#if 0
if (result != NULL)
/* XXX create proxy for result? */
;
#endif
return result;
}
static PyObject *
check2(PyObject *self, PyObject *other,
char *opname, char *ropname, binaryfunc operation)
{
PyObject *result = NULL;
PyObject *object;
if (Proxy_Check(self)) {
object = Proxy_GET_OBJECT(self);
result = operation(object, other);
}
else if (Proxy_Check(other)) {
object = Proxy_GET_OBJECT(other);
result = operation(self, object);
}
else {
Py_INCREF(Py_NotImplemented);
return Py_NotImplemented;
}
#if 0
if (result != NULL)
/* XXX create proxy for result? */
;
#endif
return result;
}
static PyObject *
check2i(ProxyObject *self, PyObject *other,
char *opname, binaryfunc operation)
{
PyObject *result = NULL;
PyObject *object = Proxy_GET_OBJECT(self);
result = operation(object, other);
if (result == object) {
/* If the operation was really carried out inplace,
don't create a new proxy, but use the old one. */
Py_INCREF(self);
Py_DECREF(object);
result = (PyObject *)self;
}
#if 0
else if (result != NULL)
/* XXX create proxy for result? */
;
#endif
return result;
}
#define UNOP(NAME, CALL) \
static PyObject *wrap_##NAME(PyObject *self) \
{ return check1((ProxyObject *)self, "__"#NAME"__", CALL); }
#define BINOP(NAME, CALL) \
static PyObject *wrap_##NAME(PyObject *self, PyObject *other) \
{ return check2(self, other, "__"#NAME"__", "__r"#NAME"__", CALL); }
#define INPLACE(NAME, CALL) \
static PyObject *wrap_i##NAME(PyObject *self, PyObject *other) \
{ return check2i((ProxyObject *)self, other, "__i"#NAME"__", CALL); }
BINOP(add, PyNumber_Add)
BINOP(sub, PyNumber_Subtract)
BINOP(mul, PyNumber_Multiply)
BINOP(div, PyNumber_Divide)
BINOP(mod, PyNumber_Remainder)
BINOP(divmod, PyNumber_Divmod)
static PyObject *
wrap_pow(PyObject *self, PyObject *other, PyObject *modulus)
{
PyObject *result = NULL;
PyObject *object;
if (Proxy_Check(self)) {
object = Proxy_GET_OBJECT(self);
result = PyNumber_Power(object, other, modulus);
}
else if (Proxy_Check(other)) {
object = Proxy_GET_OBJECT(other);
result = PyNumber_Power(self, object, modulus);
}
else if (modulus != NULL && Proxy_Check(modulus)) {
object = Proxy_GET_OBJECT(modulus);
result = PyNumber_Power(self, other, modulus);
}
else {
Py_INCREF(Py_NotImplemented);
return Py_NotImplemented;
}
return result;
}
BINOP(lshift, PyNumber_Lshift)
BINOP(rshift, PyNumber_Rshift)
BINOP(and, PyNumber_And)
BINOP(xor, PyNumber_Xor)
BINOP(or, PyNumber_Or)
static int
wrap_coerce(PyObject **p_self, PyObject **p_other)
{
PyObject *self = *p_self;
PyObject *other = *p_other;
PyObject *object;
PyObject *left;
PyObject *right;
int r;
assert(Proxy_Check(self));
object = Proxy_GET_OBJECT(self);
left = object;
right = other;
r = PyNumber_CoerceEx(&left, &right);
if (r != 0)
return r;
/* Now left and right have been INCREF'ed. Any new value that
comes out is proxied; any unchanged value is left unchanged. */
if (left == object) {
/* Keep the old proxy */
Py_INCREF(self);
Py_DECREF(left);
left = self;
}
#if 0
else {
/* XXX create proxy for left? */
}
if (right != other) {
/* XXX create proxy for right? */
}
#endif
*p_self = left;
*p_other = right;
return 0;
}
UNOP(neg, PyNumber_Negative)
UNOP(pos, PyNumber_Positive)
UNOP(abs, PyNumber_Absolute)
UNOP(invert, PyNumber_Invert)
UNOP(int, call_int)
UNOP(long, call_long)
UNOP(float, call_float)
UNOP(oct, call_oct)
UNOP(hex, call_hex)
INPLACE(add, PyNumber_InPlaceAdd)
INPLACE(sub, PyNumber_InPlaceSubtract)
INPLACE(mul, PyNumber_InPlaceMultiply)
INPLACE(div, PyNumber_InPlaceDivide)
INPLACE(mod, PyNumber_InPlaceRemainder)
INPLACE(pow, call_ipow)
INPLACE(lshift, PyNumber_InPlaceLshift)
INPLACE(rshift, PyNumber_InPlaceRshift)
INPLACE(and, PyNumber_InPlaceAnd)
INPLACE(xor, PyNumber_InPlaceXor)
INPLACE(or, PyNumber_InPlaceOr)
BINOP(floordiv, PyNumber_FloorDivide)
BINOP(truediv, PyNumber_TrueDivide)
INPLACE(floordiv, PyNumber_InPlaceFloorDivide)
INPLACE(truediv, PyNumber_InPlaceTrueDivide)
static int
wrap_nonzero(PyObject *self)
{
return PyObject_IsTrue(Proxy_GET_OBJECT(self));
}
/*
* Sequence methods
*/
static int
wrap_length(PyObject *self)
{
return PyObject_Length(Proxy_GET_OBJECT(self));
}
static PyObject *
wrap_slice(PyObject *self, int start, int end)
{
return PySequence_GetSlice(Proxy_GET_OBJECT(self), start, end);
}
static int
wrap_ass_slice(PyObject *self, int i, int j, PyObject *value)
{
return PySequence_SetSlice(Proxy_GET_OBJECT(self), i, j, value);
}
static int
wrap_contains(PyObject *self, PyObject *value)
{
return PySequence_Contains(Proxy_GET_OBJECT(self), value);
}
/*
* Mapping methods
*/
static PyObject *
wrap_getitem(PyObject *wrapper, PyObject *v) {
return PyObject_GetItem(Proxy_GET_OBJECT(wrapper), v);
}
static int
wrap_setitem(PyObject *self, PyObject *key, PyObject *value)
{
if (value == NULL)
return PyObject_DelItem(Proxy_GET_OBJECT(self), key);
else
return PyObject_SetItem(Proxy_GET_OBJECT(self), key, value);
}
/*
* Normal methods
*/
static char
reduce__doc__[] =
"__reduce__()\n"
"Raise an exception; this prevents proxies from being picklable by\n"
"default, even if the underlying object is picklable.";
static PyObject *
wrap_reduce(PyObject *self)
{
PyObject *pickle_error = NULL;
PyObject *pickle = PyImport_ImportModule("pickle");
if (pickle == NULL)
PyErr_Clear();
else {
pickle_error = PyObject_GetAttrString(pickle, "PicklingError");
if (pickle_error == NULL)
PyErr_Clear();
}
if (pickle_error == NULL) {
pickle_error = PyExc_RuntimeError;
Py_INCREF(pickle_error);
}
PyErr_SetString(pickle_error,
"proxy instances cannot be pickled");
Py_DECREF(pickle_error);
return NULL;
}
static PyNumberMethods
wrap_as_number = {
wrap_add, /* nb_add */
wrap_sub, /* nb_subtract */
wrap_mul, /* nb_multiply */
wrap_div, /* nb_divide */
wrap_mod, /* nb_remainder */
wrap_divmod, /* nb_divmod */
wrap_pow, /* nb_power */
wrap_neg, /* nb_negative */
wrap_pos, /* nb_positive */
wrap_abs, /* nb_absolute */
wrap_nonzero, /* nb_nonzero */
wrap_invert, /* nb_invert */
wrap_lshift, /* nb_lshift */
wrap_rshift, /* nb_rshift */
wrap_and, /* nb_and */
wrap_xor, /* nb_xor */
wrap_or, /* nb_or */
wrap_coerce, /* nb_coerce */
wrap_int, /* nb_int */
wrap_long, /* nb_long */
wrap_float, /* nb_float */
wrap_oct, /* nb_oct */
wrap_hex, /* nb_hex */
/* Added in release 2.0 */
/* These require the Py_TPFLAGS_HAVE_INPLACEOPS flag */
wrap_iadd, /* nb_inplace_add */
wrap_isub, /* nb_inplace_subtract */
wrap_imul, /* nb_inplace_multiply */
wrap_idiv, /* nb_inplace_divide */
wrap_imod, /* nb_inplace_remainder */
(ternaryfunc)wrap_ipow, /* nb_inplace_power */
wrap_ilshift, /* nb_inplace_lshift */
wrap_irshift, /* nb_inplace_rshift */
wrap_iand, /* nb_inplace_and */
wrap_ixor, /* nb_inplace_xor */
wrap_ior, /* nb_inplace_or */
/* Added in release 2.2 */
/* These require the Py_TPFLAGS_HAVE_CLASS flag */
wrap_floordiv, /* nb_floor_divide */
wrap_truediv, /* nb_true_divide */
wrap_ifloordiv, /* nb_inplace_floor_divide */
wrap_itruediv, /* nb_inplace_true_divide */
};
static PySequenceMethods
wrap_as_sequence = {
wrap_length, /* sq_length */
0, /* sq_concat */
0, /* sq_repeat */
0, /* sq_item */
wrap_slice, /* sq_slice */
0, /* sq_ass_item */
wrap_ass_slice, /* sq_ass_slice */
wrap_contains, /* sq_contains */
};
static PyMappingMethods
wrap_as_mapping = {
wrap_length, /* mp_length */
wrap_getitem, /* mp_subscript */
wrap_setitem, /* mp_ass_subscript */
};
static PyMethodDef
wrap_methods[] = {
{"__reduce__", (PyCFunction)wrap_reduce, METH_NOARGS, reduce__doc__},
{NULL, NULL},
};
/*
* Note that the numeric methods are not supported. This is primarily
* because of the way coercion-less operations are performed with
* new-style numbers; since we can't tell which side of the operation
* is 'self', we can't ensure we'd unwrap the right thing to perform
* the actual operation. We also can't afford to just unwrap both
* sides the way weakrefs do, since we don't know what semantics will
* be associated with the wrapper itself.
*/
statichere PyTypeObject
ProxyType = {
PyObject_HEAD_INIT(NULL) /* PyObject_HEAD_INIT(&PyType_Type) */
0,
"zope.proxy.ProxyBase",
sizeof(ProxyObject),
0,
wrap_dealloc, /* tp_dealloc */
wrap_print, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
wrap_compare, /* tp_compare */
wrap_repr, /* tp_repr */
&wrap_as_number, /* tp_as_number */
&wrap_as_sequence, /* tp_as_sequence */
&wrap_as_mapping, /* tp_as_mapping */
wrap_hash, /* tp_hash */
wrap_call, /* tp_call */
wrap_str, /* tp_str */
wrap_getattro, /* tp_getattro */
wrap_setattro, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC
| Py_TPFLAGS_CHECKTYPES | Py_TPFLAGS_BASETYPE, /* tp_flags */
0, /* tp_doc */
wrap_traverse, /* tp_traverse */
wrap_clear, /* tp_clear */
wrap_richcompare, /* tp_richcompare */
0, /* tp_weaklistoffset */
wrap_iter, /* tp_iter */
wrap_iternext, /* tp_iternext */
wrap_methods, /* tp_methods */
0, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
wrap_init, /* tp_init */
0, /* tp_alloc */
wrap_new, /* tp_new */
0, /*_PyObject_GC_Del,*/ /* tp_free */
};
static PyObject *
create_proxy(PyObject *object)
{
PyObject *result = NULL;
PyObject *args;
args = PyTuple_New(1);
if (args != NULL) {
Py_INCREF(object);
PyTuple_SET_ITEM(args, 0, object);
result = PyObject_CallObject((PyObject *)&ProxyType, args);
Py_DECREF(args);
}
return result;
}
static int
api_check(PyObject *obj)
{
return obj ? Proxy_Check(obj) : 0;
}
static PyObject *
api_create(PyObject *object)
{
if (object == NULL) {
PyErr_SetString(PyExc_ValueError,
"cannot create proxy around NULL");
return NULL;
}
return create_proxy(object);
}
static PyObject *
api_getobject(PyObject *proxy)
{
if (proxy == NULL) {
PyErr_SetString(PyExc_RuntimeError,
"cannot pass NULL to ProxyAPI.getobject()");
return NULL;
}
if (Proxy_Check(proxy))
return Proxy_GET_OBJECT(proxy);
else {
PyErr_Format(PyExc_TypeError, "expected proxy object, got %s",
proxy->ob_type->tp_name);
return NULL;
}
}
static ProxyInterface
wrapper_capi = {
&ProxyType,
api_check,
api_create,
api_getobject,
};
static PyObject *api_object = NULL;
static char
getobject__doc__[] =
"getProxiedObject(proxy) --> object\n"
"\n"
"Get the underlying object for proxy, or the object itself, if it is\n"
"not a proxy.";
static PyObject *
wrapper_getobject(PyObject *unused, PyObject *obj)
{
if (Proxy_Check(obj))
obj = Proxy_GET_OBJECT(obj);
if (obj == NULL)
obj = Py_None;
Py_INCREF(obj);
return obj;
}
static char
isProxy__doc__[] =
"Check whether the given object is a proxy\n"
"\n"
"If proxytype is not None, checkes whether the object is\n"
"proxied by the given proxytype.\n"
;
static PyObject *
wrapper_isProxy(PyObject *unused, PyObject *args)
{
PyObject *obj, *result;
PyTypeObject *proxytype=&ProxyType;
if (! PyArg_ParseTuple(args, "O|O!:isProxy",
&obj, &PyType_Type, &proxytype)
)
return NULL;
while (obj && Proxy_Check(obj))
{
if (PyObject_TypeCheck(obj, proxytype))
{
result = Py_True;
Py_INCREF(result);
return result;
}
obj = Proxy_GET_OBJECT(obj);
}
result = Py_False;
Py_INCREF(result);
return result;
}
static char
removeAllProxies__doc__[] =
"removeAllProxies(proxy) --> object\n"
"\n"
"Get the proxied object with no proxies\n"
"\n"
"If obj is not a proxied object, return obj.\n"
"\n"
"The returned object has no proxies.\n"
;
static PyObject *
wrapper_removeAllProxies(PyObject *unused, PyObject *obj)
{
while (obj && Proxy_Check(obj))
obj = Proxy_GET_OBJECT(obj);
if (obj == NULL)
obj = Py_None;
Py_INCREF(obj);
return obj;
}
static char
sameProxiedObjects__doc__[] =
"Check whether two objects are the same or proxies of the same object";
static PyObject *
wrapper_sameProxiedObjects(PyObject *unused, PyObject *args)
{
PyObject *ob1, *ob2;
if (! PyArg_ParseTuple(args, "OO:sameProxiedObjects", &ob1, &ob2))
return NULL;
while (ob1 && Proxy_Check(ob1))
ob1 = Proxy_GET_OBJECT(ob1);
while (ob2 && Proxy_Check(ob2))
ob2 = Proxy_GET_OBJECT(ob2);
if (ob1 == ob2)
ob1 = Py_True;
else
ob1 = Py_False;
Py_INCREF(ob1);
return ob1;
}
static char
queryProxy__doc__[] =
"Look for a proxy of the given type around the object\n"
"\n"
"If no such proxy can be found, return the default.\n"
;
static PyObject *
wrapper_queryProxy(PyObject *unused, PyObject *args)
{
PyObject *obj, *result=Py_None;
PyTypeObject *proxytype=&ProxyType;
if (! PyArg_ParseTuple(args, "O|O!O:queryProxy",
&obj, &PyType_Type, &proxytype, &result)
)
return NULL;
while (obj && Proxy_Check(obj))
{
if (PyObject_TypeCheck(obj, proxytype))
{
Py_INCREF(obj);
return obj;
}
obj = Proxy_GET_OBJECT(obj);
}
Py_INCREF(result);
return result;
}
static char
queryInnerProxy__doc__[] =
"Look for the inner-most proxy of the given type around the object\n"
"\n"
"If no such proxy can be found, return the default.\n"
"\n"
"If there is such a proxy, return the inner-most one.\n"
;
static PyObject *
wrapper_queryInnerProxy(PyObject *unused, PyObject *args)
{
PyObject *obj, *result=Py_None;
PyTypeObject *proxytype=&ProxyType;
if (! PyArg_ParseTuple(args, "O|O!O:queryInnerProxy",
&obj, &PyType_Type, &proxytype, &result)
)
return NULL;
while (obj && Proxy_Check(obj))
{
if (PyObject_TypeCheck(obj, proxytype))
result = obj;
obj = Proxy_GET_OBJECT(obj);
}
Py_INCREF(result);
return result;
}
static char
module___doc__[] =
"Association between an object, a context object, and a dictionary.\n\
\n\
The context object and dictionary give additional context information\n\
associated with a reference to the basic object. The wrapper objects\n\
act as proxies for the original object.";
static PyMethodDef
module_functions[] = {
{"getProxiedObject", wrapper_getobject, METH_O, getobject__doc__},
{"isProxy", wrapper_isProxy, METH_VARARGS, isProxy__doc__},
{"sameProxiedObjects", wrapper_sameProxiedObjects, METH_VARARGS,
sameProxiedObjects__doc__},
{"queryProxy", wrapper_queryProxy, METH_VARARGS, queryProxy__doc__},
{"queryInnerProxy", wrapper_queryInnerProxy, METH_VARARGS,
queryInnerProxy__doc__},
{"removeAllProxies", wrapper_removeAllProxies, METH_O,
removeAllProxies__doc__},
{NULL}
};
void
init_zope_proxy_proxy(void)
{
PyObject *m = Py_InitModule3("_zope_proxy_proxy",
module_functions, module___doc__);
if (m == NULL)
return;
if (empty_tuple == NULL)
empty_tuple = PyTuple_New(0);
ProxyType.tp_free = _PyObject_GC_Del;
if (PyType_Ready(&ProxyType) < 0)
return;
Py_INCREF(&ProxyType);
PyModule_AddObject(m, "ProxyBase", (PyObject *)&ProxyType);
if (api_object == NULL) {
api_object = PyCObject_FromVoidPtr(&wrapper_capi, NULL);
if (api_object == NULL)
return;
}
Py_INCREF(api_object);
PyModule_AddObject(m, "_CAPI", api_object);
}
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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
#
##############################################################################
"""Proxy-related interfaces.
$Id$
"""
from zope.interface import Interface
class IProxyIntrospection(Interface):
"""Provides methods for indentifying proxies and extracting proxied objects
"""
def isProxy(obj, proxytype=None):
"""Check whether the given object is a proxy
If proxytype is not None, checkes whether the object is
proxied by the given proxytype.
"""
def sameProxiedObjects(ob1, ob2):
"""Check whether ob1 and ob2 are the same or proxies of the same object
"""
def getProxiedObject(obj):
"""Get the proxied Object
If the object isn't proxied, then just return the object.
"""
def removeAllProxies(obj):
"""Get the proxied object with no proxies
If obj is not a proxied object, return obj.
The returned object has no proxies.
"""
def queryProxy(obj, proxytype, default=None):
"""Look for a proxy of the given type around the object
If no such proxy can be found, return the default.
"""
def queryInnerProxy(obj, proxytype, default=None):
"""Look for the inner-most proxy of the given type around the object
If no such proxy can be found, return the default.
If there is such a proxy, return the inner-most one.
"""
#ifndef _proxy_H_
#define _proxy_H_ 1
typedef struct {
PyObject_HEAD
PyObject *proxy_object;
} ProxyObject;
#define Proxy_GET_OBJECT(ob) (((ProxyObject *)(ob))->proxy_object)
typedef struct {
PyTypeObject *proxytype;
int (*check)(PyObject *obj);
PyObject *(*create)(PyObject *obj);
PyObject *(*getobject)(PyObject *proxy);
} ProxyInterface;
#ifndef PROXY_MODULE
/* These are only defined in the public interface, and are not
* available within the module implementation. There we use the
* classic Python/C API only.
*/
static ProxyInterface *_proxy_api = NULL;
static int
Proxy_Import(void)
{
if (_proxy_api == NULL) {
PyObject *m = PyImport_ImportModule("zope.proxy");
if (m != NULL) {
PyObject *tmp = PyObject_GetAttrString(m, "_CAPI");
if (tmp != NULL) {
if (PyCObject_Check(tmp))
_proxy_api = (ProxyInterface *)
PyCObject_AsVoidPtr(tmp);
Py_DECREF(tmp);
}
}
}
return (_proxy_api == NULL) ? -1 : 0;
}
#define ProxyType (*_proxy_api->proxytype)
#define Proxy_Check(obj) (_proxy_api->check((obj)))
#define Proxy_CheckExact(obj) ((obj)->ob_type == ProxyType)
#define Proxy_New(obj) (_proxy_api->create((obj)))
#define Proxy_GetObject(proxy) (_proxy_api->getobject((proxy)))
#endif /* PROXY_MODULE */
#endif /* _proxy_H_ */
#
# This file is necessary to make this directory a package.
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
##############################################################################
"""Test base proxy class.
$Id$
"""
import pickle
import sys
import unittest
from zope.testing.doctestunit import DocTestSuite
from zope.proxy import ProxyBase
class Thing:
"""This class is expected to be a classic class."""
class Comparable(object):
def __init__(self, value):
self.value = value
def __eq__(self, other):
if hasattr(other, "value"):
other = other.value
return self.value == other
def __ne__(self, other):
return not self.__eq__(other)
def __lt__(self, other):
if hasattr(other, "value"):
other = other.value
return self.value < other
def __ge__(self, other):
return not self.__lt__(other)
def __le__(self, other):
if hasattr(other, "value"):
other = other.value
return self.value <= other
def __gt__(self, other):
return not self.__le__(other)
def __repr__(self):
return "<Comparable: %r>" % self.value
class ProxyTestCase(unittest.TestCase):
proxy_class = ProxyBase
def setUp(self):
self.x = Thing()
self.p = self.new_proxy(self.x)
def new_proxy(self, o):
return self.proxy_class(o)
def test_constructor(self):
o = object()
self.assertRaises(TypeError, self.proxy_class, o, o)
self.assertRaises(TypeError, self.proxy_class, o, key='value')
self.assertRaises(TypeError, self.proxy_class, key='value')
def test_subclass_constructor(self):
class MyProxy(self.proxy_class):
def __new__(cls, *args, **kwds):
return super(MyProxy, cls).__new__(cls, *args, **kwds)
def __init__(self, *args, **kwds):
super(MyProxy, self).__init__(*args, **kwds)
o1 = object()
o2 = object()
o = MyProxy((o1, o2))
self.assertEquals(o1, o[0])
self.assertEquals(o2, o[1])
self.assertRaises(TypeError, MyProxy, o1, o2)
self.assertRaises(TypeError, MyProxy, o1, key='value')
self.assertRaises(TypeError, MyProxy, key='value')
# Check that are passed to __init__() overrides what's passed
# to __new__().
class MyProxy2(self.proxy_class):
def __new__(cls, *args, **kwds):
return super(MyProxy2, cls).__new__(cls, 'value')
p = MyProxy2('splat!')
self.assertEquals(list(p), list('splat!'))
class MyProxy3(MyProxy2):
def __init__(self, arg):
if list(self) != list('value'):
raise AssertionError("list(self) != list('value')")
super(MyProxy3, self).__init__('another')
p = MyProxy3('notused')
self.assertEquals(list(p), list('another'))
def test_proxy_attributes(self):
o = Thing()
o.foo = 1
w = self.new_proxy(o)
self.assert_(w.foo == 1)
def test___class__(self):
o = object()
w = self.new_proxy(o)
self.assert_(w.__class__ is o.__class__)
def test_pickle_prevention(self):
w = self.new_proxy(Thing())
self.assertRaises(pickle.PicklingError,
pickle.dumps, w)
def test_proxy_equality(self):
w = self.new_proxy('foo')
self.assertEquals(w, 'foo')
o1 = Comparable(1)
o2 = Comparable(1.0)
o3 = Comparable("splat!")
w1 = self.new_proxy(o1)
w2 = self.new_proxy(o2)
w3 = self.new_proxy(o3)
self.assertEquals(o1, w1)
self.assertEquals(o1, w2)
self.assertEquals(o2, w1)
self.assertEquals(w1, o2)
self.assertEquals(w2, o1)
self.assertNotEquals(o3, w1)
self.assertNotEquals(w1, o3)
self.assertNotEquals(w3, o1)
self.assertNotEquals(o1, w3)
def test_proxy_ordering_lt(self):
o1 = Comparable(1)
o2 = Comparable(2.0)
w1 = self.new_proxy(o1)
w2 = self.new_proxy(o2)
self.assert_(w1 < w2)
self.assert_(w1 <= w2)
self.assert_(o1 < w2)
self.assert_(o1 <= w2)
self.assert_(w1 < o2)
self.assert_(w2 <= o2)
def test_proxy_callable(self):
w = self.new_proxy({}.get)
self.assert_(callable(w))
def test_proxy_item_protocol(self):
w = self.new_proxy({})
self.assertRaises(KeyError, lambda: w[1])
w[1] = 'a'
self.assertEquals(w[1], 'a')
del w[1]
self.assertRaises(KeyError, lambda: w[1])
def del_w_1():
del w[1]
self.assertRaises(KeyError, del_w_1)
def test_wrapped_iterable(self):
a = [1, 2, 3]
b = []
for x in self.new_proxy(a):
b.append(x)
self.assertEquals(a, b)
def test_iteration_over_proxy(self):
# Wrap an iterator before starting iteration.
# PyObject_GetIter() will still be called on the proxy.
a = [1, 2, 3]
b = []
for x in self.new_proxy(iter(a)):
b.append(x)
self.assertEquals(a, b)
t = tuple(self.new_proxy(iter(a)))
self.assertEquals(t, (1, 2, 3))
def test_iteration_using_proxy(self):
# Wrap an iterator within the iteration protocol, expecting it
# still to work. PyObject_GetIter() will not be called on the
# proxy, so the tp_iter slot won't unwrap it.
class Iterable(object):
def __init__(self, test, data):
self.test = test
self.data = data
def __iter__(self):
return self.test.new_proxy(iter(self.data))
a = [1, 2, 3]
b = []
for x in Iterable(self, a):
b.append(x)
self.assertEquals(a, b)
def test_bool_wrapped_None(self):
w = self.new_proxy(None)
self.assertEquals(not w, 1)
# Numeric ops.
unops = [
"-x", "+x", "abs(x)", "~x",
"int(x)", "long(x)", "float(x)",
]
def test_unops(self):
P = self.new_proxy
for expr in self.unops:
x = 1
y = eval(expr)
x = P(1)
z = eval(expr)
self.assertEqual(z, y,
"x=%r; expr=%r" % (x, expr))
def test_odd_unops(self):
# unops that don't return a proxy
P = self.new_proxy
for func in hex, oct, lambda x: not x:
self.assertEqual(func(P(100)), func(100))
binops = [
"x+y", "x-y", "x*y", "x/y", "divmod(x, y)", "x**y", "x//y",
"x<<y", "x>>y", "x&y", "x|y", "x^y",
]
def test_binops(self):
P = self.new_proxy
for expr in self.binops:
first = 1
for x in [1, P(1)]:
for y in [2, P(2)]:
if first:
z = eval(expr)
first = 0
else:
self.assertEqual(eval(expr), z,
"x=%r; y=%r; expr=%r" % (x, y, expr))
def test_inplace(self):
# TODO: should test all inplace operators...
P = self.new_proxy
pa = P(1)
pa += 2
self.assertEqual(pa, 3)
a = [1, 2, 3]
pa = qa = P(a)
pa += [4, 5, 6]
self.failUnless(pa is qa)
self.assertEqual(a, [1, 2, 3, 4, 5, 6])
pa = P(2)
pa **= 2
self.assertEqual(pa, 4)
def test_coerce(self):
P = self.new_proxy
# Before 2.3, coerce() of two proxies returns them unchanged
fixed_coerce = sys.version_info >= (2, 3, 0)
x = P(1)
y = P(2)
a, b = coerce(x, y)
self.failUnless(a is x and b is y)
x = P(1)
y = P(2.1)
a, b = coerce(x, y)
self.failUnless(a == 1.0)
self.failUnless(b is y)
if fixed_coerce:
self.failUnless(a.__class__ is float, a.__class__)
x = P(1.1)
y = P(2)
a, b = coerce(x, y)
self.failUnless(a is x)
self.failUnless(b == 2.0)
if fixed_coerce:
self.failUnless(b.__class__ is float, b.__class__)
x = P(1)
y = 2
a, b = coerce(x, y)
self.failUnless(a is x)
self.failUnless(b is y)
x = P(1)
y = 2.1
a, b = coerce(x, y)
self.failUnless(a.__class__ is float, a.__class__)
self.failUnless(b is y)
x = P(1.1)
y = 2
a, b = coerce(x, y)
self.failUnless(a is x)
self.failUnless(b.__class__ is float, b.__class__)
x = 1
y = P(2)
a, b = coerce(x, y)
self.failUnless(a is x)
self.failUnless(b is y)
x = 1.1
y = P(2)
a, b = coerce(x, y)
self.failUnless(a is x)
self.failUnless(b.__class__ is float, b.__class__)
x = 1
y = P(2.1)
a, b = coerce(x, y)
self.failUnless(a.__class__ is float, a.__class__)
self.failUnless(b is y)
def test_isProxy():
"""
>>> from zope.proxy import ProxyBase, isProxy
>>> class P1(ProxyBase):
... pass
>>> class P2(ProxyBase):
... pass
>>> class C(object):
... pass
>>> c = C()
>>> int(isProxy(c))
0
>>> p = P1(c)
>>> int(isProxy(p))
1
>>> int(isProxy(p, P1))
1
>>> int(isProxy(p, P2))
0
>>> p = P2(p)
>>> int(isProxy(p, P1))
1
>>> int(isProxy(p, P2))
1
"""
def test_getProxiedObject():
"""
>>> from zope.proxy import ProxyBase, getProxiedObject
>>> class C(object):
... pass
>>> c = C()
>>> int(getProxiedObject(c) is c)
1
>>> p = ProxyBase(c)
>>> int(getProxiedObject(p) is c)
1
>>> p2 = ProxyBase(p)
>>> int(getProxiedObject(p2) is p)
1
"""
def test_ProxyIterator():
"""
>>> from zope.proxy import ProxyBase, ProxyIterator
>>> class C(object):
... pass
>>> c = C()
>>> p1 = ProxyBase(c)
>>> class P(ProxyBase):
... pass
>>> p2 = P(p1)
>>> p3 = ProxyBase(p2)
>>> list(ProxyIterator(p3)) == [p3, p2, p1, c]
1
"""
def test_removeAllProxies():
"""
>>> from zope.proxy import ProxyBase, removeAllProxies
>>> class C(object):
... pass
>>> c = C()
>>> int(removeAllProxies(c) is c)
1
>>> p = ProxyBase(c)
>>> int(removeAllProxies(p) is c)
1
>>> p2 = ProxyBase(p)
>>> int(removeAllProxies(p2) is c)
1
"""
def test_queryProxy():
"""
>>> from zope.proxy import ProxyBase, queryProxy
>>> class P1(ProxyBase):
... pass
>>> class P2(ProxyBase):
... pass
>>> class C(object):
... pass
>>> c = C()
>>> queryProxy(c, P1)
>>> queryProxy(c, P1, 42)
42
>>> p1 = P1(c)
>>> int(queryProxy(p1, P1) is p1)
1
>>> queryProxy(c, P2)
>>> queryProxy(c, P2, 42)
42
>>> p2 = P2(p1)
>>> int(queryProxy(p2, P1) is p1)
1
>>> int(queryProxy(p2, P2) is p2)
1
>>> int(queryProxy(p2, ProxyBase) is p2)
1
"""
def test_queryInnerProxy():
"""
>>> from zope.proxy import ProxyBase, queryProxy, queryInnerProxy
>>> class P1(ProxyBase):
... pass
>>> class P2(ProxyBase):
... pass
>>> class C(object):
... pass
>>> c = C()
>>> queryInnerProxy(c, P1)
>>> queryInnerProxy(c, P1, 42)
42
>>> p1 = P1(c)
>>> int(queryProxy(p1, P1) is p1)
1
>>> queryInnerProxy(c, P2)
>>> queryInnerProxy(c, P2, 42)
42
>>> p2 = P2(p1)
>>> int(queryInnerProxy(p2, P1) is p1)
1
>>> int(queryInnerProxy(p2, P2) is p2)
1
>>> int(queryInnerProxy(p2, ProxyBase) is p1)
1
>>> p3 = P1(p2)
>>> int(queryProxy(p3, P1) is p3)
1
>>> int(queryInnerProxy(p3, P1) is p1)
1
>>> int(queryInnerProxy(p3, P2) is p2)
1
"""
def test_sameProxiedObjects():
"""
>>> from zope.proxy import ProxyBase, sameProxiedObjects
>>> class C(object):
... pass
>>> c1 = C()
>>> c2 = C()
>>> int(sameProxiedObjects(c1, c1))
1
>>> int(sameProxiedObjects(ProxyBase(c1), c1))
1
>>> int(sameProxiedObjects(ProxyBase(c1), ProxyBase(c1)))
1
>>> int(sameProxiedObjects(ProxyBase(ProxyBase(c1)), c1))
1
>>> int(sameProxiedObjects(c1, ProxyBase(c1)))
1
>>> int(sameProxiedObjects(c1, ProxyBase(ProxyBase(c1))))
1
>>> int(sameProxiedObjects(c1, c2))
0
>>> int(sameProxiedObjects(ProxyBase(c1), c2))
0
>>> int(sameProxiedObjects(ProxyBase(c1), ProxyBase(c2)))
0
>>> int(sameProxiedObjects(ProxyBase(ProxyBase(c1)), c2))
0
>>> int(sameProxiedObjects(c1, ProxyBase(c2)))
0
>>> int(sameProxiedObjects(c1, ProxyBase(ProxyBase(c2))))
0
"""
def test_subclassing_proxies():
"""You can subclass ProxyBase
If you subclass a proxy, instances of the subclass have access to
data defined in the class, including descriptors.
Your subclass instances don't get instance dictionaries, but they
can have slots.
>>> class MyProxy(ProxyBase):
... __slots__ = 'x', 'y'
...
... def f(self):
... return self.x
>>> l = [1, 2, 3]
>>> p = MyProxy(l)
I can use attributes defined by the class, including slots:
>>> p.x = 'x'
>>> p.x
'x'
>>> p.f()
'x'
I can also use attributes of the proxied object:
>>> p
[1, 2, 3]
>>> p.pop()
3
>>> p
[1, 2]
"""
def test_suite():
suite = unittest.makeSuite(ProxyTestCase)
suite.addTest(DocTestSuite())
return suite
if __name__ == "__main__":
runner = unittest.TextTestRunner(sys.stdout)
result = runner.run(test_suite())
newerrs = len(result.errors) + len(result.failures)
sys.exit(newerrs and 1 or 0)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment