Permalink
Browse files

Merge commit 'jefftriplett/master' into refactor

Conflicts:

	django_proxy/signals.py
  • Loading branch information...
2 parents 4f83cda + bbecfe4 commit 37f5e24c4426aeec6fc20473d1213fafb6e5c489 @montylounge committed Jan 30, 2010
Showing with 380 additions and 96 deletions.
  1. +12 −10 django_proxy/models.py
  2. +107 −86 django_proxy/signals.py
  3. +32 −0 runtests.py
  4. 0 tests/__init__.py
  5. +14 −0 tests/base.py
  6. +93 −0 tests/models.py
  7. +116 −0 tests/test_models.py
  8. +6 −0 tests/test_settings.py
@@ -4,14 +4,16 @@
from datetime import datetime
from django_proxy.managers import PublicManager
+
class ProxyBase(models.Model):
- '''Represents the proxy objects. Retains the name, description,
- and any tags related to the associated object.
-
+ """
+ Represents the proxy objects. Retains the name, description, and any tags
+ related to the associated object.
+
A good use, I've found, for this type of object is aggregate view of all
content types in an RSS feed (post, links, tweets, etc.).
-
- '''
+
+ """
content_type = models.ForeignKey(ContentType)
object_id = models.PositiveIntegerField()
content_object = generic.GenericForeignKey()
@@ -21,19 +23,19 @@ class ProxyBase(models.Model):
description = models.TextField(blank=True, null=True)
tags = models.CharField(max_length=255, blank=True, null=True)
pub_date = models.DateTimeField(default=datetime.now)
-
+
#audit fields
created_on = models.DateTimeField(default=datetime.now)
updated_on = models.DateTimeField(default=datetime.now)
-
+
objects = PublicManager()
-
+
def __unicode__(self):
return '%s' % self.title
-
+
class Meta:
abstract = True
-
+
def save(self, *args, **kwargs):
self.updated_on = datetime.now()
super(ProxyBase, self).save(*args, **kwargs)
@@ -1,105 +1,126 @@
-from django.db.models import get_model
+import datetime
+
from django.contrib.contenttypes.models import ContentType
+from django.db.models import get_model
-def proxy_save(sender, **kwargs):
- ''' Handles the save/update of a Proxy instance. '''
- instance = kwargs['instance']
- created = kwargs['created']
+class DjangoProxy(object):
+ _required_fields = ['title', 'description']
+ _field_mappings = [
+ ('title', None),
+ ('description', None),
+ ('pub_date', datetime.datetime.now()),
+ ('tags', None),
+ ]
+ content_object = None
+ proxy_model = None
- cls = instance.ProxyMeta
- model_str = getattr(cls, 'model_str', 'django_proxy.proxy')
- model = get_model(*model_str.split('.'))
- obj = model()
- active = False
+ def __init__(self, instance, created=None):
+ self.content_object = instance
+ model = self._get_proxy_model(instance)
+ self.proxy_model = model()
- if created:
- obj.content_object = instance
- else:
try:
ctype = ContentType.objects.get_for_model(instance)
- obj = model._default_manager.get(object_id=instance.id, content_type=ctype)
+ self.proxy_model = model._default_manager.get(object_id=instance.id, content_type=ctype)
+ except model.DoesNotExist:
+ self.proxy_model.content_object = instance
+
+ def _get_attr(self, attr, obj):
+ if hasattr(self.content_object.ProxyMeta, attr):
+ value = getattr(self.content_object, getattr(obj, attr))
+ if callable(value):
+ value = value()
+ return value
+
+ def _get_proxy_model(self, instance):
+ model_str = getattr(instance.ProxyMeta, 'model_str', 'django_proxy.proxy')
+ model = get_model(*model_str.split('.'))
+ return model
+
+ def _validate(self):
+ missing = []
+ for field in self._required_fields:
+ if not getattr(self.content_object.ProxyMeta, field, None):
+ missing.append(field)
+ if len(missing):
+ raise ValueError('Missing required fields: %s' % (', '.join(missing)))
+
+ def delete(self):
+ """
+ Remove any remaining child/associated Proxy records.
+
+ """
+ model = self._get_proxy_model(self.content_object)
+ ctype = ContentType.objects.get_for_model(self.content_object)
+ try:
+ self.proxy_model = model._default_manager.get(object_id=self.content_object.id, content_type=ctype)
+ self.proxy_model.delete()
except model.DoesNotExist:
- obj = model()
- obj.content_object = instance
-
- if hasattr(cls, 'active'):
- if isinstance(cls.active, basestring):
- active_field = getattr(instance, cls.active, None)
- if callable(active_field):
- active = active_field()
- else:
+ pass
+
+ def update(self):
+ """
+ Updates the status of the ProxyModel to either show or or get deleted
+ depending on if the object is active.
+
+ """
+ self._validate()
+ active = self.get_active()
+
+ if active:
+ for mapping in self._field_mappings:
+ setattr(self.proxy_model, mapping[0], self._get_attr(mapping[0], self.content_object.ProxyMeta) or mapping[1])
+ self.proxy_model.active = active
+ self.proxy_model.save()
+ elif self.proxy_model.id:
+ self.proxy_model.delete()
+
+ def get_active(self):
+ active = False
+ if hasattr(self.content_object.ProxyMeta, 'active'):
+ if isinstance(self.content_object.ProxyMeta.active, basestring):
+ active_field = getattr(self.content_object, self.content_object.ProxyMeta.active)
+ if callable(active_field):
+ active_field = active_field()
active = active_field
- else:
- try:
- active_field = cls.active
- objfield = active_field.keys()[0]
- active_status = active_field.values()[0]
- actual_status = getattr(instance, objfield, None)
- if active_status == actual_status:
- active = True
- except Exception:
- # deal with this better...
- pass
- else:
- active = True
-
- if not active and obj.id:
- obj.delete()
- return
-
- if hasattr(cls, 'title'):
- title = getattr(instance, cls.title, None)
- if callable(title):
- obj.title = title()
- else:
- obj.title = title
- else:
- raise Exception('Missing title field')
-
- if hasattr(cls, 'description'):
- description = getattr(instance, cls.description, None)
- if callable(description):
- obj.description = description()
- else:
- obj.description = description
- else:
- raise Exception('Missing description field')
-
- #proxy pub_date isn't required so confirm
- if hasattr(cls, 'pub_date'):
- pub_date = getattr(instance, cls.pub_date, None)
- if callable(pub_date):
- obj.pub_date = pub_date()
- else:
- obj.pub_date = pub_date
- #proxy tag isn't require so confirm
- if hasattr(cls, 'tags'):
- tags = getattr(instance, cls.tags, None)
- if callable(tags):
- obj.tags = tags()
+ elif isinstance(self.content_object.ProxyMeta.active, dict):
+ try:
+ active_field = self.content_object.ProxyMeta.active
+ objfield = active_field.keys()[0]
+ active_status = active_field.values()[0]
+ actual_status = getattr(self.content_object, objfield)
+ if active_status == actual_status:
+ active = True
+ except Exception:
+ pass
else:
- obj.tags = tags
+ active = True
+
+ return active
+
+
+def proxy_save(sender, **kwargs):
+ """
+ Handles the save/update of a Proxy instance.
- if active:
- obj.save()
+ """
+ instance = kwargs['instance']
+ created = kwargs['created']
+
+ dp = DjangoProxy(instance, created)
+ dp.update()
def proxy_delete(sender, **kwargs):
- '''Responsible for handling the deletion of any child/associated Proxy records.
+ """
+ Responsible for handling the deletion of any child/associated Proxy records.
Coupled to associated object's post_delete signal.
- '''
+ """
instance = kwargs['instance']
- ctype = ContentType.objects.get_for_model(instance)
-
- cls = instance.ProxyMeta
- model_str = getattr(cls, 'model_str', 'django_proxy.proxy')
- model = get_model(*model_str.split('.'))
- try:
- obj = model._default_manager.get(object_id=instance.id, content_type=ctype)
- obj.delete()
- except DoesNotExist:
- pass
+
+ dp = DjangoProxy(instance)
+ dp.delete()
View
@@ -0,0 +1,32 @@
+#!/usr/bin/env python
+
+import os
+import unittest
+
+if not 'DJANGO_SETTINGS_MODULE' in os.environ:
+ os.environ['DJANGO_SETTINGS_MODULE'] = 'tests.test_settings'
+
+from django.conf import settings
+from django.db import connection
+from django.test.utils import setup_test_environment
+from django.test.utils import teardown_test_environment
+from django.utils.importlib import import_module
+
+TEST_MODULES = ["tests.base"]
+
+
+def run_tests():
+ setup_test_environment()
+ db_name = settings.DATABASE_NAME
+ suite = unittest.TestSuite()
+ for module_name in TEST_MODULES:
+ module = import_module(module_name)
+ suite.addTest(unittest.defaultTestLoader.loadTestsFromModule(module))
+ connection.creation.create_test_db()
+ unittest.TextTestRunner().run(suite)
+ connection.creation.destroy_test_db(db_name)
+ teardown_test_environment()
+
+
+if __name__ == "__main__":
+ run_tests()
View
No changes.
View
@@ -0,0 +1,14 @@
+import datetime
+import unittest
+
+from django.core import signals
+from django.db import models
+from django.db.models import signals
+from django.test import TestCase
+from django.utils.translation import ugettext_lazy as _
+
+from django_proxy.models import Proxy
+from django_proxy.signals import proxy_save, proxy_delete
+
+from tests.models import *
+from tests.test_models import *
Oops, something went wrong.

0 comments on commit 37f5e24

Please sign in to comment.