# -*- coding: utf-8 -*-
# Copyright: See the LICENSE file.
"""factory_boy extensions for use with the Django framework."""
from __future__ import absolute_import
from __future__ import unicode_literals
import os
import logging
import functools
try:
import django
from django.core import files as django_files
except ImportError as e: # pragma: no cover
django = None
django_files = None
import_failure = e
from . import base
from . import declarations
from . import errors
from .compat import BytesIO, is_string
logger = logging.getLogger('factory.generate')
DEFAULT_DB_ALIAS = 'default' # Same as django.db.DEFAULT_DB_ALIAS
def require_django():
"""Simple helper to ensure Django is available."""
if django_files is None: # pragma: no cover
raise import_failure
_LAZY_LOADS = {}
def get_model(app, model):
"""Wrapper around django's get_model."""
if 'get_model' not in _LAZY_LOADS:
_lazy_load_get_model()
_get_model = _LAZY_LOADS['get_model']
return _get_model(app, model)
def _lazy_load_get_model():
"""Lazy loading of get_model.
get_model loads django.conf.settings, which may fail if
the settings haven't been configured yet.
"""
if django is None:
def _get_model(app, model):
raise import_failure
else:
from django import apps as django_apps
_get_model = django_apps.apps.get_model
_LAZY_LOADS['get_model'] = _get_model
[docs]class DjangoOptions(base.FactoryOptions):
def _build_default_options(self):
return super(DjangoOptions, self)._build_default_options() + [
base.OptionDefault('django_get_or_create', (), inherit=True),
base.OptionDefault('database', DEFAULT_DB_ALIAS, inherit=True),
]
def _get_counter_reference(self):
counter_reference = super(DjangoOptions, self)._get_counter_reference()
if (counter_reference == self.base_factory
and self.base_factory._meta.model is not None
and self.base_factory._meta.model._meta.abstract
and self.model is not None
and not self.model._meta.abstract):
# Target factory is for an abstract model, yet we're for another,
# concrete subclass => don't reuse the counter.
return self.factory
return counter_reference
def get_model_class(self):
if is_string(self.model) and '.' in self.model:
app, model_name = self.model.split('.', 1)
self.model = get_model(app, model_name)
return self.model
[docs]class DjangoModelFactory(base.Factory):
"""Factory for Django models.
This makes sure that the 'sequence' field of created objects is a new id.
Possible improvement: define a new 'attribute' type, AutoField, which would
handle those for non-numerical primary keys.
"""
_options_class = DjangoOptions
class Meta:
abstract = True # Optional, but explicit.
@classmethod
def _load_model_class(cls, definition):
if is_string(definition) and '.' in definition:
app, model = definition.split('.', 1)
return get_model(app, model)
return definition
@classmethod
def _get_manager(cls, model_class):
if model_class is None:
raise errors.AssociatedClassError(
"No model set on %s.%s.Meta" % (cls.__module__, cls.__name__))
try:
manager = model_class.objects
except AttributeError:
# When inheriting from an abstract model with a custom
# manager, the class has no 'objects' field.
manager = model_class._default_manager
if cls._meta.database != DEFAULT_DB_ALIAS:
manager = manager.using(cls._meta.database)
return manager
@classmethod
def _get_or_create(cls, model_class, *args, **kwargs):
"""Create an instance of the model through objects.get_or_create."""
manager = cls._get_manager(model_class)
assert 'defaults' not in cls._meta.django_get_or_create, (
"'defaults' is a reserved keyword for get_or_create "
"(in %s._meta.django_get_or_create=%r)"
% (cls, cls._meta.django_get_or_create))
key_fields = {}
for field in cls._meta.django_get_or_create:
if field not in kwargs:
raise errors.FactoryError(
"django_get_or_create - "
"Unable to find initialization value for '%s' in factory %s" %
(field, cls.__name__))
key_fields[field] = kwargs.pop(field)
key_fields['defaults'] = kwargs
instance, _created = manager.get_or_create(*args, **key_fields)
return instance
@classmethod
def _create(cls, model_class, *args, **kwargs):
"""Create an instance of the model, and save it to the database."""
manager = cls._get_manager(model_class)
if cls._meta.django_get_or_create:
return cls._get_or_create(model_class, *args, **kwargs)
return manager.create(*args, **kwargs)
@classmethod
def _after_postgeneration(cls, instance, create, results=None):
"""Save again the instance if creating and at least one hook ran."""
if create and results:
# Some post-generation hooks ran, and may have modified us.
instance.save()
[docs]class FileField(declarations.ParameteredAttribute):
"""Helper to fill in django.db.models.FileField from a Factory."""
DEFAULT_FILENAME = 'example.dat'
[docs] def __init__(self, **defaults):
require_django()
super(FileField, self).__init__(**defaults)
def _make_data(self, params):
"""Create data for the field."""
return params.get('data', b'')
def _make_content(self, params):
path = ''
_content_params = [params.get('from_path'), params.get('from_file'), params.get('from_func')]
if len([p for p in _content_params if p]) > 1:
raise ValueError(
"At most one argument from 'from_file', 'from_path', and 'from_func' should "
"be non-empty when calling factory.django.FileField."
)
if params.get('from_path'):
path = params['from_path']
f = open(path, 'rb')
content = django_files.File(f, name=path)
elif params.get('from_file'):
f = params['from_file']
content = django_files.File(f)
path = content.name
elif params.get('from_func'):
func = params['from_func']
content = django_files.File(func())
path = content.name
else:
data = self._make_data(params)
content = django_files.base.ContentFile(data)
if path:
default_filename = os.path.basename(path)
else:
default_filename = self.DEFAULT_FILENAME
filename = params.get('filename', default_filename)
return filename, content
def generate(self, step, params):
"""Fill in the field."""
# Recurse into a DictFactory: allows users to have some params depending
# on others.
params = step.recurse(base.DictFactory, params, force_sequence=step.sequence)
filename, content = self._make_content(params)
return django_files.File(content.file, filename)
[docs]class ImageField(FileField):
DEFAULT_FILENAME = 'example.jpg'
def _make_data(self, params):
# ImageField (both django's and factory_boy's) require PIL.
# Try to import it along one of its known installation paths.
try:
from PIL import Image
except ImportError:
import Image
width = params.get('width', 100)
height = params.get('height', width)
color = params.get('color', 'blue')
image_format = params.get('format', 'JPEG')
thumb = Image.new('RGB', (width, height), color)
thumb_io = BytesIO()
thumb.save(thumb_io, format=image_format)
return thumb_io.getvalue()
[docs]class mute_signals(object):
"""Temporarily disables and then restores any django signals.
Args:
*signals (django.dispatch.dispatcher.Signal): any django signals
Examples:
with mute_signals(pre_init):
user = UserFactory.build()
...
@mute_signals(pre_save, post_save)
class UserFactory(factory.Factory):
...
@mute_signals(post_save)
def generate_users():
UserFactory.create_batch(10)
"""
def __init__(self, *signals):
self.signals = signals
self.paused = {}
def __enter__(self):
for signal in self.signals:
logger.debug('mute_signals: Disabling signal handlers %r',
signal.receivers)
# Note that we're using implementation details of
# django.signals, since arguments to signal.connect()
# are lost in signal.receivers
self.paused[signal] = signal.receivers
signal.receivers = []
def __exit__(self, exc_type, exc_value, traceback):
for signal, receivers in self.paused.items():
logger.debug('mute_signals: Restoring signal handlers %r',
receivers)
signal.receivers = receivers
with signal.lock:
# Django uses some caching for its signals.
# Since we're bypassing signal.connect and signal.disconnect,
# we have to keep messing with django's internals.
signal.sender_receivers_cache.clear()
self.paused = {}
def copy(self):
return mute_signals(*self.signals)
def __call__(self, callable_obj):
if isinstance(callable_obj, base.FactoryMetaClass):
# Retrieve __func__, the *actual* callable object.
generate_method = callable_obj._generate.__func__
@classmethod
@functools.wraps(generate_method)
def wrapped_generate(*args, **kwargs):
# A mute_signals() object is not reentrant; use a copy every time.
with self.copy():
return generate_method(*args, **kwargs)
callable_obj._generate = wrapped_generate
return callable_obj
else:
@functools.wraps(callable_obj)
def wrapper(*args, **kwargs):
# A mute_signals() object is not reentrant; use a copy every time.
with self.copy():
return callable_obj(*args, **kwargs)
return wrapper