Commit ed9efb35 authored by Aurélien Bompard's avatar Aurélien Bompard Committed by Barry Warsaw

Add a find method for pending requests

parent 6e99812b
......@@ -35,8 +35,8 @@ from mailman.database.transaction import dbconnection
from mailman.interfaces.pending import (
IPendable, IPended, IPendedKeyValue, IPendings)
from mailman.utilities.datetime import now
from sqlalchemy import Column, DateTime, ForeignKey, Integer, Unicode
from sqlalchemy.orm import relationship
from sqlalchemy import and_, Column, DateTime, ForeignKey, Integer, Unicode
from sqlalchemy.orm import aliased, relationship
from zope.interface import implementer
from zope.interface.verify import verifyObject
......@@ -156,6 +156,24 @@ class Pendings:
if pending.expiration_date < right_now:
def find(self, store, mlist=None, type=None):
query = store.query(Pended)
if mlist is not None:
pkv_alias_mlist = aliased(PendedKeyValue)
query = query.join(pkv_alias_mlist).filter(and_(
pkv_alias_mlist.key == 'list_id',
pkv_alias_mlist.value == json.dumps(mlist.list_id)
if type is not None:
pkv_alias_type = aliased(PendedKeyValue)
query = query.join(pkv_alias_type).filter(and_(
pkv_alias_type.key == 'type',
pkv_alias_type.value == json.dumps(type)
for pending in query:
yield pending.token, self.confirm(pending.token, expunge=False)
def __iter__(self, store):
for pending in store.query(Pended).all():
......@@ -24,11 +24,10 @@ __all__ = [
import unittest
from import create_list
from mailman.config import config
from import InvalidEmailAddressError
from mailman.interfaces.pending import (
IPendable, IPended, IPendedKeyValue, IPendings)
from mailman.model.pending import PendedKeyValue, Pended, Pendings
from mailman.interfaces.pending import IPendable, IPendings
from mailman.model.pending import PendedKeyValue
from mailman.testing.layers import ConfigLayer
from zope.component import getUtility
from zope.interface import implementer
......@@ -59,3 +58,25 @@ class TestPendings(unittest.TestCase):
pendable = pendingdb.confirm(token)
self.assertEqual(pendingdb.count, 0)
self.assertEqual(, 0)
def test_find(self):
# Test getting pendables for a mailing-list
mlist = create_list('[email protected]')
pendingdb = getUtility(IPendings)
subscription_1 = SimplePendable(
subscription_2 = SimplePendable(
subscription_3 = SimplePendable(
type='hold request',
token_1 = pendingdb.add(subscription_1)
self.assertEqual(pendingdb.count, 3)
mlist_pendings = list(pendingdb.find(mlist=mlist, type='subscription'))
self.assertEqual(len(mlist_pendings), 1)
self.assertEqual(mlist_pendings[0][0], token_1)
self.assertEqual(mlist_pendings[0][1]['list_id'], '')
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment