Commits

David Wolever committed 071cee4

Adding a proper enqueue method and a dequeue instancemethod

Comments (0)

Files changed (3)

     class Meta:
         unique_together = (('queue_name', 'url'), )
 
+    def __init__(self, *args, **kwargs):
+        super(QueuedUrl, self).__init__(*args, **kwargs)
+        self.dequeue = self._dequeue_instancemethod
+
+    @classmethod
+    def enqueue(cls, queue_name, url, title):
+        """ Put a url onto the end of a queue. """
+        qrl, _ = QueuedUrl.objects.get_or_create(queue_name=queue_name, url=url)
+        qrl.title = title
+        qrl.dequeued = None
+        qrl.enqueued = datetime.now()
+        qrl.save()
+        return qrl
+
     @classmethod
     def dequeue(cls, queue_name):
         """ Dequeue the next URL in the given queue and mark it dequeued.
             If the queue is empty, None is returned.. """
         url = cls.objects_in_queue(queue_name)[:1]
         if len(url) > 0:
-            url[0].dequeued = datetime.now()
-            url[0].save()
+            url[0].dequeue()
             return url[0]
         return None
 
+    def _dequeue_instancemethod(self):
+        """ Dequeue this QueuedUrl. """
+        self.dequeued = datetime.now()
+        self.save()
+
     @classmethod
     def objects_in_queue(cls, queue_name):
         """ Get a QuerySet matching all the queued items the given queue,
         # Creating a new queue with the same values should be an error
         assert_raises(IntegrityError, new_queued_url, **queue_props)
 
+    def test_enqueue(self):
+        queue, url, title = ("queue", "http://url", "title")
+
+        def sanity_checks(qrl):
+            assert_equal(QueuedUrl.objects.all().count(), 1)
+            assert_equal(qrl.queue_name, queue)
+            assert_equal(qrl.url, url)
+            assert_equal(qrl.title, title)
+
+        qrl = QueuedUrl.enqueue(queue, url, title)
+
+        # The enqueue method is responsible for saving the qrl it returns
+        assert_equal(QueuedUrl.objects.all()[0], qrl)
+        sanity_checks(qrl)
+
+        # Enqueueing the same link with a different title should cause an update
+        title = "new title"
+        qrl = QueuedUrl.enqueue(queue, url, title)
+        sanity_checks(qrl)
+
+        # Dequeueing and enqueueing a url should result it it being queued
+        old_queue_position = qrl.enqueued
+        qrl.dequeue()
+
+        # Enqueue the same thing - qrl should be queued again, but its position
+        # should be greater than its old position (ie, it's at the end of the queue)
+        qrl = QueuedUrl.enqueue(queue, url, title)
+        assert_equal(qrl.dequeued, None)
+        assert qrl.enqueued > old_queue_position
+        sanity_checks(qrl)
+
 
 class TestViews(TestCase):
     
     def test_queue_enqueue_post(self):
         c = Client()
 
+        def assert_enqueue(data):
+            resp = c.post('/queue/some_queue/enqueue', data)
+            assert_equal(resp['Location'],
+                        'http://testserver/queue/some_queue/enqueue?urlsaved=True')
+            assert_equal(resp.status_code, 302)
+
         # Check that the data can be POSTed
         data = {
             'url': 'http://new',
             'title': 'new url',
         }
-        resp = c.post('/queue/some_queue/enqueue', data)
-        assert_equal(resp['Location'],
-                    'http://testserver/queue/some_queue/enqueue?urlsaved=True')
-        assert_equal(resp.status_code, 302)
+        assert_enqueue(data)
 
-        # And now check that the URL was actually saved
-        url = QueuedUrl.dequeue("some_queue")
-        assert_not_equal(url, None)
-        assert_equal(url.url, data['url'])
-        assert_equal(url.title, data['title'])
+        # And now check that the QRL was actually saved
+        qrl = QueuedUrl.dequeue("some_queue")
+        assert_not_equal(qrl, None)
+        assert_equal(qrl.url, data['url'])
+        assert_equal(qrl.title, data['title'])
+        assert qrl.dequeued
+
+        # Now post the same URL, but with a differnt title, and make
+        # sure the title has been updated and it is queued again
+        data = {
+            'url': 'http://new',
+            'title': 'new title',
+        }
+        assert_enqueue(data)
+
+        # There should only be one QRL
+        assert_equal(QueuedUrl.objects.all().count(), 1)
+        qrl = QueuedUrl.objects.all()[0]
+        assert_equal(qrl.title, 'new title')
+        assert_equal(qrl.dequeued, None)
 
     def test_queue_enqueue_get(self):
         c = Client()
 @assert_method("GET", "POST")
 def queue_enqueue(request, queue_name):
     if request.method == "POST":
-        new_url = QueuedUrl()
-        new_url.queue_name = queue_name
-        new_url.url = request.POST['url']
-        new_url.title = request.POST['title']
-        new_url.save()
+        (url, title) = (request.POST['url'], request.POST['title'])
+        QueuedUrl.enqueue(queue_name, url, title)
         return redirect(request.build_absolute_uri() + "?urlsaved=True")
 
     data = {
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.