Commit 09cea559 authored by Tres Seaver's avatar Tres Seaver

Unfutz naive timezon issues.

parent fbf2f48c
...@@ -16,18 +16,6 @@ $Id$ ...@@ -16,18 +16,6 @@ $Id$
""" """
import unittest import unittest
import Testing
import Zope2
Zope2.startup()
from datetime import date, datetime, tzinfo, timedelta
import time
from types import IntType, FloatType
from DateTime import DateTime
from Products.PluginIndexes.DateIndex.DateIndex import DateIndex, Local
class Dummy: class Dummy:
...@@ -48,67 +36,86 @@ class Dummy: ...@@ -48,67 +36,86 @@ class Dummy:
############################################################################### ###############################################################################
# excerpted from the Python module docs # excerpted from the Python module docs
############################################################################### ###############################################################################
ZERO = timedelta(0)
HOUR = timedelta(hours=1)
def first_sunday_on_or_after(dt):
days_to_go = 6 - dt.weekday()
if days_to_go:
dt += timedelta(days_to_go)
return dt
# In the US, DST starts at 2am (standard time) on the first Sunday in April.
DSTSTART = datetime(1, 4, 1, 2)
# and ends at 2am (DST time; 1am standard time) on the last Sunday of Oct.
# which is the first Sunday on or after Oct 25.
DSTEND = datetime(1, 10, 25, 1)
class USTimeZone(tzinfo):
def __init__(self, hours, reprname, stdname, dstname):
self.stdoffset = timedelta(hours=hours)
self.reprname = reprname
self.stdname = stdname
self.dstname = dstname
def __repr__(self):
return self.reprname
def tzname(self, dt):
if self.dst(dt):
return self.dstname
else:
return self.stdname
def utcoffset(self, dt):
return self.stdoffset + self.dst(dt)
def dst(self, dt):
if dt is None or dt.tzinfo is None:
# An exception may be sensible here, in one or both cases.
# It depends on how you want to treat them. The default
# fromutc() implementation (called by the default astimezone()
# implementation) passes a datetime with dt.tzinfo is self.
return ZERO
assert dt.tzinfo is self
# Find first Sunday in April & the last in October.
start = first_sunday_on_or_after(DSTSTART.replace(year=dt.year))
end = first_sunday_on_or_after(DSTEND.replace(year=dt.year))
# Can't compare naive to aware objects, so strip the timezone from
# dt first.
if start <= dt.replace(tzinfo=None) < end:
return HOUR
else:
return ZERO
Eastern = USTimeZone(-5, "Eastern", "EST", "EDT") def _getEastern():
from datetime import date
from datetime import datetime
from datetime import timedelta
from datetime import tzinfo
ZERO = timedelta(0)
HOUR = timedelta(hours=1)
def first_sunday_on_or_after(dt):
days_to_go = 6 - dt.weekday()
if days_to_go:
dt += timedelta(days_to_go)
return dt
# In the US, DST starts at 2am (standard time) on the first Sunday in
# April...
DSTSTART = datetime(1, 4, 1, 2)
# and ends at 2am (DST time; 1am standard time) on the last Sunday of
# October, which is the first Sunday on or after Oct 25.
DSTEND = datetime(1, 10, 25, 1)
class USTimeZone(tzinfo):
def __init__(self, hours, reprname, stdname, dstname):
self.stdoffset = timedelta(hours=hours)
self.reprname = reprname
self.stdname = stdname
self.dstname = dstname
def __repr__(self):
return self.reprname
def tzname(self, dt):
if self.dst(dt):
return self.dstname
else:
return self.stdname
def utcoffset(self, dt):
return self.stdoffset + self.dst(dt)
def dst(self, dt):
if dt is None or dt.tzinfo is None:
# An exception may be sensible here, in one or both cases.
# It depends on how you want to treat them. The default
# fromutc() implementation (called by the default astimezone()
# implementation) passes a datetime with dt.tzinfo is self.
return ZERO
assert dt.tzinfo is self
# Find first Sunday in April & the last in October.
start = first_sunday_on_or_after(DSTSTART.replace(year=dt.year))
end = first_sunday_on_or_after(DSTEND.replace(year=dt.year))
# Can't compare naive to aware objects, so strip the timezone from
# dt first.
if start <= dt.replace(tzinfo=None) < end:
return HOUR
else:
return ZERO
return USTimeZone(-5, "Eastern", "EST", "EDT")
############################################################################### ###############################################################################
class DI_Tests(unittest.TestCase): class DI_Tests(unittest.TestCase):
def setUp(self):
self._values = ( def _getTargetClass(self):
from Products.PluginIndexes.DateIndex.DateIndex import DateIndex
return DateIndex
def _makeOne(self, id='date'):
return self._getTargetClass()(id)
def _getValues(self):
from DateTime import DateTime
from datetime import date
from datetime import datetime
return [
(0, Dummy('a', None)), # None (0, Dummy('a', None)), # None
(1, Dummy('b', DateTime(0))), # 1055335680 (1, Dummy('b', DateTime(0))), # 1055335680
(2, Dummy('c', DateTime('2002-05-08 15:16:17'))), # 1072667236 (2, Dummy('c', DateTime('2002-05-08 15:16:17'))), # 1072667236
...@@ -120,29 +127,15 @@ class DI_Tests(unittest.TestCase): ...@@ -120,29 +127,15 @@ class DI_Tests(unittest.TestCase):
(8, Dummy('g', date(2034,2,5))), # 1073599200 (8, Dummy('g', date(2034,2,5))), # 1073599200
(9, Dummy('h', datetime(2034,2,5,15,20,5))), # (varies) (9, Dummy('h', datetime(2034,2,5,15,20,5))), # (varies)
(10, Dummy('i', datetime(2034,2,5,10,17,5, (10, Dummy('i', datetime(2034,2,5,10,17,5,
tzinfo=Eastern))), # 1073600117 tzinfo=_getEastern()))), # 1073600117
) ]
self._index = DateIndex('date')
self._noop_req = {'bar': 123} def _populateIndex(self, index):
self._request = {'date': DateTime(0)} for k, v in self._getValues():
self._min_req = {'date': {'query': DateTime('2032-05-08 15:16:17'), index.index_object(k, v)
'range': 'min'}}
self._max_req = {'date': {'query': DateTime('2032-05-08 15:16:17'), def _checkApply(self, index, req, expectedValues):
'range': 'max'}} result, used = index._apply_index(req)
self._range_req = {'date': {'query':(DateTime('2002-05-08 15:16:17'),
DateTime('2062-05-08 15:16:17')),
'range': 'min:max'}}
self._zero_req = {'date': 0}
self._none_req = {'date': None}
self._float_req = {'date': 1072742620.0}
self._int_req = {'date': 1072742900}
def _populateIndex( self ):
for k, v in self._values:
self._index.index_object(k, v)
def _checkApply(self, req, expectedValues):
result, used = self._index._apply_index(req)
if hasattr(result, 'keys'): if hasattr(result, 'keys'):
result = result.keys() result = result.keys()
self.failUnlessEqual(used, ('date',)) self.failUnlessEqual(used, ('date',))
...@@ -152,8 +145,12 @@ class DI_Tests(unittest.TestCase): ...@@ -152,8 +145,12 @@ class DI_Tests(unittest.TestCase):
self.failUnless(k in result) self.failUnless(k in result)
def _convert(self, dt): def _convert(self, dt):
if type(dt) in (FloatType, IntType): from time import gmtime
yr, mo, dy, hr, mn = time.gmtime(dt)[:5] from datetime import date
from datetime import datetime
from Products.PluginIndexes.DateIndex.DateIndex import Local
if isinstance(dt, (float, int)):
yr, mo, dy, hr, mn = gmtime(dt)[:5]
elif type(dt) is date: elif type(dt) is date:
yr, mo, dy, hr, mn = dt.timetuple()[:5] yr, mo, dy, hr, mn = dt.timetuple()[:5]
elif type(dt) is datetime: elif type(dt) is datetime:
...@@ -171,37 +168,50 @@ class DI_Tests(unittest.TestCase): ...@@ -171,37 +168,50 @@ class DI_Tests(unittest.TestCase):
from Products.PluginIndexes.interfaces import IUniqueValueIndex from Products.PluginIndexes.interfaces import IUniqueValueIndex
from zope.interface.verify import verifyClass from zope.interface.verify import verifyClass
verifyClass(IDateIndex, DateIndex) verifyClass(IDateIndex, self._getTargetClass())
verifyClass(IPluggableIndex, DateIndex) verifyClass(IPluggableIndex, self._getTargetClass())
verifyClass(ISortIndex, DateIndex) verifyClass(ISortIndex, self._getTargetClass())
verifyClass(IUniqueValueIndex, DateIndex) verifyClass(IUniqueValueIndex, self._getTargetClass())
def test_empty(self): def test_empty(self):
empty = self._index from DateTime import DateTime
index = self._makeOne()
self.failUnlessEqual(len(empty), 0) self.failUnlessEqual(len(index), 0)
self.failUnlessEqual(len(empty.referencedObjects()), 0) self.failUnlessEqual(len(index.referencedObjects()), 0)
self.failUnless(empty.getEntryForObject(1234) is None) self.failUnless(index.getEntryForObject(1234) is None)
marker = [] marker = []
self.failUnless(empty.getEntryForObject(1234, marker) is marker) self.failUnless(index.getEntryForObject(1234, marker) is marker)
empty.unindex_object(1234) # shouldn't throw index.unindex_object(1234) # shouldn't throw
self.failUnless(empty.hasUniqueValuesFor('date'))
self.failIf(empty.hasUniqueValuesFor('foo'))
self.failUnlessEqual(len(empty.uniqueValues('date')), 0)
self.failUnless(empty._apply_index({'zed': 12345}) is None)
self._checkApply(self._request, []) self.failUnless(index.hasUniqueValuesFor('date'))
self._checkApply(self._min_req, []) self.failIf(index.hasUniqueValuesFor('foo'))
self._checkApply(self._max_req, []) self.failUnlessEqual(len(index.uniqueValues('date')), 0)
self._checkApply(self._range_req, [])
self.failUnless(index._apply_index({'zed': 12345}) is None)
self._checkApply(index,
{'date': DateTime(0)}, [])
self._checkApply(index,
{'date': {'query': DateTime('2032-05-08 15:16:17'),
'range': 'min'}},
[])
self._checkApply(index,
{'date': {'query': DateTime('2032-05-08 15:16:17'),
'range': 'max'}},
[])
self._checkApply(index,
{'date': {'query':(DateTime('2002-05-08 15:16:17'),
DateTime('2062-05-08 15:16:17')),
'range': 'min:max'}},
[])
def test_retrieval( self ): def test_retrieval( self ):
self._populateIndex() from DateTime import DateTime
values = self._values index = self._makeOne()
index = self._index self._populateIndex(index)
values = self._getValues()
self.failUnlessEqual(len(index), len(values) - 2) # One dupe, one empty self.failUnlessEqual(len(index), len(values) - 2) # One dupe, one empty
self.failUnlessEqual(len(index.referencedObjects()), len(values) - 1) self.failUnlessEqual(len(index.referencedObjects()), len(values) - 1)
...@@ -214,30 +224,43 @@ class DI_Tests(unittest.TestCase): ...@@ -214,30 +224,43 @@ class DI_Tests(unittest.TestCase):
for k, v in values: for k, v in values:
if v.date(): if v.date():
self.failUnlessEqual(self._index.getEntryForObject(k), self.failUnlessEqual(index.getEntryForObject(k),
self._convert(v.date())) self._convert(v.date()))
self.failUnlessEqual(len(index.uniqueValues('date')), len(values) - 2) self.failUnlessEqual(len(index.uniqueValues('date')), len(values) - 2)
self.failUnless(index._apply_index(self._noop_req) is None) self.failUnless(index._apply_index({'bar': 123}) is None)
self._checkApply(self._request, values[1:2]) self._checkApply(index,
self._checkApply(self._min_req, values[3:6] + values[8:]) {'date': DateTime(0)}, values[1:2])
self._checkApply(self._max_req, values[1:4] + values[6:8]) self._checkApply(index,
self._checkApply(self._range_req, values[2:] ) {'date': {'query': DateTime('2032-05-08 15:16:17'),
self._checkApply(self._float_req, [values[6]] ) 'range': 'min'}},
self._checkApply(self._int_req, [values[7]] ) values[3:6] + values[8:])
self._checkApply(index,
{'date': {'query': DateTime('2032-05-08 15:16:17'),
'range': 'max'}},
values[1:4] + values[6:8])
self._checkApply(index,
{'date': {'query':(DateTime('2002-05-08 15:16:17'),
DateTime('2062-05-08 15:16:17')),
'range': 'min:max'}},
values[2:] )
self._checkApply(index,
{'date': 1072742620.0}, [values[6]])
self._checkApply(index,
{'date': 1072742900}, [values[7]])
def test_naive_convert_to_utc(self): def test_naive_convert_to_utc(self):
values = self._values index = self._makeOne()
index = self._index values = self._getValues()
index.index_naive_time_as_local = False index.index_naive_time_as_local = False
self._populateIndex() self._populateIndex(index)
for k, v in values[9:]: for k, v in values[9:]:
# assert that the timezone is effectively UTC for item 9, # assert that the timezone is effectively UTC for item 9,
# and still correct for item 10 # and still correct for item 10
yr, mo, dy, hr, mn = v.date().utctimetuple()[:5] yr, mo, dy, hr, mn = v.date().utctimetuple()[:5]
val = (((yr * 12 + mo) * 31 + dy) * 24 + hr) * 60 + mn val = (((yr * 12 + mo) * 31 + dy) * 24 + hr) * 60 + mn
self.failUnlessEqual(self._index.getEntryForObject(k), val) self.failUnlessEqual(index.getEntryForObject(k), val)
def test_removal(self): def test_removal(self):
""" DateIndex would hand back spurious entries when used as a """ DateIndex would hand back spurious entries when used as a
...@@ -246,10 +269,11 @@ class DI_Tests(unittest.TestCase): ...@@ -246,10 +269,11 @@ class DI_Tests(unittest.TestCase):
None. The catalog consults a sort_index's None. The catalog consults a sort_index's
documentToKeyMap() to build the brains. documentToKeyMap() to build the brains.
""" """
values = self._values values = self._getValues()
index = self._index index = self._makeOne()
self._populateIndex() self._populateIndex(index)
self._checkApply(self._int_req, [values[7]]) self._checkApply(index,
{'date': 1072742900}, [values[7]])
index.index_object(7, None) index.index_object(7, None)
self.failIf(7 in index.documentToKeyMap().keys()) self.failIf(7 in index.documentToKeyMap().keys())
......
...@@ -16,14 +16,6 @@ $Id$ ...@@ -16,14 +16,6 @@ $Id$
""" """
import unittest import unittest
import Testing
import Zope2
Zope2.startup()
import sys
from Products.PluginIndexes.DateRangeIndex.DateRangeIndex import DateRangeIndex
class Dummy: class Dummy:
...@@ -74,11 +66,22 @@ def matchingDummies( value ): ...@@ -74,11 +66,22 @@ def matchingDummies( value ):
class DRI_Tests( unittest.TestCase ): class DRI_Tests( unittest.TestCase ):
def setUp( self ):
pass
def tearDown( self ): def _getTargetClass(self):
pass from Products.PluginIndexes.DateRangeIndex.DateRangeIndex \
import DateRangeIndex
return DateRangeIndex
def _makeOne(self,
id,
since_field=None,
until_field=None,
caller=None,
extra=None,
):
klass = self._getTargetClass()
return klass(id, since_field, until_field, caller, extra)
def test_z3interfaces(self): def test_z3interfaces(self):
from Products.PluginIndexes.interfaces import IDateRangeIndex from Products.PluginIndexes.interfaces import IDateRangeIndex
...@@ -87,14 +90,14 @@ class DRI_Tests( unittest.TestCase ): ...@@ -87,14 +90,14 @@ class DRI_Tests( unittest.TestCase ):
from Products.PluginIndexes.interfaces import IUniqueValueIndex from Products.PluginIndexes.interfaces import IUniqueValueIndex
from zope.interface.verify import verifyClass from zope.interface.verify import verifyClass
verifyClass(IDateRangeIndex, DateRangeIndex) verifyClass(IDateRangeIndex, self._getTargetClass())
verifyClass(IPluggableIndex, DateRangeIndex) verifyClass(IPluggableIndex, self._getTargetClass())
verifyClass(ISortIndex, DateRangeIndex) verifyClass(ISortIndex, self._getTargetClass())
verifyClass(IUniqueValueIndex, DateRangeIndex) verifyClass(IUniqueValueIndex, self._getTargetClass())
def test_empty( self ): def test_empty( self ):
empty = DateRangeIndex( 'empty' ) empty = self._makeOne( 'empty' )
self.failUnless(empty.getEntryForObject( 1234 ) is None) self.failUnless(empty.getEntryForObject( 1234 ) is None)
empty.unindex_object( 1234 ) # shouldn't throw empty.unindex_object( 1234 ) # shouldn't throw
...@@ -111,18 +114,18 @@ class DRI_Tests( unittest.TestCase ): ...@@ -111,18 +114,18 @@ class DRI_Tests( unittest.TestCase ):
def test_retrieval( self ): def test_retrieval( self ):
work = DateRangeIndex( 'work', 'start', 'stop' ) index = self._makeOne( 'work', 'start', 'stop' )
for i in range( len( dummies ) ): for i in range( len( dummies ) ):
work.index_object( i, dummies[i] ) index.index_object( i, dummies[i] )
for i in range( len( dummies ) ): for i in range( len( dummies ) ):
self.assertEqual(work.getEntryForObject( i ), dummies[i].datum()) self.assertEqual(index.getEntryForObject( i ), dummies[i].datum())
for value in range( -1, 15 ): for value in range( -1, 15 ):
matches = matchingDummies( value ) matches = matchingDummies( value )
results, used = work._apply_index( { 'work' : value } ) results, used = index._apply_index( { 'work' : value } )
self.assertEqual(used, ( 'start', 'stop' )) self.assertEqual(used, ( 'start', 'stop' ))
self.assertEqual(len( matches ), len( results )) self.assertEqual(len( matches ), len( results ))
...@@ -130,43 +133,84 @@ class DRI_Tests( unittest.TestCase ): ...@@ -130,43 +133,84 @@ class DRI_Tests( unittest.TestCase ):
matches.sort( lambda x, y: cmp( x.name(), y.name() ) ) matches.sort( lambda x, y: cmp( x.name(), y.name() ) )
for result, match in map( None, results, matches ): for result, match in map( None, results, matches ):
self.assertEqual(work.getEntryForObject(result), match.datum()) self.assertEqual(index.getEntryForObject(result), match.datum())
def test_longdates( self ): def test_longdates( self ):
self.assertRaises(OverflowError, self._badlong ) self.assertRaises(OverflowError, self._badlong )
def _badlong(self): def _badlong(self):
work = DateRangeIndex ('work', 'start', 'stop' ) import sys
index = self._makeOne ('work', 'start', 'stop' )
bad = Dummy( 'bad', long(sys.maxint) + 1, long(sys.maxint) + 1 ) bad = Dummy( 'bad', long(sys.maxint) + 1, long(sys.maxint) + 1 )
work.index_object( 0, bad ) index.index_object( 0, bad )
def test_datetime(self): def test_datetime(self):
from datetime import datetime from datetime import datetime
from DateTime.DateTime import DateTime
from Products.PluginIndexes.DateIndex.tests.test_DateIndex \
import _getEastern
before = datetime(2009, 7, 11, 0, 0, tzinfo=_getEastern())
start = datetime(2009, 7, 13, 5, 15, tzinfo=_getEastern())
between = datetime(2009, 7, 13, 5, 45, tzinfo=_getEastern())
stop = datetime(2009, 7, 13, 6, 30, tzinfo=_getEastern())
after = datetime(2009, 7, 14, 0, 0, tzinfo=_getEastern())
dummy = Dummy('test', start, stop)
index = self._makeOne( 'work', 'start', 'stop' )
index.index_object(0, dummy)
self.assertEqual(index.getEntryForObject(0),
(DateTime(start).millis() / 60000,
DateTime(stop).millis() / 60000))
results, used = index._apply_index( { 'work' : before } )
self.assertEqual(len(results), 0)
results, used = index._apply_index( { 'work' : start } )
self.assertEqual(len(results), 1)
results, used = index._apply_index( { 'work' : between } )
self.assertEqual(len(results), 1)
results, used = index._apply_index( { 'work' : stop } )
self.assertEqual(len(results), 1)
results, used = index._apply_index( { 'work' : after } )
self.assertEqual(len(results), 0)
def test_datetime_naive_timezone(self):
from datetime import datetime
from DateTime.DateTime import DateTime
from Products.PluginIndexes.DateIndex.DateIndex import Local
before = datetime(2009, 7, 11, 0, 0) before = datetime(2009, 7, 11, 0, 0)
start = datetime(2009, 7, 13, 5, 15) start = datetime(2009, 7, 13, 5, 15)
start_local = datetime(2009, 7, 13, 5, 15, tzinfo=Local)
between = datetime(2009, 7, 13, 5, 45) between = datetime(2009, 7, 13, 5, 45)
stop = datetime(2009, 7, 13, 6, 30) stop = datetime(2009, 7, 13, 6, 30)
stop_local = datetime(2009, 7, 13, 6, 30, tzinfo=Local)
after = datetime(2009, 7, 14, 0, 0) after = datetime(2009, 7, 14, 0, 0)
dummy = Dummy('test', start, stop) dummy = Dummy('test', start, stop)
work = DateRangeIndex( 'work', 'start', 'stop' ) index = self._makeOne( 'work', 'start', 'stop' )
work.index_object(0, dummy) index.index_object(0, dummy)
self.assertEqual(work.getEntryForObject(0), (20790915, 20790990)) self.assertEqual(index.getEntryForObject(0),
(DateTime(start_local).millis() / 60000,
DateTime(stop_local).millis() / 60000))
results, used = work._apply_index( { 'work' : before } ) results, used = index._apply_index( { 'work' : before } )
self.assertEqual(len(results), 0) self.assertEqual(len(results), 0)
results, used = work._apply_index( { 'work' : start } ) results, used = index._apply_index( { 'work' : start } )
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
results, used = work._apply_index( { 'work' : between } ) results, used = index._apply_index( { 'work' : between } )
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
results, used = work._apply_index( { 'work' : stop } ) results, used = index._apply_index( { 'work' : stop } )
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
results, used = work._apply_index( { 'work' : after } ) results, used = index._apply_index( { 'work' : after } )
self.assertEqual(len(results), 0) self.assertEqual(len(results), 0)
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment