def test_add_operators_to_field_reference_field():
from mongoengine import ReferenceField, Document, StringField
from graphene_mongo.operators import gen_operators_of_field, allowed_operators
from graphene_mongo.fields import respective_special_fields
class Other(Document):
name = StringField()
class Test(Document):
test = ReferenceField(Other)
field = Test.test
r_graphene = respective_special_fields[type(field)]
applied_operators = gen_operators_of_field('test', field, r_graphene('test', field), allowed_operators(field))
assert sorted(list(applied_operators.keys())) == format_fields(['in', 'nin', 'ne'])
assert isinstance(applied_operators['test__in'], graphene.List)
assert isinstance(applied_operators['test__nin'], graphene.List)
assert isinstance(applied_operators['test__ne'], graphene.ID)
python类Document()的实例源码
def test_mutate_wrong_number_arguments():
from mongoengine import Document
class Test(Document):
pass
def mutate():
pass
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test, 'mutate': staticmethod(mutate)})
assert str(e_info.value) == 'Failed to generate schema {}, mutate method must accept two params. ' \
'The first is the arguments passed to mutate in query, for instance: ' \
'username:"NewObjName". Second is the context of the application, if it is flask, ' \
'will be flask global request.'.format('TestSchema')
def test_validator_wrong():
from mongoengine import Document, StringField
class Test(Document):
parent = StringField()
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test, 'validator': True})
assert str(e_info.value) == "'validator' attribute must be callable."
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test, 'validator': lambda x: x})
assert str(e_info.value) == ("The 'validator' attribute must be a callable that accepts four arguments: "
"model, fields, query, special_params")
def setUp(self):
"""Setup class."""
from bson import DBRef, ObjectId
class Source(db.Document):
pass
class Model(db.Document):
src = db.ReferenceField(Source, dbref=True)
self.src_cls = Source
self.model_cls = Model
self.src_id = ObjectId()
self.data = json.dumps({
"src": {"collection": "source", "id": str(self.src_id)}
})
self.expected_data = {"src": DBRef("source", self.src_id)}
self.hook = generate_object_hook(self.model_cls)
def setUp(self):
"""Setup class."""
from bson import ObjectId, DBRef
class Source(db.Document):
pass
class Model(db.Document):
src = db.ReferenceField(Source)
self.src_cls = Source
self.model_cls = Model
self.src_id = ObjectId()
self.data = json.dumps({
"src": str(self.src_id)
})
self.expected_data = {
"src": DBRef("source", self.src_id)
}
self.hook = generate_object_hook(self.model_cls)
def setUp(self):
"""Setup test."""
from datetime import datetime, timedelta
from calendar import timegm
class DateTime(db.Document):
date = db.DateTimeField()
self.model_cls = DateTime
now = datetime.utcnow()
epoch_mil = int(timegm(now.timetuple())*1000 + now.microsecond / 1000)
self.data = json.dumps({"date": epoch_mil})
self.expected_data = {
"date": datetime.utcfromtimestamp(
int(epoch_mil / 1000)
) + timedelta(milliseconds=int(epoch_mil % 1000))
}
self.hook = generate_object_hook(self.model_cls)
def setUp(self):
"""Setup test."""
from datetime import datetime
class DateTime(db.Document):
date = db.DateTimeField()
self.model_cls = DateTime
now = datetime.utcnow()
# This format shouldn't be supported.
self.data = json.dumps({"date": {
"year": now.year,
"month": now.month,
"date": now.day
}})
self.hook = generate_object_hook(self.model_cls)
def init_mappings(self):
"""Initialize RBAC Mappings.
RBAC Mappings always refer to a (Organization, Team) combination.
In order to reference the Organization instance of this Team, we
are using `self._instance`, which is a proxy object to the upper
level Document.
"""
if not RBACMapping:
return
if self.name == 'Owners':
return
if RBACMapping.objects(org=self._instance.id, team=self.id).only('id'):
raise me.ValidationError(
'RBAC Mappings already initialized for Team %s' % self
)
for perm in ('read', 'read_logs'):
RBACMapping(
org=self._instance.id, team=self.id, permission=perm
).save()
def test_not_mongoengine_document():
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': False})
assert str(e_info.value) == 'Failed to generate schema {}, model must be ' \
'a subclass of mongoengine.Document.'.format('TestSchema')
def test_mutate_not_static():
from mongoengine import Document
class Test(Document):
pass
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test, 'mutate': 'To fails'})
assert str(e_info.value) == 'Failed to generate schema {}, mutate method must ' \
'be a method with the decorator staticmethod.'.format('TestSchema')
def test_mongoengine_field_not_implemented():
from mongoengine import FileField, Document
class Test(Document):
field_error = FileField()
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test})
assert str(e_info.value) == "It was not possible to generate schema for {} because the " \
"field {} is of the type {}, and that field is not supported yet." \
.format("Test", 'field_error', FileField)
def test_mongoengine_field_references_self():
from mongoengine import Document, ReferenceField
class Test(Document):
parent = ReferenceField('self')
with pytest.raises(Exception) as e_info:
Options('TestSchema', {'model': Test})
assert str(e_info.value) == "It was not possible to generate schema for {} because the field {} is a " \
"ReferenceField to self and this is not supported yet."\
.format("TestSchema", 'parent')
def to_mongo(self, document, **kwargs):
"""
Convert to python-typed dict.
Parameters:
document: The document.
"""
cur_depth = getattr(self, "$$cur_depth$$", self.max_depth)
if not getattr(self, "$$good_json$$", None) or \
cur_depth >= self.max_depth:
return super(FollowReferenceField, self).to_mongo(
document, **kwargs
)
doc = document
if isinstance(document, db.Document):
if document.pk is None and self.id_check:
self.error("The referenced document needs ID.")
else:
doc = self.document_type.objects(
pk=super(
FollowReferenceField, self
).to_mongo(document, **kwargs)
).get()
if isinstance(doc, Document):
doc.begin_goodjson(cur_depth + 1)
ret = doc.to_mongo(**kwargs)
if isinstance(doc, Document):
doc.end_goodjson(cur_depth + 1)
if "_id" in ret and issubclass(self.document_type, Document):
ret["id"] = ret.pop("_id", None)
return ret
def setUp(self):
"""Setup class."""
from bson import ObjectId
class TestModel(db.Document):
user = db.ObjectIdField()
self.model_cls = TestModel
self.hook = generate_object_hook(self.model_cls)
self.data = json.dumps({"user": "56f63a716a8dec7705f36409"})
self.expected_data = {"user": ObjectId("56f63a716a8dec7705f36409")}
def setUp(self):
"""Setup test."""
from base64 import b64encode, b64decode
from bson.binary import Binary, BINARY_SUBTYPE
class BinaryTest(db.Document):
text = db.BinaryField()
self.model_cls = BinaryTest
self.data = {
"text": {
"data": b64encode(
("This is a test").encode("utf-8")
).decode("utf-8"),
"type": BINARY_SUBTYPE
}
}
self.expected_data = {
"text": Binary(
b64decode(self.data["text"]["data"]),
self.data["text"]["type"]
)
}
self.data = json.dumps(self.data)
self.hook = generate_object_hook(BinaryTest)
def setUp(self):
"""Setup test."""
from uuid import uuid5, NAMESPACE_DNS
class UUIDModel(db.Document):
uuid = db.UUIDField()
self.model_cls = UUIDModel
uuid = uuid5(NAMESPACE_DNS, "This is a test")
self.expected_data = {
"uuid": uuid
}
self.data = json.dumps({"uuid": str(uuid)})
self.hook = generate_object_hook(UUIDModel)
def __init__(self, app=None, config=None):
_include_mongoengine(self)
self.app = None
self.Document = Document
self.DynamicDocument = DynamicDocument
if app is not None:
self.init_app(app, config)
def _is_foreign_key(field_cls, field_name):
return field_cls and issubclass(field_cls, Document) and not _is_dereference_disabled(field_name)
def verified_attrs(self, attrs):
""" Function to verify if the attributes is of the right type """
model = attrs.get('model')
mutate = attrs.get('mutate')
validator = attrs.get('validator')
validator = validator.__func__ if isinstance(validator, staticmethod) else validator
if model is None:
raise AttributeError('Failed to generate schema {},'
' model attribute was not given.'.format(self.class_name))
if not model or not issubclass(model, Document):
raise TypeError('Failed to generate schema {}, model must be '
'a subclass of mongoengine.Document.'.format(self.class_name))
if mutate and not isinstance(mutate, staticmethod):
raise TypeError('Failed to generate schema {}, mutate method must '
'be a method with the decorator staticmethod.'.format(self.class_name))
if mutate and len(inspect.signature(mutate.__func__).parameters) != 2:
raise TypeError('Failed to generate schema {}, mutate method must accept two params. '
'The first is the arguments passed to mutate in query, for instance: username:"NewObjName".'
' Second is the context of the application, if it is flask, will be flask global request.'
.format(self.class_name))
# verify if all fields of document is supported
for f_name, m_field in model._fields.items():
if type(m_field) not in respective_fields and type(m_field) not in respective_special_fields:
raise NotImplementedError("It was not possible to generate schema for {} because the "
"field {} is of the type {}, and that field is not supported yet."
.format(model.__name__, f_name, type(m_field)))
if isinstance(m_field, list_fields):
if type(m_field.field) not in respective_fields and type(m_field.field) not in respective_special_fields:
raise NotImplementedError("It was not possible to generate schema for {} because the "
"field {} is a List of the type {}, and that field is not supported yet."
.format(model.__name__, f_name, type(m_field.field)))
if isinstance(m_field, reference_fields):
if isinstance(m_field.document_type_obj, str) and m_field.document_type_obj == 'self':
raise NotImplementedError("It was not possible to generate schema for {} because the "
"field {} is a ReferenceField to self and this is not supported yet."
.format(self.class_name, f_name))
if validator and not callable(validator):
raise AttributeError("'validator' attribute must be callable.")
elif validator and len(inspect.signature(validator).parameters) != 4:
raise AttributeError("The 'validator' attribute must be a callable that accepts four arguments: "
"model, fields, query, special_params")
return model, mutate.__func__ if mutate else None, model._fields, validator
