# The key *has* to exist on sql.
return getattr(self, key)
- def save(self):
+ def save(self, commit=True):
sess = object_session(self)
if sess is None:
sess = Session()
sess.add(self)
- sess.commit()
+ if commit:
+ sess.commit()
+ else:
+ sess.flush()
def delete(self, commit=True):
"""Delete the object and commit the change immediately by default"""
from sqlalchemy import Column, Integer, Unicode, UnicodeText, DateTime, \
Boolean, ForeignKey, UniqueConstraint, PrimaryKeyConstraint, \
SmallInteger, Date
-from sqlalchemy.orm import relationship, backref, with_polymorphic
+from sqlalchemy.orm import relationship, backref, with_polymorphic, validates
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy.sql.expression import desc
from sqlalchemy.ext.associationproxy import association_proxy
if key is None:
raise ValueError("Invalid type of object given")
- # We need to save so that self.id is populated
self.type = key
- self.save()
- # First set self as activity
+ # We need to populate the self.id so we need to save but, we don't
+ # want to save this AI in the database (yet) so commit=False.
+ self.save(commit=False)
obj.activity = self.id
obj.save()
model = self.TYPES[self.type]
return model.query.filter_by(activity=self.id).first()
- def save(self, *args, **kwargs):
- if self.type not in self.TYPES.keys():
- raise ValueError("Invalid type set")
- Base.save(self, *args, **kwargs)
+ @validates("type")
+ def validate_type(self, key, value):
+ """ Validate that the type set is a valid type """
+ assert value in self.TYPES
+ return value
class Activity(Base, ActivityMixin):
"""
from __future__ import print_function
from mediagoblin.db.base import Session
-from mediagoblin.db.models import MediaEntry, User, Privilege
+from mediagoblin.db.models import MediaEntry, User, Privilege, Activity, \
+ Generator
from mediagoblin.tests import MGClientTestCase
-from mediagoblin.tests.tools import fixture_add_user
+from mediagoblin.tests.tools import fixture_add_user, fixture_media_entry, \
+ fixture_add_activity
try:
import mock
self.user(u'lindsay').url_for_self(fake_urlgen())
assert excinfo.errisinstance(TypeError)
assert 'object is not callable' in str(excinfo)
+
+class TestActivitySetGet(object):
+ """ Test methods on the Activity and ActivityIntermediator models """
+
+ @pytest.fixture(autouse=True)
+ def setup(self, test_app):
+ self.app = test_app
+ self.user = fixture_add_user()
+ self.obj = fixture_media_entry()
+ self.target = fixture_media_entry()
+
+ def test_set_activity_object(self):
+ """ Activity.set_object should produce ActivityIntermediator """
+ # The fixture will set self.obj as the object on the activity.
+ activity = fixture_add_activity(self.obj, actor=self.user)
+
+ # Assert the media has been associated with an AI
+ assert self.obj.activity is not None
+
+ # Assert the AI on the media and object are the same
+ assert activity.object == self.obj.activity
+
+ def test_activity_set_target(self):
+ """ Activity.set_target should produce ActivityIntermediator """
+ # This should set everything needed on the target
+ activity = fixture_add_activity(self.obj, actor=self.user)
+ activity.set_target(self.target)
+
+ # Assert the media has been associated with the AI
+ assert self.target.activity is not None
+
+ # assert the AI on the media and target are the same
+ assert activity.target == self.target.activity
+
+ def test_get_activity_object(self):
+ """ Activity.get_object should return a set object """
+ activity = fixture_add_activity(self.obj, actor=self.user)
+
+ print("self.obj.activity = {0}".format(self.obj.activity))
+
+ # check we now can get the object
+ assert activity.get_object is not None
+ assert activity.get_object.id == self.obj.id
+
+ def test_get_activity_target(self):
+ """ Activity.set_target should return a set target """
+ activity = fixture_add_activity(self.obj, actor=self.user)
+ activity.set_target(self.target)
+
+ # check we can get the target
+ assert activity.get_target is not None
+ assert activity.get_target.id == self.target.id
from mediagoblin import mg_globals
from mediagoblin.db.models import User, MediaEntry, Collection, MediaComment, \
CommentSubscription, CommentNotification, Privilege, CommentReport, Client, \
- RequestToken, AccessToken
+ RequestToken, AccessToken, Activity, Generator
from mediagoblin.tools import testing
from mediagoblin.init.config import read_mediagoblin_config
from mediagoblin.db.base import Session
Session.expunge(comment_report)
return comment_report
+
+def fixture_add_activity(obj, verb="post", target=None, generator=None, actor=None):
+ if generator is None:
+ generator = Generator(
+ name="GNU MediaGoblin",
+ object_type="service"
+ )
+ generator.save()
+
+ if actor is None:
+ actor = fixture_add_user()
+
+ activity = Activity(
+ verb=verb,
+ actor=actor.id,
+ generator=generator.id,
+ )
+
+ activity.set_object(obj)
+
+ if target is not None:
+ activity.set_target(target)
+
+ activity.save()
+ return activity
\ No newline at end of file
projects / mediagoblin.git / commitdiff