![[tokkee]](http://tokkee.org/images/avatar.png){kind=avatar}
diff --git a/test/db_test_base.py b/test/db_test_base.py
index 20bdafbab6dd08c3a759903ff59b90b3f5191ce8..ea468e749bc9221be88e5b2a55fa9bdbd1ba32d0 100644 (file)
--- a/test/db_test_base.py
+++ b/test/db_test_base.py
#
# $Id: db_test_base.py,v 1.101 2008-08-19 01:40:59 richard Exp $
-import unittest, os, shutil, errno, imp, sys, time, pprint, sets, base64, os.path
+import unittest, os, shutil, errno, imp, sys, time, pprint, base64, os.path
+import gpgmelib
+# Python 2.3 ... 2.6 compatibility:
+from roundup.anypy.sets_ import set
+from email.parser import FeedParser
from roundup.hyperdb import String, Password, Link, Multilink, Date, \
Interval, DatabaseError, Boolean, Number, Node
from roundup.mailer import Mailer
-from roundup import date, password, init, instance, configuration, support
+from roundup import date, password, init, instance, configuration, \
+ roundupdb, i18n
from mocknull import MockNull
config.RDBMS_HOST = "localhost"
config.RDBMS_USER = "rounduptest"
config.RDBMS_PASSWORD = "rounduptest"
+config.RDBMS_TEMPLATE = "template0"
#config.logging = MockNull()
# these TRACKER_WEB and MAIL_DOMAIN values are used in mailgw tests
config.MAIL_DOMAIN = "your.tracker.email.domain.example"
except OSError, error:
if error.errno not in (errno.ENOENT, errno.ESRCH): raise
# create the instance
- init.install(dirname, os.path.join(os.path.dirname(__file__), '..',
- 'templates/classic'))
+ init.install(dirname, os.path.join(os.path.dirname(__file__),
+ '..',
+ 'share',
+ 'roundup',
+ 'templates',
+ 'classic'))
init.write_select_db(dirname, backend)
config.save(os.path.join(dirname, 'config.ini'))
tracker = instance.open(dirname)
priority.create(name="bug", order="1")
db.commit()
+ # nosy tests require this
+ db.security.addPermissionToRole('User', 'View', 'msg')
+
class MyTestCase(unittest.TestCase):
def tearDown(self):
if hasattr(self, 'db'):
if os.path.exists(config.DATABASE):
shutil.rmtree(config.DATABASE)
+ def open_database(self):
+ self.db = self.module.Database(config, 'admin')
+
+
if os.environ.has_key('LOGGING_LEVEL'):
from roundup import rlog
config.logging = rlog.BasicLogging()
config.logging.setLevel(os.environ['LOGGING_LEVEL'])
- config.logging.getLogger('hyperdb').setFormat('%(message)s')
+ config.logging.getLogger('roundup.hyperdb').setFormat('%(message)s')
-class DBTest(MyTestCase):
+class commonDBTest(MyTestCase):
def setUp(self):
# remove previous test, ignore errors
if os.path.exists(config.DATABASE):
shutil.rmtree(config.DATABASE)
os.makedirs(config.DATABASE + '/files')
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
setupSchema(self.db, 1, self.module)
+ def iterSetup(self, classname='issue'):
+ cls = getattr(self.db, classname)
+ def filt_iter(*args):
+ """ for checking equivalence of filter and filter_iter """
+ return list(cls.filter_iter(*args))
+ return self.assertEqual, cls.filter, filt_iter
+
+ def filteringSetupTransitiveSearch(self, classname='issue'):
+ u_m = {}
+ k = 30
+ for user in (
+ {'username': 'ceo', 'age': 129},
+ {'username': 'grouplead1', 'age': 29, 'supervisor': '3'},
+ {'username': 'grouplead2', 'age': 29, 'supervisor': '3'},
+ {'username': 'worker1', 'age': 25, 'supervisor' : '4'},
+ {'username': 'worker2', 'age': 24, 'supervisor' : '4'},
+ {'username': 'worker3', 'age': 23, 'supervisor' : '5'},
+ {'username': 'worker4', 'age': 22, 'supervisor' : '5'},
+ {'username': 'worker5', 'age': 21, 'supervisor' : '5'}):
+ u = self.db.user.create(**user)
+ u_m [u] = self.db.msg.create(author = u, content = ' '
+ , date = date.Date ('2006-01-%s' % k))
+ k -= 1
+ i = date.Interval('-1d')
+ for issue in (
+ {'title': 'ts1', 'status': '2', 'assignedto': '6',
+ 'priority': '3', 'messages' : [u_m ['6']], 'nosy' : ['4']},
+ {'title': 'ts2', 'status': '1', 'assignedto': '6',
+ 'priority': '3', 'messages' : [u_m ['6']], 'nosy' : ['5']},
+ {'title': 'ts4', 'status': '2', 'assignedto': '7',
+ 'priority': '3', 'messages' : [u_m ['7']]},
+ {'title': 'ts5', 'status': '1', 'assignedto': '8',
+ 'priority': '3', 'messages' : [u_m ['8']]},
+ {'title': 'ts6', 'status': '2', 'assignedto': '9',
+ 'priority': '3', 'messages' : [u_m ['9']]},
+ {'title': 'ts7', 'status': '1', 'assignedto': '10',
+ 'priority': '3', 'messages' : [u_m ['10']]},
+ {'title': 'ts8', 'status': '2', 'assignedto': '10',
+ 'priority': '3', 'messages' : [u_m ['10']], 'foo' : i},
+ {'title': 'ts9', 'status': '1', 'assignedto': '10',
+ 'priority': '3', 'messages' : [u_m ['10'], u_m ['9']]}):
+ self.db.issue.create(**issue)
+ return self.iterSetup(classname)
+
+
+class DBTest(commonDBTest):
+
def testRefresh(self):
self.db.refresh_database()
def testCreatorProperty(self):
i = self.db.issue
id1 = i.create(title='spam')
- self.db.commit()
- self.db.close()
- self.db = self.module.Database(config, 'fred')
- setupSchema(self.db, 0, self.module)
- i = self.db.issue
+ self.db.journaltag = 'fred'
id2 = i.create(title='spam')
self.assertNotEqual(id1, id2)
self.assertNotEqual(i.get(id1, 'creator'), i.get(id2, 'creator'))
def testActorProperty(self):
i = self.db.issue
id1 = i.create(title='spam')
- self.db.commit()
- self.db.close()
- self.db = self.module.Database(config, 'fred')
- setupSchema(self.db, 0, self.module)
- i = self.db.issue
+ self.db.journaltag = 'fred'
i.set(id1, title='asfasd')
self.assertNotEqual(i.get(id1, 'creator'), i.get(id1, 'actor'))
if commit: self.db.commit()
self.assertEqual(self.db.issue.get(nid, "nosy"), [])
+ def testMakeSeveralMultilinkedNodes(self):
+ for commit in (0,1):
+ u1 = self.db.user.create(username='foo%s'%commit)
+ u2 = self.db.user.create(username='bar%s'%commit)
+ u3 = self.db.user.create(username='baz%s'%commit)
+ nid = self.db.issue.create(title="spam", nosy=[u1])
+ if commit: self.db.commit()
+ self.assertEqual(self.db.issue.get(nid, "nosy"), [u1])
+ self.db.issue.set(nid, deadline=date.Date('.'))
+ self.db.issue.set(nid, nosy=[u1,u2], title='ta%s'%commit)
+ if commit: self.db.commit()
+ self.assertEqual(self.db.issue.get(nid, "nosy"), [u1,u2])
+ self.db.issue.set(nid, deadline=date.Date('.'))
+ self.db.issue.set(nid, nosy=[u1,u2,u3], title='tb%s'%commit)
+ if commit: self.db.commit()
+ self.assertEqual(self.db.issue.get(nid, "nosy"), [u1,u2,u3])
+
def testMultilinkChangeIterable(self):
for commit in (0,1):
# invalid nosy value assertion
# try a couple of the built-in iterable types to make
# sure that we accept them and handle them properly
# try a set as input for the multilink
- nid = self.db.issue.create(title="spam", nosy=sets.Set(u1))
+ nid = self.db.issue.create(title="spam", nosy=set(u1))
if commit: self.db.commit()
self.assertEqual(self.db.issue.get(nid, "nosy"), [u1])
self.assertRaises(TypeError, self.db.issue.set, nid,
if commit: self.db.commit()
self.assertEqual(self.db.issue.get(nid, "nosy"), [])
# make sure we accept a frozen set
- self.db.issue.set(nid, nosy=sets.Set([u1,u2]))
+ self.db.issue.set(nid, nosy=set([u1,u2]))
if commit: self.db.commit()
l = [u1,u2]; l.sort()
m = self.db.issue.get(nid, "nosy"); m.sort()
'2008-02-29.00:00:00')
self.assertEquals(self.db.issue.filter(None,
{'deadline': '2008-02-29'}), [nid])
+ self.assertEquals(list(self.db.issue.filter_iter(None,
+ {'deadline': '2008-02-29'})), [nid])
self.db.issue.set(nid, deadline=date.Date('2008-03-01'))
self.assertEquals(str(self.db.issue.get(nid, 'deadline')),
'2008-03-01.00:00:00')
self.assertEquals(self.db.issue.filter(None,
{'deadline': '2008-02-29'}), [])
+ self.assertEquals(list(self.db.issue.filter_iter(None,
+ {'deadline': '2008-02-29'})), [])
def testDateUnset(self):
for commit in (0,1):
others = nodeids[:]
others.remove('1')
- self.assertEqual(sets.Set(self.db.status.getnodeids()),
- sets.Set(nodeids))
- self.assertEqual(sets.Set(self.db.status.getnodeids(retired=True)),
- sets.Set(['1']))
- self.assertEqual(sets.Set(self.db.status.getnodeids(retired=False)),
- sets.Set(others))
+ self.assertEqual(set(self.db.status.getnodeids()),
+ set(nodeids))
+ self.assertEqual(set(self.db.status.getnodeids(retired=True)),
+ set(['1']))
+ self.assertEqual(set(self.db.status.getnodeids(retired=False)),
+ set(others))
self.assert_(self.db.status.is_retired('1'))
self.assertEquals(self.db.indexer.search([], self.db.issue), {})
self.assertEquals(self.db.indexer.search(['hello'], self.db.issue),
{i1: {'files': [f1]}})
+ # content='world' has the wrong content-type and shouldn't be indexed
self.assertEquals(self.db.indexer.search(['world'], self.db.issue), {})
self.assertEquals(self.db.indexer.search(['frooz'], self.db.issue),
{i2: {}})
self.assertEquals(self.db.indexer.search(['flebble'], self.db.issue),
{'1': {}})
- def testIndexingOnImport(self):
- # import a message
- msgcontent = 'Glrk'
- msgid = self.db.msg.import_list(['content', 'files', 'recipients'],
- [repr(msgcontent), '[]', '[]'])
- msg_filename = self.db.filename(self.db.msg.classname, msgid,
- create=1)
- support.ensureParentsExist(msg_filename)
- msg_file = open(msg_filename, 'w')
- msg_file.write(msgcontent)
- msg_file.close()
-
- # import a file
- filecontent = 'Brrk'
- fileid = self.db.file.import_list(['content'], [repr(filecontent)])
- file_filename = self.db.filename(self.db.file.classname, fileid,
- create=1)
- support.ensureParentsExist(file_filename)
- file_file = open(file_filename, 'w')
- file_file.write(filecontent)
- file_file.close()
-
+ def testIndexingPropertiesOnImport(self):
# import an issue
title = 'Bzzt'
nodeid = self.db.issue.import_list(['title', 'messages', 'files',
- 'spam', 'nosy', 'superseder'], [repr(title), repr([msgid]),
- repr([fileid]), '[]', '[]', '[]'])
+ 'spam', 'nosy', 'superseder'], [repr(title), '[]', '[]',
+ '[]', '[]', '[]'])
self.db.commit()
# Content of title attribute is indexed
self.assertEquals(self.db.indexer.search([title], self.db.issue),
{str(nodeid):{}})
- # Content of message is indexed
- self.assertEquals(self.db.indexer.search([msgcontent], self.db.issue),
- {str(nodeid):{'messages':[str(msgid)]}})
- # Content of file is indexed
- self.assertEquals(self.db.indexer.search([filecontent], self.db.issue),
- {str(nodeid):{'files':[str(fileid)]}})
-
#
self.db.issue.retire(ids[0])
self.assertEqual(len(self.db.issue.stringFind(title='spam')), 1)
- def filteringSetup(self):
+ def filteringSetup(self, classname='issue'):
for user in (
- {'username': 'bleep', 'age': 1},
- {'username': 'blop', 'age': 1.5},
- {'username': 'blorp', 'age': 2}):
+ {'username': 'bleep', 'age': 1, 'assignable': True},
+ {'username': 'blop', 'age': 1.5, 'assignable': True},
+ {'username': 'blorp', 'age': 2, 'assignable': False}):
self.db.user.create(**user)
- iss = self.db.issue
file_content = ''.join([chr(i) for i in range(255)])
f = self.db.file.create(content=file_content)
for issue in (
'files': [f]}):
self.db.issue.create(**issue)
self.db.commit()
- return self.assertEqual, self.db.issue.filter
+ return self.iterSetup(classname)
def testFilteringID(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'id': '1'}, ('+','id'), (None,None)), ['1'])
- ae(filt(None, {'id': '2'}, ('+','id'), (None,None)), ['2'])
- ae(filt(None, {'id': '100'}, ('+','id'), (None,None)), [])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'id': '1'}, ('+','id'), (None,None)), ['1'])
+ ae(filt(None, {'id': '2'}, ('+','id'), (None,None)), ['2'])
+ ae(filt(None, {'id': '100'}, ('+','id'), (None,None)), [])
+
+ def testFilteringBoolean(self):
+ ae, filter, filter_iter = self.filteringSetup('user')
+ a = 'assignable'
+ for filt in filter, filter_iter:
+ ae(filt(None, {a: '1'}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: '0'}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: ['1']}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: ['0']}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: ['0','1']}, ('+','id'), (None,None)),
+ ['3','4','5'])
+ ae(filt(None, {a: 'True'}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: 'False'}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: ['True']}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: ['False']}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: ['False','True']}, ('+','id'), (None,None)),
+ ['3','4','5'])
+ ae(filt(None, {a: True}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: False}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: 1}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: 0}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: [1]}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: [0]}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: [0,1]}, ('+','id'), (None,None)), ['3','4','5'])
+ ae(filt(None, {a: [True]}, ('+','id'), (None,None)), ['3','4'])
+ ae(filt(None, {a: [False]}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {a: [False,True]}, ('+','id'), (None,None)),
+ ['3','4','5'])
def testFilteringNumber(self):
- self.filteringSetup()
- ae, filt = self.assertEqual, self.db.user.filter
- ae(filt(None, {'age': '1'}, ('+','id'), (None,None)), ['3'])
- ae(filt(None, {'age': '1.5'}, ('+','id'), (None,None)), ['4'])
- ae(filt(None, {'age': '2'}, ('+','id'), (None,None)), ['5'])
- ae(filt(None, {'age': ['1','2']}, ('+','id'), (None,None)), ['3','5'])
+ ae, filter, filter_iter = self.filteringSetup('user')
+ for filt in filter, filter_iter:
+ ae(filt(None, {'age': '1'}, ('+','id'), (None,None)), ['3'])
+ ae(filt(None, {'age': '1.5'}, ('+','id'), (None,None)), ['4'])
+ ae(filt(None, {'age': '2'}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {'age': ['1','2']}, ('+','id'), (None,None)),
+ ['3','5'])
+ ae(filt(None, {'age': 2}, ('+','id'), (None,None)), ['5'])
+ ae(filt(None, {'age': [1,2]}, ('+','id'), (None,None)), ['3','5'])
def testFilteringString(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'title': ['one']}, ('+','id'), (None,None)), ['1'])
- ae(filt(None, {'title': ['issue one']}, ('+','id'), (None,None)),
- ['1'])
- ae(filt(None, {'title': ['issue', 'one']}, ('+','id'), (None,None)),
- ['1'])
- ae(filt(None, {'title': ['issue']}, ('+','id'), (None,None)),
- ['1','2','3'])
- ae(filt(None, {'title': ['one', 'two']}, ('+','id'), (None,None)),
- [])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'title': ['one']}, ('+','id'), (None,None)), ['1'])
+ ae(filt(None, {'title': ['issue one']}, ('+','id'), (None,None)),
+ ['1'])
+ ae(filt(None, {'title': ['issue', 'one']}, ('+','id'), (None,None)),
+ ['1'])
+ ae(filt(None, {'title': ['issue']}, ('+','id'), (None,None)),
+ ['1','2','3'])
+ ae(filt(None, {'title': ['one', 'two']}, ('+','id'), (None,None)),
+ [])
def testFilteringLink(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'status': '1'}, ('+','id'), (None,None)), ['2','3'])
- ae(filt(None, {'assignedto': '-1'}, ('+','id'), (None,None)), ['3','4'])
- ae(filt(None, {'assignedto': None}, ('+','id'), (None,None)), ['3','4'])
- ae(filt(None, {'assignedto': [None]}, ('+','id'), (None,None)),
- ['3','4'])
- ae(filt(None, {'assignedto': ['-1', None]}, ('+','id'), (None,None)),
- ['3','4'])
- ae(filt(None, {'assignedto': ['1', None]}, ('+','id'), (None,None)),
- ['1', '3','4'])
+ ae, filter, filter_iter = self.filteringSetup()
+ a = 'assignedto'
+ grp = (None, None)
+ for filt in filter, filter_iter:
+ ae(filt(None, {'status': '1'}, ('+','id'), grp), ['2','3'])
+ ae(filt(None, {a: '-1'}, ('+','id'), grp), ['3','4'])
+ ae(filt(None, {a: None}, ('+','id'), grp), ['3','4'])
+ ae(filt(None, {a: [None]}, ('+','id'), grp), ['3','4'])
+ ae(filt(None, {a: ['-1', None]}, ('+','id'), grp), ['3','4'])
+ ae(filt(None, {a: ['1', None]}, ('+','id'), grp), ['1', '3','4'])
def testFilteringMultilinkAndGroup(self):
"""testFilteringMultilinkAndGroup:
See roundup Bug 1541128: apparently grouping by something and
searching a Multilink failed with MySQL 5.0
"""
- ae, filt = self.filteringSetup()
- ae(filt(None, {'files': '1'}, ('-','activity'), ('+','status')), ['4'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for f in filter, filter_iter:
+ ae(f(None, {'files': '1'}, ('-','activity'), ('+','status')), ['4'])
def testFilteringRetired(self):
- ae, filt = self.filteringSetup()
+ ae, filter, filter_iter = self.filteringSetup()
self.db.issue.retire('2')
- ae(filt(None, {'status': '1'}, ('+','id'), (None,None)), ['3'])
+ for f in filter, filter_iter:
+ ae(f(None, {'status': '1'}, ('+','id'), (None,None)), ['3'])
def testFilteringMultilink(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'nosy': '3'}, ('+','id'), (None,None)), ['4'])
- ae(filt(None, {'nosy': '-1'}, ('+','id'), (None,None)), ['1', '2'])
- ae(filt(None, {'nosy': ['1','2']}, ('+', 'status'),
- ('-', 'deadline')), ['4', '3'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'nosy': '3'}, ('+','id'), (None,None)), ['4'])
+ ae(filt(None, {'nosy': '-1'}, ('+','id'), (None,None)), ['1', '2'])
+ ae(filt(None, {'nosy': ['1','2']}, ('+', 'status'),
+ ('-', 'deadline')), ['4', '3'])
def testFilteringMany(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'nosy': '2', 'status': '1'}, ('+','id'), (None,None)),
- ['3'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for f in filter, filter_iter:
+ ae(f(None, {'nosy': '2', 'status': '1'}, ('+','id'), (None,None)),
+ ['3'])
def testFilteringRangeBasic(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'deadline': 'from 2003-02-10 to 2003-02-23'}), ['1','3'])
- ae(filt(None, {'deadline': '2003-02-10; 2003-02-23'}), ['1','3'])
- ae(filt(None, {'deadline': '; 2003-02-16'}), ['2'])
+ ae, filter, filter_iter = self.filteringSetup()
+ d = 'deadline'
+ for f in filter, filter_iter:
+ ae(f(None, {d: 'from 2003-02-10 to 2003-02-23'}), ['1','3'])
+ ae(f(None, {d: '2003-02-10; 2003-02-23'}), ['1','3'])
+ ae(f(None, {d: '; 2003-02-16'}), ['2'])
def testFilteringRangeTwoSyntaxes(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'deadline': 'from 2003-02-16'}), ['1', '3', '4'])
- ae(filt(None, {'deadline': '2003-02-16;'}), ['1', '3', '4'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'deadline': 'from 2003-02-16'}), ['1', '3', '4'])
+ ae(filt(None, {'deadline': '2003-02-16;'}), ['1', '3', '4'])
def testFilteringRangeYearMonthDay(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'deadline': '2002'}), [])
- ae(filt(None, {'deadline': '2003'}), ['1', '2', '3'])
- ae(filt(None, {'deadline': '2004'}), ['4'])
- ae(filt(None, {'deadline': '2003-02-16'}), ['1'])
- ae(filt(None, {'deadline': '2003-02-17'}), [])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'deadline': '2002'}), [])
+ ae(filt(None, {'deadline': '2003'}), ['1', '2', '3'])
+ ae(filt(None, {'deadline': '2004'}), ['4'])
+ ae(filt(None, {'deadline': '2003-02-16'}), ['1'])
+ ae(filt(None, {'deadline': '2003-02-17'}), [])
def testFilteringRangeMonths(self):
- ae, filt = self.filteringSetup()
+ ae, filter, filter_iter = self.filteringSetup()
for month in range(1, 13):
for n in range(1, month+1):
i = self.db.issue.create(title='%d.%d'%(month, n),
self.db.commit()
for month in range(1, 13):
- r = filt(None, dict(deadline='2001-%02d'%month))
- assert len(r) == month, 'month %d != length %d'%(month, len(r))
+ for filt in filter, filter_iter:
+ r = filt(None, dict(deadline='2001-%02d'%month))
+ assert len(r) == month, 'month %d != length %d'%(month, len(r))
def testFilteringRangeInterval(self):
- ae, filt = self.filteringSetup()
- ae(filt(None, {'foo': 'from 0:50 to 2:00'}), ['1'])
- ae(filt(None, {'foo': 'from 0:50 to 1d 2:00'}), ['1', '2'])
- ae(filt(None, {'foo': 'from 5:50'}), ['2'])
- ae(filt(None, {'foo': 'to 0:05'}), [])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'foo': 'from 0:50 to 2:00'}), ['1'])
+ ae(filt(None, {'foo': 'from 0:50 to 1d 2:00'}), ['1', '2'])
+ ae(filt(None, {'foo': 'from 5:50'}), ['2'])
+ ae(filt(None, {'foo': 'to 0:05'}), [])
def testFilteringRangeGeekInterval(self):
- ae, filt = self.filteringSetup()
+ ae, filter, filter_iter = self.filteringSetup()
for issue in (
{ 'deadline': date.Date('. -2d')},
{ 'deadline': date.Date('. -1d')},
{ 'deadline': date.Date('. -8d')},
):
self.db.issue.create(**issue)
- ae(filt(None, {'deadline': '-2d;'}), ['5', '6'])
- ae(filt(None, {'deadline': '-1d;'}), ['6'])
- ae(filt(None, {'deadline': '-1w;'}), ['5', '6'])
+ for filt in filter, filter_iter:
+ ae(filt(None, {'deadline': '-2d;'}), ['5', '6'])
+ ae(filt(None, {'deadline': '-1d;'}), ['6'])
+ ae(filt(None, {'deadline': '-1w;'}), ['5', '6'])
def testFilteringIntervalSort(self):
# 1: '1:10'
# 2: '1d'
# 3: None
# 4: '0:10'
- ae, filt = self.filteringSetup()
- # ascending should sort None, 1:10, 1d
- ae(filt(None, {}, ('+','foo'), (None,None)), ['3', '4', '1', '2'])
- # descending should sort 1d, 1:10, None
- ae(filt(None, {}, ('-','foo'), (None,None)), ['2', '1', '4', '3'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ # ascending should sort None, 1:10, 1d
+ ae(filt(None, {}, ('+','foo'), (None,None)), ['3', '4', '1', '2'])
+ # descending should sort 1d, 1:10, None
+ ae(filt(None, {}, ('-','foo'), (None,None)), ['2', '1', '4', '3'])
def testFilteringStringSort(self):
# 1: 'issue one'
# 2: 'issue two'
# 3: 'issue three'
# 4: 'non four'
- ae, filt = self.filteringSetup()
- ae(filt(None, {}, ('+','title')), ['1', '3', '2', '4'])
- ae(filt(None, {}, ('-','title')), ['4', '2', '3', '1'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {}, ('+','title')), ['1', '3', '2', '4'])
+ ae(filt(None, {}, ('-','title')), ['4', '2', '3', '1'])
# Test string case: For now allow both, w/wo case matching.
# 1: 'issue one'
# 2: 'issue two'
# 3: 'Issue three'
# 4: 'non four'
self.db.issue.set('3', title='Issue three')
- ae(filt(None, {}, ('+','title')), ['1', '3', '2', '4'])
- ae(filt(None, {}, ('-','title')), ['4', '2', '3', '1'])
+ for filt in filter, filter_iter:
+ ae(filt(None, {}, ('+','title')), ['1', '3', '2', '4'])
+ ae(filt(None, {}, ('-','title')), ['4', '2', '3', '1'])
# Obscure bug in anydbm backend trying to convert to number
# 1: '1st issue'
# 2: '2'
# 4: 'non four'
self.db.issue.set('1', title='1st issue')
self.db.issue.set('2', title='2')
- ae(filt(None, {}, ('+','title')), ['1', '2', '3', '4'])
- ae(filt(None, {}, ('-','title')), ['4', '3', '2', '1'])
+ for filt in filter, filter_iter:
+ ae(filt(None, {}, ('+','title')), ['1', '2', '3', '4'])
+ ae(filt(None, {}, ('-','title')), ['4', '3', '2', '1'])
def testFilteringMultilinkSort(self):
# 1: [] Reverse: 1: []
# Note the sort order for the multilink doen't change when
# reversing the sort direction due to the re-sorting of the
# multilink!
- ae, filt = self.filteringSetup()
+ # Note that we don't test filter_iter here, Multilink sort-order
+ # isn't defined for that.
+ ae, filt, dummy = self.filteringSetup()
ae(filt(None, {}, ('+','nosy'), (None,None)), ['1', '2', '4', '3'])
ae(filt(None, {}, ('-','nosy'), (None,None)), ['4', '3', '1', '2'])
# 2: status: 1 "unread" nosy: []
# 3: status: 1 "unread" nosy: ['admin','fred']
# 4: status: 3 "testing" nosy: ['admin','bleep','fred']
- ae, filt = self.filteringSetup()
+ # Note that we don't test filter_iter here, Multilink sort-order
+ # isn't defined for that.
+ ae, filt, dummy = self.filteringSetup()
ae(filt(None, {}, ('+','nosy'), ('+','status')), ['1', '4', '2', '3'])
ae(filt(None, {}, ('-','nosy'), ('+','status')), ['1', '4', '3', '2'])
ae(filt(None, {}, ('+','nosy'), ('-','status')), ['2', '3', '4', '1'])
# 2: status: 1 -> 'u', priority: 3 -> 1
# 3: status: 1 -> 'u', priority: 2 -> 3
# 4: status: 3 -> 't', priority: 2 -> 3
- ae, filt = self.filteringSetup()
- ae(filt(None, {}, ('+','status'), ('+','priority')),
- ['1', '2', '4', '3'])
- ae(filt(None, {'priority':'2'}, ('+','status'), ('+','priority')),
- ['4', '3'])
- ae(filt(None, {'priority.order':'3'}, ('+','status'), ('+','priority')),
- ['4', '3'])
- ae(filt(None, {'priority':['2','3']}, ('+','priority'), ('+','status')),
- ['1', '4', '2', '3'])
- ae(filt(None, {}, ('+','priority'), ('+','status')),
- ['1', '4', '2', '3'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for filt in filter, filter_iter:
+ ae(filt(None, {}, ('+','status'), ('+','priority')),
+ ['1', '2', '4', '3'])
+ ae(filt(None, {'priority':'2'}, ('+','status'), ('+','priority')),
+ ['4', '3'])
+ ae(filt(None, {'priority.order':'3'}, ('+','status'),
+ ('+','priority')), ['4', '3'])
+ ae(filt(None, {'priority':['2','3']}, ('+','priority'),
+ ('+','status')), ['1', '4', '2', '3'])
+ ae(filt(None, {}, ('+','priority'), ('+','status')),
+ ['1', '4', '2', '3'])
def testFilteringDateSort(self):
# '1': '2003-02-16.22:50'
# '2': '2003-01-01.00:00'
# '3': '2003-02-18'
# '4': '2004-03-08'
- ae, filt = self.filteringSetup()
- # ascending
- ae(filt(None, {}, ('+','deadline'), (None,None)), ['2', '1', '3', '4'])
- # descending
- ae(filt(None, {}, ('-','deadline'), (None,None)), ['4', '3', '1', '2'])
+ ae, filter, filter_iter = self.filteringSetup()
+ for f in filter, filter_iter:
+ # ascending
+ ae(f(None, {}, ('+','deadline'), (None,None)), ['2', '1', '3', '4'])
+ # descending
+ ae(f(None, {}, ('-','deadline'), (None,None)), ['4', '3', '1', '2'])
def testFilteringDateSortPriorityGroup(self):
# '1': '2003-02-16.22:50' 1 => 2
# '2': '2003-01-01.00:00' 3 => 1
# '3': '2003-02-18' 2 => 3
# '4': '2004-03-08' 1 => 2
- ae, filt = self.filteringSetup()
-
- # ascending
- ae(filt(None, {}, ('+','deadline'), ('+','priority')),
- ['2', '1', '3', '4'])
- ae(filt(None, {}, ('-','deadline'), ('+','priority')),
- ['1', '2', '4', '3'])
- # descending
- ae(filt(None, {}, ('+','deadline'), ('-','priority')),
- ['3', '4', '2', '1'])
- ae(filt(None, {}, ('-','deadline'), ('-','priority')),
- ['4', '3', '1', '2'])
-
- def filteringSetupTransitiveSearch(self):
- u_m = {}
- k = 30
- for user in (
- {'username': 'ceo', 'age': 129},
- {'username': 'grouplead1', 'age': 29, 'supervisor': '3'},
- {'username': 'grouplead2', 'age': 29, 'supervisor': '3'},
- {'username': 'worker1', 'age': 25, 'supervisor' : '4'},
- {'username': 'worker2', 'age': 24, 'supervisor' : '4'},
- {'username': 'worker3', 'age': 23, 'supervisor' : '5'},
- {'username': 'worker4', 'age': 22, 'supervisor' : '5'},
- {'username': 'worker5', 'age': 21, 'supervisor' : '5'}):
- u = self.db.user.create(**user)
- u_m [u] = self.db.msg.create(author = u, content = ' '
- , date = date.Date ('2006-01-%s' % k))
- k -= 1
- iss = self.db.issue
- for issue in (
- {'title': 'ts1', 'status': '2', 'assignedto': '6',
- 'priority': '3', 'messages' : [u_m ['6']], 'nosy' : ['4']},
- {'title': 'ts2', 'status': '1', 'assignedto': '6',
- 'priority': '3', 'messages' : [u_m ['6']], 'nosy' : ['5']},
- {'title': 'ts4', 'status': '2', 'assignedto': '7',
- 'priority': '3', 'messages' : [u_m ['7']]},
- {'title': 'ts5', 'status': '1', 'assignedto': '8',
- 'priority': '3', 'messages' : [u_m ['8']]},
- {'title': 'ts6', 'status': '2', 'assignedto': '9',
- 'priority': '3', 'messages' : [u_m ['9']]},
- {'title': 'ts7', 'status': '1', 'assignedto': '10',
- 'priority': '3', 'messages' : [u_m ['10']]},
- {'title': 'ts8', 'status': '2', 'assignedto': '10',
- 'priority': '3', 'messages' : [u_m ['10']]},
- {'title': 'ts9', 'status': '1', 'assignedto': '10',
- 'priority': '3', 'messages' : [u_m ['10'], u_m ['9']]}):
- self.db.issue.create(**issue)
- return self.assertEqual, self.db.issue.filter
+ ae, filter, filter_iter = self.filteringSetup()
+
+ for filt in filter, filter_iter:
+ # ascending
+ ae(filt(None, {}, ('+','deadline'), ('+','priority')),
+ ['2', '1', '3', '4'])
+ ae(filt(None, {}, ('-','deadline'), ('+','priority')),
+ ['1', '2', '4', '3'])
+ # descending
+ ae(filt(None, {}, ('+','deadline'), ('-','priority')),
+ ['3', '4', '2', '1'])
+ ae(filt(None, {}, ('-','deadline'), ('-','priority')),
+ ['4', '3', '1', '2'])
def testFilteringTransitiveLinkUser(self):
- ae, filt = self.filteringSetupTransitiveSearch()
- ufilt = self.db.user.filter
- ae(ufilt(None, {'supervisor.username': 'ceo'}, ('+','username')),
- ['4', '5'])
- ae(ufilt(None, {'supervisor.supervisor.username': 'ceo'},
- ('+','username')), ['6', '7', '8', '9', '10'])
- ae(ufilt(None, {'supervisor.supervisor': '3'}, ('+','username')),
- ['6', '7', '8', '9', '10'])
- ae(ufilt(None, {'supervisor.supervisor.id': '3'}, ('+','username')),
- ['6', '7', '8', '9', '10'])
- ae(ufilt(None, {'supervisor.username': 'grouplead1'}, ('+','username')),
- ['6', '7'])
- ae(ufilt(None, {'supervisor.username': 'grouplead2'}, ('+','username')),
- ['8', '9', '10'])
- ae(ufilt(None, {'supervisor.username': 'grouplead2',
- 'supervisor.supervisor.username': 'ceo'}, ('+','username')),
- ['8', '9', '10'])
- ae(ufilt(None, {'supervisor.supervisor': '3', 'supervisor': '4'},
- ('+','username')), ['6', '7'])
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch('user')
+ for f in filter, filter_iter:
+ ae(f(None, {'supervisor.username': 'ceo'}, ('+','username')),
+ ['4', '5'])
+ ae(f(None, {'supervisor.supervisor.username': 'ceo'},
+ ('+','username')), ['6', '7', '8', '9', '10'])
+ ae(f(None, {'supervisor.supervisor': '3'}, ('+','username')),
+ ['6', '7', '8', '9', '10'])
+ ae(f(None, {'supervisor.supervisor.id': '3'}, ('+','username')),
+ ['6', '7', '8', '9', '10'])
+ ae(f(None, {'supervisor.username': 'grouplead1'}, ('+','username')),
+ ['6', '7'])
+ ae(f(None, {'supervisor.username': 'grouplead2'}, ('+','username')),
+ ['8', '9', '10'])
+ ae(f(None, {'supervisor.username': 'grouplead2',
+ 'supervisor.supervisor.username': 'ceo'}, ('+','username')),
+ ['8', '9', '10'])
+ ae(f(None, {'supervisor.supervisor': '3', 'supervisor': '4'},
+ ('+','username')), ['6', '7'])
def testFilteringTransitiveLinkSort(self):
- ae, filt = self.filteringSetupTransitiveSearch()
- ufilt = self.db.user.filter
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch()
+ ae, ufilter, ufilter_iter = self.iterSetup('user')
# Need to make ceo his own (and first two users') supervisor,
# otherwise we will depend on sorting order of NULL values.
# Leave that to a separate test.
self.db.user.set('1', supervisor = '3')
self.db.user.set('2', supervisor = '3')
self.db.user.set('3', supervisor = '3')
- ae(ufilt(None, {'supervisor':'3'}, []), ['1', '2', '3', '4', '5'])
- ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
- ('+','supervisor.supervisor'), ('+','supervisor'),
- ('+','username')]),
- ['1', '3', '2', '4', '5', '6', '7', '8', '9', '10'])
- ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
- ('-','supervisor.supervisor'), ('-','supervisor'),
- ('+','username')]),
- ['8', '9', '10', '6', '7', '1', '3', '2', '4', '5'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('+','assignedto.supervisor'), ('+','assignedto')]),
- ['1', '2', '3', '4', '5', '6', '7', '8'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('-','assignedto.supervisor'), ('+','assignedto')]),
- ['4', '5', '6', '7', '8', '1', '2', '3'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('+','assignedto.supervisor'), ('+','assignedto'),
- ('-','status')]),
- ['2', '1', '3', '4', '5', '6', '8', '7'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('+','assignedto.supervisor'), ('+','assignedto'),
- ('+','status')]),
- ['1', '2', '3', '4', '5', '7', '6', '8'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('-','assignedto.supervisor'), ('+','assignedto'), ('+','status')]),
- ['4', '5', '7', '6', '8', '1', '2', '3'])
- ae(filt(None, {'assignedto':['6','7','8','9','10']},
- [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('-','assignedto.supervisor'), ('+','assignedto'), ('+','status')]),
- ['4', '5', '7', '6', '8', '1', '2', '3'])
- ae(filt(None, {'assignedto':['6','7','8','9']},
- [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('-','assignedto.supervisor'), ('+','assignedto'), ('+','status')]),
- ['4', '5', '1', '2', '3'])
+ for ufilt in ufilter, ufilter_iter:
+ ae(ufilt(None, {'supervisor':'3'}, []), ['1', '2', '3', '4', '5'])
+ ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
+ ('+','supervisor.supervisor'), ('+','supervisor'),
+ ('+','username')]),
+ ['1', '3', '2', '4', '5', '6', '7', '8', '9', '10'])
+ ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
+ ('-','supervisor.supervisor'), ('-','supervisor'),
+ ('+','username')]),
+ ['8', '9', '10', '6', '7', '1', '3', '2', '4', '5'])
+ for f in filter, filter_iter:
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('+','assignedto.supervisor'), ('+','assignedto')]),
+ ['1', '2', '3', '4', '5', '6', '7', '8'])
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto')]),
+ ['4', '5', '6', '7', '8', '1', '2', '3'])
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('+','assignedto.supervisor'), ('+','assignedto'),
+ ('-','status')]),
+ ['2', '1', '3', '4', '5', '6', '8', '7'])
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('+','assignedto.supervisor'), ('+','assignedto'),
+ ('+','status')]),
+ ['1', '2', '3', '4', '5', '7', '6', '8'])
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto'),
+ ('+','status')]), ['4', '5', '7', '6', '8', '1', '2', '3'])
+ ae(f(None, {'assignedto':['6','7','8','9','10']},
+ [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto'),
+ ('+','status')]), ['4', '5', '7', '6', '8', '1', '2', '3'])
+ ae(f(None, {'assignedto':['6','7','8','9']},
+ [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto'),
+ ('+','status')]), ['4', '5', '1', '2', '3'])
def testFilteringTransitiveLinkSortNull(self):
"""Check sorting of NULL values"""
- ae, filt = self.filteringSetupTransitiveSearch()
- ufilt = self.db.user.filter
- ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
- ('+','supervisor.supervisor'), ('+','supervisor'),
- ('+','username')]),
- ['1', '3', '2', '4', '5', '6', '7', '8', '9', '10'])
- ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
- ('-','supervisor.supervisor'), ('-','supervisor'),
- ('+','username')]),
- ['8', '9', '10', '6', '7', '4', '5', '1', '3', '2'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('+','assignedto.supervisor'), ('+','assignedto')]),
- ['1', '2', '3', '4', '5', '6', '7', '8'])
- ae(filt(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
- ('+','assignedto.supervisor.supervisor'),
- ('-','assignedto.supervisor'), ('+','assignedto')]),
- ['4', '5', '6', '7', '8', '1', '2', '3'])
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch()
+ ae, ufilter, ufilter_iter = self.iterSetup('user')
+ for ufilt in ufilter, ufilter_iter:
+ ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
+ ('+','supervisor.supervisor'), ('+','supervisor'),
+ ('+','username')]),
+ ['1', '3', '2', '4', '5', '6', '7', '8', '9', '10'])
+ ae(ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
+ ('-','supervisor.supervisor'), ('-','supervisor'),
+ ('+','username')]),
+ ['8', '9', '10', '6', '7', '4', '5', '1', '3', '2'])
+ for f in filter, filter_iter:
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('+','assignedto.supervisor'), ('+','assignedto')]),
+ ['1', '2', '3', '4', '5', '6', '7', '8'])
+ ae(f(None, {}, [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto')]),
+ ['4', '5', '6', '7', '8', '1', '2', '3'])
def testFilteringTransitiveLinkIssue(self):
- ae, filt = self.filteringSetupTransitiveSearch()
- ae(filt(None, {'assignedto.supervisor.username': 'grouplead1'},
- ('+','id')), ['1', '2', '3'])
- ae(filt(None, {'assignedto.supervisor.username': 'grouplead2'},
- ('+','id')), ['4', '5', '6', '7', '8'])
- ae(filt(None, {'assignedto.supervisor.username': 'grouplead2',
- 'status': '1'}, ('+','id')), ['4', '6', '8'])
- ae(filt(None, {'assignedto.supervisor.username': 'grouplead2',
- 'status': '2'}, ('+','id')), ['5', '7'])
- ae(filt(None, {'assignedto.supervisor.username': ['grouplead2'],
- 'status': '2'}, ('+','id')), ['5', '7'])
- ae(filt(None, {'assignedto.supervisor': ['4', '5'], 'status': '2'},
- ('+','id')), ['1', '3', '5', '7'])
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'assignedto.supervisor.username': 'grouplead1'},
+ ('+','id')), ['1', '2', '3'])
+ ae(filt(None, {'assignedto.supervisor.username': 'grouplead2'},
+ ('+','id')), ['4', '5', '6', '7', '8'])
+ ae(filt(None, {'assignedto.supervisor.username': 'grouplead2',
+ 'status': '1'}, ('+','id')), ['4', '6', '8'])
+ ae(filt(None, {'assignedto.supervisor.username': 'grouplead2',
+ 'status': '2'}, ('+','id')), ['5', '7'])
+ ae(filt(None, {'assignedto.supervisor.username': ['grouplead2'],
+ 'status': '2'}, ('+','id')), ['5', '7'])
+ ae(filt(None, {'assignedto.supervisor': ['4', '5'], 'status': '2'},
+ ('+','id')), ['1', '3', '5', '7'])
def testFilteringTransitiveMultilink(self):
- ae, filt = self.filteringSetupTransitiveSearch()
- ae(filt(None, {'messages.author.username': 'grouplead1'},
- ('+','id')), [])
- ae(filt(None, {'messages.author': '6'},
- ('+','id')), ['1', '2'])
- ae(filt(None, {'messages.author.id': '6'},
- ('+','id')), ['1', '2'])
- ae(filt(None, {'messages.author.username': 'worker1'},
- ('+','id')), ['1', '2'])
- ae(filt(None, {'messages.author': '10'},
- ('+','id')), ['6', '7', '8'])
- ae(filt(None, {'messages.author': '9'},
- ('+','id')), ['5', '8'])
- ae(filt(None, {'messages.author': ['9', '10']},
- ('+','id')), ['5', '6', '7', '8'])
- ae(filt(None, {'messages.author': ['8', '9']},
- ('+','id')), ['4', '5', '8'])
- ae(filt(None, {'messages.author': ['8', '9'], 'status' : '1'},
- ('+','id')), ['4', '8'])
- ae(filt(None, {'messages.author': ['8', '9'], 'status' : '2'},
- ('+','id')), ['5'])
- ae(filt(None, {'messages.author': ['8', '9', '10'],
- 'messages.date': '2006-01-22.21:00;2006-01-23'}, ('+','id')),
- ['6', '7', '8'])
- ae(filt(None, {'nosy.supervisor.username': 'ceo'},
- ('+','id')), ['1', '2'])
- ae(filt(None, {'messages.author': ['6', '9']},
- ('+','id')), ['1', '2', '5', '8'])
- ae(filt(None, {'messages': ['5', '7']},
- ('+','id')), ['3', '5', '8'])
- ae(filt(None, {'messages.author': ['6', '9'], 'messages': ['5', '7']},
- ('+','id')), ['5', '8'])
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch()
+ for filt in filter, filter_iter:
+ ae(filt(None, {'messages.author.username': 'grouplead1'},
+ ('+','id')), [])
+ ae(filt(None, {'messages.author': '6'},
+ ('+','id')), ['1', '2'])
+ ae(filt(None, {'messages.author.id': '6'},
+ ('+','id')), ['1', '2'])
+ ae(filt(None, {'messages.author.username': 'worker1'},
+ ('+','id')), ['1', '2'])
+ ae(filt(None, {'messages.author': '10'},
+ ('+','id')), ['6', '7', '8'])
+ ae(filt(None, {'messages.author': '9'},
+ ('+','id')), ['5', '8'])
+ ae(filt(None, {'messages.author': ['9', '10']},
+ ('+','id')), ['5', '6', '7', '8'])
+ ae(filt(None, {'messages.author': ['8', '9']},
+ ('+','id')), ['4', '5', '8'])
+ ae(filt(None, {'messages.author': ['8', '9'], 'status' : '1'},
+ ('+','id')), ['4', '8'])
+ ae(filt(None, {'messages.author': ['8', '9'], 'status' : '2'},
+ ('+','id')), ['5'])
+ ae(filt(None, {'messages.author': ['8', '9', '10'],
+ 'messages.date': '2006-01-22.21:00;2006-01-23'}, ('+','id')),
+ ['6', '7', '8'])
+ ae(filt(None, {'nosy.supervisor.username': 'ceo'},
+ ('+','id')), ['1', '2'])
+ ae(filt(None, {'messages.author': ['6', '9']},
+ ('+','id')), ['1', '2', '5', '8'])
+ ae(filt(None, {'messages': ['5', '7']},
+ ('+','id')), ['3', '5', '8'])
+ ae(filt(None, {'messages.author': ['6', '9'],
+ 'messages': ['5', '7']}, ('+','id')), ['5', '8'])
def testFilteringTransitiveMultilinkSort(self):
- ae, filt = self.filteringSetupTransitiveSearch()
+ # Note that we don't test filter_iter here, Multilink sort-order
+ # isn't defined for that.
+ ae, filt, dummy = self.filteringSetupTransitiveSearch()
ae(filt(None, {}, [('+','messages.author')]),
['1', '2', '3', '4', '5', '8', '6', '7'])
ae(filt(None, {}, [('-','messages.author')]),
['3', '1', '2', '6', '7', '5', '4', '8'])
def testFilteringSortId(self):
- ae, filt = self.filteringSetupTransitiveSearch()
- ae(self.db.user.filter(None, {}, ('+','id')),
- ['1', '2', '3', '4', '5', '6', '7', '8', '9', '10'])
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch('user')
+ for filt in filter, filter_iter:
+ ae(filt(None, {}, ('+','id')),
+ ['1', '2', '3', '4', '5', '6', '7', '8', '9', '10'])
# XXX add sorting tests for other types
+ # nuke and re-create db for restore
+ def nukeAndCreate(self):
+ # shut down this db and nuke it
+ self.db.close()
+ self.nuke_database()
+
+ # open a new, empty database
+ os.makedirs(config.DATABASE + '/files')
+ self.db = self.module.Database(config, 'admin')
+ setupSchema(self.db, 0, self.module)
+
def testImportExport(self):
# use the filtering setup to create a bunch of items
- ae, filt = self.filteringSetup()
+ ae, dummy1, dummy2 = self.filteringSetup()
# Get some stuff into the journal for testing import/export of
# journal data:
self.db.user.set('4', password = password.Password('xyzzy'))
klass.export_files('_test_export', id)
journals[cn] = klass.export_journals()
- # shut down this db and nuke it
- self.db.close()
- self.nuke_database()
-
- # open a new, empty database
- os.makedirs(config.DATABASE + '/files')
- self.db = self.module.Database(config, 'admin')
- setupSchema(self.db, 0, self.module)
+ self.nukeAndCreate()
# import
for cn, items in export.items():
newid = self.db.user.create(username='testing')
assert newid > maxid
+ # test import/export via admin interface
+ def testAdminImportExport(self):
+ import roundup.admin
+ import csv
+ # use the filtering setup to create a bunch of items
+ ae, dummy1, dummy2 = self.filteringSetup()
+ # create large field
+ self.db.priority.create(name = 'X' * 500)
+ self.db.config.CSV_FIELD_SIZE = 400
+ self.db.commit()
+ output = []
+ # ugly hack to get stderr output and disable stdout output
+ # during regression test. Depends on roundup.admin not using
+ # anything but stdout/stderr from sys (which is currently the
+ # case)
+ def stderrwrite(s):
+ output.append(s)
+ roundup.admin.sys = MockNull ()
+ try:
+ roundup.admin.sys.stderr.write = stderrwrite
+ tool = roundup.admin.AdminTool()
+ home = '.'
+ tool.tracker_home = home
+ tool.db = self.db
+ tool.verbose = False
+ tool.do_export (['_test_export'])
+ self.assertEqual(len(output), 2)
+ self.assertEqual(output [1], '\n')
+ self.failUnless(output [0].startswith
+ ('Warning: config csv_field_size should be at least'))
+ self.failUnless(int(output[0].split()[-1]) > 500)
+
+ if hasattr(roundup.admin.csv, 'field_size_limit'):
+ self.nukeAndCreate()
+ self.db.config.CSV_FIELD_SIZE = 400
+ tool = roundup.admin.AdminTool()
+ tool.tracker_home = home
+ tool.db = self.db
+ tool.verbose = False
+ self.assertRaises(csv.Error, tool.do_import, ['_test_export'])
+
+ self.nukeAndCreate()
+ self.db.config.CSV_FIELD_SIZE = 3200
+ tool = roundup.admin.AdminTool()
+ tool.tracker_home = home
+ tool.db = self.db
+ tool.verbose = False
+ tool.do_import(['_test_export'])
+ finally:
+ roundup.admin.sys = sys
+ shutil.rmtree('_test_export')
+
def testAddProperty(self):
self.db.issue.create(title="spam", status='1')
self.db.commit()
"""Creates one issue with two attachments, one smaller and one larger
than the set max_attachment_size.
"""
+ old_translate_ = roundupdb._
+ roundupdb._ = i18n.get_translation(language='C').gettext
db = self.db
db.config.NOSY_MAX_ATTACHMENT_SIZE = 4096
res = dict(mail_to = None, mail_msg = None)
messages = [m], nosy = [db.user.lookup("fred")])
db.issue.nosymessage(i, m, {})
- mail_msg = res["mail_msg"].getvalue()
+ mail_msg = str(res["mail_msg"])
self.assertEqual(res["mail_to"], ["fred@example.com"])
- self.failUnless("From: admin" in mail_msg)
- self.failUnless("Subject: [issue1] spam" in mail_msg)
- self.failUnless("New submission from admin" in mail_msg)
- self.failUnless("one two" in mail_msg)
- self.failIf("File 'test1.txt' not attached" in mail_msg)
- self.failUnless(base64.encodestring("xxx").rstrip() in mail_msg)
- self.failUnless("File 'test2.txt' not attached" in mail_msg)
- self.failIf(base64.encodestring("yyy").rstrip() in mail_msg)
+ self.assert_("From: admin" in mail_msg)
+ self.assert_("Subject: [issue1] spam" in mail_msg)
+ self.assert_("New submission from admin" in mail_msg)
+ self.assert_("one two" in mail_msg)
+ self.assert_("File 'test1.txt' not attached" not in mail_msg)
+ self.assert_(base64.encodestring("xxx").rstrip() in mail_msg)
+ self.assert_("File 'test2.txt' not attached" in mail_msg)
+ self.assert_(base64.encodestring("yyy").rstrip() not in mail_msg)
+ finally :
+ roundupdb._ = old_translate_
+ Mailer.smtp_send = backup
+
+ def testPGPNosyMail(self) :
+ """Creates one issue with two attachments, one smaller and one larger
+ than the set max_attachment_size. Recipients are one with and
+ one without encryption enabled via a gpg group.
+ """
+ if gpgmelib.pyme is None:
+ print "Skipping PGPNosy test"
+ return
+ old_translate_ = roundupdb._
+ roundupdb._ = i18n.get_translation(language='C').gettext
+ db = self.db
+ db.config.NOSY_MAX_ATTACHMENT_SIZE = 4096
+ db.config['PGP_HOMEDIR'] = gpgmelib.pgphome
+ db.config['PGP_ROLES'] = 'pgp'
+ db.config['PGP_ENABLE'] = True
+ db.config['PGP_ENCRYPT'] = True
+ gpgmelib.setUpPGP()
+ res = []
+ def dummy_snd(s, to, msg, res=res) :
+ res.append (dict (mail_to = to, mail_msg = msg))
+ backup, Mailer.smtp_send = Mailer.smtp_send, dummy_snd
+ try :
+ john = db.user.create(username="john", roles='User,pgp',
+ address='john@test.test', realname='John Doe')
+ f1 = db.file.create(name="test1.txt", content="x" * 20)
+ f2 = db.file.create(name="test2.txt", content="y" * 5000)
+ m = db.msg.create(content="one two", author="admin",
+ files = [f1, f2])
+ i = db.issue.create(title='spam', files = [f1, f2],
+ messages = [m], nosy = [db.user.lookup("fred"), john])
+
+ db.issue.nosymessage(i, m, {})
+ res.sort(key=lambda x: x['mail_to'])
+ self.assertEqual(res[0]["mail_to"], ["fred@example.com"])
+ self.assertEqual(res[1]["mail_to"], ["john@test.test"])
+ mail_msg = str(res[0]["mail_msg"])
+ self.assert_("From: admin" in mail_msg)
+ self.assert_("Subject: [issue1] spam" in mail_msg)
+ self.assert_("New submission from admin" in mail_msg)
+ self.assert_("one two" in mail_msg)
+ self.assert_("File 'test1.txt' not attached" not in mail_msg)
+ self.assert_(base64.encodestring("xxx").rstrip() in mail_msg)
+ self.assert_("File 'test2.txt' not attached" in mail_msg)
+ self.assert_(base64.encodestring("yyy").rstrip() not in mail_msg)
+ fp = FeedParser()
+ mail_msg = str(res[1]["mail_msg"])
+ fp.feed(mail_msg)
+ parts = fp.close().get_payload()
+ self.assertEqual(len(parts),2)
+ self.assertEqual(parts[0].get_payload().strip(), 'Version: 1')
+ crypt = gpgmelib.pyme.core.Data(parts[1].get_payload())
+ plain = gpgmelib.pyme.core.Data()
+ ctx = gpgmelib.pyme.core.Context()
+ res = ctx.op_decrypt(crypt, plain)
+ self.assertEqual(res, None)
+ plain.seek(0,0)
+ fp = FeedParser()
+ fp.feed(plain.read())
+ self.assert_("From: admin" in mail_msg)
+ self.assert_("Subject: [issue1] spam" in mail_msg)
+ mail_msg = str(fp.close())
+ self.assert_("New submission from admin" in mail_msg)
+ self.assert_("one two" in mail_msg)
+ self.assert_("File 'test1.txt' not attached" not in mail_msg)
+ self.assert_(base64.encodestring("xxx").rstrip() in mail_msg)
+ self.assert_("File 'test2.txt' not attached" in mail_msg)
+ self.assert_(base64.encodestring("yyy").rstrip() not in mail_msg)
finally :
+ roundupdb._ = old_translate_
Mailer.smtp_send = backup
+ gpgmelib.tearDownPGP()
class ROTest(MyTestCase):
def setUp(self):
os.makedirs(config.DATABASE + '/files')
def test_reservedProperties(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
self.assertRaises(ValueError, self.module.Class, self.db, "a",
creation=String())
self.assertRaises(ValueError, self.module.Class, self.db, "a",
actor=String())
def init_a(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.Class(self.db, "a", name=String())
a.setkey("name")
self.db.post_init()
def test_fileClassProps(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.FileClass(self.db, 'a')
l = a.getprops().keys()
l.sort()
'creation', 'type'])
def init_ab(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.Class(self.db, "a", name=String())
a.setkey("name")
b = self.module.Class(self.db, "b", name=String(),
self.db.getjournal('b', bid)
def init_amod(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.Class(self.db, "a", name=String(), newstr=String(),
newint=Interval(), newnum=Number(), newbool=Boolean(),
newdate=Date())
self.db.getjournal('a', aid2)
def init_amodkey(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.Class(self.db, "a", name=String(), newstr=String())
a.setkey("newstr")
b = self.module.Class(self.db, "b", name=String())
def init_amodml(self):
- self.db = self.module.Database(config, 'admin')
+ self.open_database()
a = self.module.Class(self.db, "a", name=String(),
newml=Multilink('a'))
a.setkey('name')
self.db.getjournal('a', aid)
class RDBMSTest:
- ''' tests specific to RDBMS backends '''
+ """ tests specific to RDBMS backends """
def test_indexTest(self):
self.assertEqual(self.db.sql_index_exists('_issue', '_issue_id_idx'), 1)
self.assertEqual(self.db.sql_index_exists('_issue', '_issue_x_idx'), 0)
+class FilterCacheTest(commonDBTest):
+ def testFilteringTransitiveLinkCache(self):
+ ae, filter, filter_iter = self.filteringSetupTransitiveSearch()
+ ae, ufilter, ufilter_iter = self.iterSetup('user')
+ # Need to make ceo his own (and first two users') supervisor
+ self.db.user.set('1', supervisor = '3')
+ self.db.user.set('2', supervisor = '3')
+ self.db.user.set('3', supervisor = '3')
+ # test bool value
+ self.db.user.set('4', assignable = True)
+ self.db.user.set('3', assignable = False)
+ filt = self.db.issue.filter_iter
+ ufilt = self.db.user.filter_iter
+ user_result = \
+ { '1' : {'username': 'admin', 'assignable': None,
+ 'supervisor': '3', 'realname': None, 'roles': 'Admin',
+ 'creator': '1', 'age': None, 'actor': '1',
+ 'address': None}
+ , '2' : {'username': 'fred', 'assignable': None,
+ 'supervisor': '3', 'realname': None, 'roles': 'User',
+ 'creator': '1', 'age': None, 'actor': '1',
+ 'address': 'fred@example.com'}
+ , '3' : {'username': 'ceo', 'assignable': False,
+ 'supervisor': '3', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 129.0, 'actor': '1',
+ 'address': None}
+ , '4' : {'username': 'grouplead1', 'assignable': True,
+ 'supervisor': '3', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 29.0, 'actor': '1',
+ 'address': None}
+ , '5' : {'username': 'grouplead2', 'assignable': None,
+ 'supervisor': '3', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 29.0, 'actor': '1',
+ 'address': None}
+ , '6' : {'username': 'worker1', 'assignable': None,
+ 'supervisor': '4', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 25.0, 'actor': '1',
+ 'address': None}
+ , '7' : {'username': 'worker2', 'assignable': None,
+ 'supervisor': '4', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 24.0, 'actor': '1',
+ 'address': None}
+ , '8' : {'username': 'worker3', 'assignable': None,
+ 'supervisor': '5', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 23.0, 'actor': '1',
+ 'address': None}
+ , '9' : {'username': 'worker4', 'assignable': None,
+ 'supervisor': '5', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 22.0, 'actor': '1',
+ 'address': None}
+ , '10' : {'username': 'worker5', 'assignable': None,
+ 'supervisor': '5', 'realname': None, 'roles': None,
+ 'creator': '1', 'age': 21.0, 'actor': '1',
+ 'address': None}
+ }
+ foo = date.Interval('-1d')
+ issue_result = \
+ { '1' : {'title': 'ts1', 'status': '2', 'assignedto': '6',
+ 'priority': '3', 'messages' : ['4'], 'nosy' : ['4']}
+ , '2' : {'title': 'ts2', 'status': '1', 'assignedto': '6',
+ 'priority': '3', 'messages' : ['4'], 'nosy' : ['5']}
+ , '3' : {'title': 'ts4', 'status': '2', 'assignedto': '7',
+ 'priority': '3', 'messages' : ['5']}
+ , '4' : {'title': 'ts5', 'status': '1', 'assignedto': '8',
+ 'priority': '3', 'messages' : ['6']}
+ , '5' : {'title': 'ts6', 'status': '2', 'assignedto': '9',
+ 'priority': '3', 'messages' : ['7']}
+ , '6' : {'title': 'ts7', 'status': '1', 'assignedto': '10',
+ 'priority': '3', 'messages' : ['8'], 'foo' : None}
+ , '7' : {'title': 'ts8', 'status': '2', 'assignedto': '10',
+ 'priority': '3', 'messages' : ['8'], 'foo' : foo}
+ , '8' : {'title': 'ts9', 'status': '1', 'assignedto': '10',
+ 'priority': '3', 'messages' : ['7', '8']}
+ }
+ result = []
+ self.db.clearCache()
+ for id in ufilt(None, {}, [('+','supervisor.supervisor.supervisor'),
+ ('-','supervisor.supervisor'), ('-','supervisor'),
+ ('+','username')]):
+ result.append(id)
+ nodeid = id
+ for x in range(4):
+ assert(('user', nodeid) in self.db.cache)
+ n = self.db.user.getnode(nodeid)
+ for k, v in user_result[nodeid].iteritems():
+ ae((k, n[k]), (k, v))
+ for k in 'creation', 'activity':
+ assert(n[k])
+ nodeid = n.supervisor
+ self.db.clearCache()
+ ae (result, ['8', '9', '10', '6', '7', '1', '3', '2', '4', '5'])
+
+ result = []
+ self.db.clearCache()
+ for id in filt(None, {},
+ [('+','assignedto.supervisor.supervisor.supervisor'),
+ ('+','assignedto.supervisor.supervisor'),
+ ('-','assignedto.supervisor'), ('+','assignedto')]):
+ result.append(id)
+ assert(('issue', id) in self.db.cache)
+ n = self.db.issue.getnode(id)
+ for k, v in issue_result[id].iteritems():
+ ae((k, n[k]), (k, v))
+ for k in 'creation', 'activity':
+ assert(n[k])
+ nodeid = n.assignedto
+ for x in range(4):
+ assert(('user', nodeid) in self.db.cache)
+ n = self.db.user.getnode(nodeid)
+ for k, v in user_result[nodeid].iteritems():
+ ae((k, n[k]), (k, v))
+ for k in 'creation', 'activity':
+ assert(n[k])
+ nodeid = n.supervisor
+ self.db.clearCache()
+ ae (result, ['4', '5', '6', '7', '8', '1', '2', '3'])
+
class ClassicInitTest(unittest.TestCase):
count = 0
except OSError, error:
if error.errno not in (errno.ENOENT, errno.ESRCH): raise
+class ConcurrentDBTest(ClassicInitTest):
+ def testConcurrency(self):
+ # The idea here is a read-modify-update cycle in the presence of
+ # a cache that has to be properly handled. The same applies if
+ # we extend a String or otherwise modify something that depends
+ # on the previous value.
+
+ # set up and open a tracker
+ tracker = setupTracker(self.dirname, self.backend)
+ # open the database
+ self.db = tracker.open('admin')
+
+ prio = '1'
+ self.assertEqual(self.db.priority.get(prio, 'order'), 1.0)
+ def inc(db):
+ db.priority.set(prio, order=db.priority.get(prio, 'order') + 1)
+
+ inc(self.db)
+
+ db2 = tracker.open("admin")
+ self.assertEqual(db2.priority.get(prio, 'order'), 1.0)
+ db2.commit()
+ self.db.commit()
+ self.assertEqual(self.db.priority.get(prio, 'order'), 2.0)
+
+ inc(db2)
+ db2.commit()
+ db2.clearCache()
+ self.assertEqual(db2.priority.get(prio, 'order'), 3.0)
+ db2.close()
+
+
# vim: set et sts=4 sw=4 :