[Zope-CVS] SVN: psycopgda/ Since psycopgda seems to be the most
popular
Philipp von Weitershausen
philikon at philikon.de
Thu Aug 12 14:18:31 EDT 2004
Log message for revision 27069:
Since psycopgda seems to be the most popular
database connector for ZopeX3 right now, it makes
sense to develop it in svn together with ZopeX3
itself.
Changed:
A psycopgda/
A psycopgda/trunk/
A psycopgda/trunk/README.txt
A psycopgda/trunk/__init__.py
A psycopgda/trunk/adapter.py
A psycopgda/trunk/configure.zcml
A psycopgda/trunk/psycopgda-configure.zcml
A psycopgda/trunk/tests/
A psycopgda/trunk/tests/__init__.py
A psycopgda/trunk/tests/test_adapter.py
-=-
Added: psycopgda/trunk/README.txt
===================================================================
--- psycopgda/trunk/README.txt 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/README.txt 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1,79 @@
+=========
+psycopgda
+=========
+
+This file outlines the basics of using Zope3 with PostgreSQL via PsycopgDA.
+
+Installing PsycopgDA
+--------------------
+
+1. Check out the psycopgda package into a directory in your
+ PYTHONPATH. INSTANCE_HOME/lib/python or Zope3/src is usually the
+ most convenient place:
+
+
+ svn co svn://svn.zope.org/repos/main/psycopg
+
+
+2. Copy `psycopg-configure.zcml` to the `package-includes` directory
+ of your Zope instance.
+
+
+Creating Database Connections
+------------------------------
+
+It is time to add some connections. A connection in Zope 3 is
+registered as a utility.
+
+3. Open a web browser on your Zope root folder (http://localhost:8080/
+ if you use the default settings in zope.conf.in).
+
+4. Click on the 'Manage Site' action on the right side of the
+ screen. You should see a screen which reads 'Common Site Management
+ Tasks'
+
+5. Around the middle of that page, you should see a link named 'Add
+ Utility'. Click on it.
+
+6. Select 'Psycopg DA' and type in a name at the bottom of the page.
+
+7. Enter the database connection string. It looks like this:
+
+ dbi://username:password@host:port/databasename
+
+8. Click on the 'Add' button.
+
+9. You should be on a page which reads 'Add Database Connection
+ Registration'. There you can configure the permission needed to use
+ the database connection, the name of the registration and the
+ registration status. You can use any name for 'Register As' field,
+ as long as it doesn't clash with an existing one. Choose a
+ permission. Choose between 'Registered' and 'Active' for the
+ 'Registration Status'. Only one component of a kind can be 'Active'
+ at a time, so be careful.
+
+10. You should be redirected to the 'Edit' screen of the connection
+ utility.
+
+11. If you want to, you can go to the Test page and execute arbitrary
+ SQL queries to see whether the connection is working as expected.
+
+
+Using SQL Scripts
+-----------------
+
+You can create SQL Scripts in the content space. For example:
+
+12. Go to Zope root.
+
+13. Add an SQL script (you can use the Common Tasks box on the left,
+ or the Add action on the right).
+
+14. Click on the name of your new SQL script.
+
+15. Choose a connection name (the one you entered in step 29) from the
+ drop-down.
+
+16. Enter your query and click on the 'Save Changes' button.
+
+17. You can test the script in the -- surprise! -- Test page.
Added: psycopgda/trunk/__init__.py
===================================================================
--- psycopgda/trunk/__init__.py 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/__init__.py 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1 @@
+# make this directory a package
Added: psycopgda/trunk/adapter.py
===================================================================
--- psycopgda/trunk/adapter.py 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/adapter.py 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1,308 @@
+##############################################################################
+#
+# Copyright (c) 2002-2004 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""PostgreSQL Database Adapter for Zope 3
+
+$Id$
+"""
+from zope.app.rdb import ZopeDatabaseAdapter, parseDSN
+
+from datetime import date, time, datetime, timedelta
+import psycopg
+import re
+
+# These OIDs are taken from pg_types.h from PostgreSQL headers.
+# Unfortunatelly psycopg does not export them as constants, and
+# we cannot use psycopg.FOO.values because they overlap.
+DATE_OID = 1082
+TIME_OID = 1083
+TIMETZ_OID = 1266
+TIMESTAMP_OID = 1114
+TIMESTAMPTZ_OID = 1184
+INTERVAL_OID = 1186
+# The following ones are obsolete and we don't handle them
+#ABSTIME_OID = 702
+#RELTIME_OID = 703
+#TINTERVAL_OID = 704
+
+# Date/time parsing functions
+
+_dateFmt = re.compile(r"^(\d\d\d\d)-?([01]\d)-?([0-3]\d)$")
+
+def parse_date(s):
+ """Parses ISO-8601 compliant dates and returns a tuple (year, month,
+ day).
+
+ The following formats are accepted:
+ YYYY-MM-DD (extended format)
+ YYYYMMDD (basic format)
+ """
+ m = _dateFmt.match(s)
+ if m is None:
+ raise ValueError, 'invalid date string: %s' % s
+ year, month, day = m.groups()
+ return int(year), int(month), int(day)
+
+
+_timeFmt = re.compile(
+ r"^([0-2]\d)(?::?([0-5]\d)(?::?([0-5]\d)(?:[.,](\d+))?)?)?$")
+
+def parse_time(s):
+ """Parses ISO-8601 compliant times and returns a tuple (hour, minute,
+ second).
+
+ The following formats are accepted:
+ HH:MM:SS.ssss or HHMMSS.ssss
+ HH:MM:SS,ssss or HHMMSS,ssss
+ HH:MM:SS or HHMMSS
+ HH:MM or HHMM
+ HH
+ """
+ m = _timeFmt.match(s)
+ if m is None:
+ raise ValueError, 'invalid time string: %s' % s
+ hr, mn, sc, msc = m.groups(0)
+ if msc != 0:
+ sc = float("%s.%s" % (sc, msc))
+ else:
+ sc = int(sc)
+ return int(hr), int(mn), sc
+
+
+_tzFmt = re.compile(r"^([+-])([0-2]\d)(?::?([0-5]\d))?$")
+
+def parse_tz(s):
+ """Parses ISO-8601 timezones and returns the offset east of UTC in
+ minutes.
+
+ The following formats are accepted:
+ +/-HH:MM
+ +/-HHMM
+ +/-HH
+ Z (equivalent to +0000)
+ """
+ if s == 'Z':
+ return 0
+ m = _tzFmt.match(s)
+ if m is None:
+ raise ValueError, 'invalid time zone: %s' % s
+ d, hoff, moff = m.groups(0)
+ if d == "-":
+ return - int(hoff) * 60 - int(moff)
+ return int(hoff) * 60 + int(moff)
+
+
+_tzPos = re.compile(r"[Z+-]")
+
+def parse_timetz(s):
+ """Parses ISO-8601 compliant times that may include timezone information
+ and returns a tuple (hour, minute, second, tzoffset).
+
+ tzoffset is the offset east of UTC in minutes. It will be None if s does
+ not include time zone information.
+
+ Formats accepted are those listed in the descriptions of parse_time() and
+ parse_tz(). Time zone should immediatelly follow time without intervening
+ spaces.
+ """
+ m = _tzPos.search(s)
+ if m is None:
+ return parse_time(s) + (None,)
+ pos = m.start()
+ return parse_time(s[:pos]) + (parse_tz(s[pos:]),)
+
+
+_datetimeFmt = re.compile(r"[T ]")
+
+def _split_datetime(s):
+ """Split date and time parts of ISO-8601 compliant timestamp and
+ return a tuple (date, time).
+
+ ' ' or 'T' used to separate date and time parts.
+ """
+ m = _datetimeFmt.search(s)
+ if m is None:
+ raise ValueError, 'time part of datetime missing: %s' % s
+ pos = m.start()
+ return s[:pos], s[pos + 1:]
+
+
+def parse_datetime(s):
+ """Parses ISO-8601 compliant timestamp and returns a tuple (year, month,
+ day, hour, minute, second).
+
+ Formats accepted are those listed in the descriptions of parse_date() and
+ parse_time() with ' ' or 'T' used to separate date and time parts.
+ """
+ dt, tm = _split_datetime(s)
+ return parse_date(dt) + parse_time(tm)
+
+
+def parse_datetimetz(s):
+ """Parses ISO-8601 compliant timestamp that may include timezone
+ information and returns a tuple (year, month, day, hour, minute, second,
+ tzoffset).
+
+ tzoffset is the offset east of UTC in minutes. It will be None if s does
+ not include time zone information.
+
+ Formats accepted are those listed in the descriptions of parse_date() and
+ parse_timetz() with ' ' or 'T' used to separate date and time parts.
+ """
+ dt, tm = _split_datetime(s)
+ return parse_date(dt) + parse_timetz(tm)
+
+
+def parse_interval(s):
+ """Parses PostgreSQL interval notation and returns a tuple (years, months,
+ days, hours, minutes, seconds).
+
+ Values accepted:
+ interval ::= date
+ | time
+ | date time
+ date ::= date_comp
+ | date date_comp
+ date_comp ::= 1 'day'
+ | number 'days'
+ | 1 'month'
+ | number 'months'
+ | 1 'year'
+ | number 'years'
+ time ::= number ':' number
+ | number ':' number ':' number
+ | number ':' number ':' number '.' fraction
+ """
+ years = months = days = 0
+ hours = minutes = seconds = 0
+ elements = s.split()
+ for i in range(0, len(elements) - 1, 2):
+ count, unit = elements[i:i+2]
+ if unit == 'day' and count == '1':
+ days += 1
+ elif unit == 'days':
+ days += int(count)
+ elif unit == 'month' and count == '1':
+ months += 1
+ elif unit == 'months':
+ months += int(count)
+ elif unit == 'year' and count == '1':
+ years += 1
+ elif unit == 'years':
+ years += int(count)
+ else:
+ raise ValueError, 'unknown time interval %s %s' % (count, unit)
+ if len(elements) % 2 == 1:
+ hours, minutes, seconds = parse_time(elements[-1])
+ return (years, months, days, hours, minutes, seconds)
+
+
+# Type conversions
+def _conv_date(s):
+ if s:
+ return date(*parse_date(s))
+
+def _conv_time(s):
+ if s:
+ hr, mn, sc = parse_time(s)
+ sc, micro = divmod(sc, 1.0)
+ micro = round(micro * 1000000)
+ return time(hr, mn, int(sc), int(micro))
+
+def _conv_timetz(s):
+ if s:
+ from zope.app.datetimeutils import tzinfo
+ hr, mn, sc, tz = parse_timetz(s)
+ sc, micro = divmod(sc, 1.0)
+ micro = round(micro * 1000000)
+ if tz: tz = tzinfo(tz)
+ return time(hr, mn, int(sc), int(micro), tz)
+
+def _conv_timestamp(s):
+ if s:
+ y, m, d, hr, mn, sc = parse_datetime(s)
+ sc, micro = divmod(sc, 1.0)
+ micro = round(micro * 1000000)
+ return datetime(y, m, d, hr, mn, int(sc), int(micro))
+
+def _conv_timestamptz(s):
+ if s:
+ from zope.app.datetimeutils import tzinfo
+ y, m, d, hr, mn, sc, tz = parse_datetimetz(s)
+ sc, micro = divmod(sc, 1.0)
+ micro = round(micro * 1000000)
+ if tz: tz = tzinfo(tz)
+ return datetime(y, m, d, hr, mn, int(sc), int(micro), tz)
+
+def _conv_interval(s):
+ if s:
+ y, m, d, hr, mn, sc = parse_interval(s)
+ if (y, m) != (0, 0):
+ # XXX: Currently there's no way to represent years and months as
+ # timedeltas
+ return s
+ else:
+ return timedelta(days=d, hours=hr, minutes=mn, seconds=sc)
+
+
+# User-defined types
+DATE = psycopg.new_type((DATE_OID,), "ZDATE", _conv_date)
+TIME = psycopg.new_type((TIME_OID,), "ZTIME", _conv_time)
+TIMETZ = psycopg.new_type((TIMETZ_OID,), "ZTIMETZ", _conv_timetz)
+TIMESTAMP = psycopg.new_type((TIMESTAMP_OID,), "ZTIMESTAMP", _conv_timestamp)
+TIMESTAMPTZ = psycopg.new_type((TIMESTAMPTZ_OID,), "ZTIMESTAMPTZ",
+ _conv_timestamptz)
+INTERVAL = psycopg.new_type((INTERVAL_OID,), "ZINTERVAL", _conv_interval)
+
+
+dsn2option_mapping = {'host': 'host',
+ 'port': 'port',
+ 'dbname': 'dbname',
+ 'username': 'user',
+ 'password': 'password'}
+
+
+class PsycopgAdapter(ZopeDatabaseAdapter):
+ """A PsycoPG adapter for Zope3.
+
+ The following type conversions are performed:
+
+ DATE -> datetime.date
+ TIME -> datetime.time
+ TIMETZ -> datetime.time
+ TIMESTAMP -> datetime.datetime
+ TIMESTAMPTZ -> datetime.datetime
+
+ XXX: INTERVAL cannot be represented exactly as datetime.timedelta since
+ it might be something like '1 month', which is a variable number of days.
+ """
+
+ def _connection_factory(self):
+ """Create a Psycopg DBI connection based on the DSN"""
+ conn_info = parseDSN(self.dsn)
+ conn_list = []
+ for dsnname, optname in dsn2option_mapping.iteritems():
+ if conn_info[dsnname]:
+ conn_list.append('%s=%s' % (optname, conn_info[dsnname]))
+ conn_str = ' '.join(conn_list)
+ self._registerTypes()
+ return psycopg.connect(conn_str)
+
+ def _registerTypes(self):
+ """Register type conversions for psycopg"""
+ psycopg.register_type(DATE)
+ psycopg.register_type(TIME)
+ psycopg.register_type(TIMETZ)
+ psycopg.register_type(TIMESTAMP)
+ psycopg.register_type(TIMESTAMPTZ)
+ psycopg.register_type(INTERVAL)
Added: psycopgda/trunk/configure.zcml
===================================================================
--- psycopgda/trunk/configure.zcml 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/configure.zcml 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1,42 @@
+<configure
+ xmlns="http://namespaces.zope.org/zope"
+ xmlns:browser="http://namespaces.zope.org/browser"
+ i18n_domain="psycopgda">
+
+ <content class=".adapter.PsycopgAdapter">
+ <factory id="zope.da.PsycopgDA" />
+ <require
+ permission="zope.Public"
+ interface="zope.app.rdb.interfaces.IZopeDatabaseAdapter"
+ />
+ </content>
+
+ <browser:addform
+ name="AddPsycopgDA"
+ schema="zope.app.rdb.interfaces.IZopeDatabaseAdapter"
+ label="Add Psycopg (PostGreSQL) Database Adapter"
+ content_factory=".adapter.PsycopgAdapter"
+ arguments="dsn"
+ fields="dsn"
+ permission="zope.ManageContent"
+ />
+
+ <!-- Menu entry for "add utility" menu -->
+ <browser:addMenuItem
+ class=".adapter.PsycopgAdapter"
+ title="Psycopg DA"
+ description="A PostgreSQL Database Adapter using the Psycopg driver"
+ permission="zope.ManageApplication"
+ view="AddPsycopgDA"
+ />
+
+ <!-- Menu entry for "add connection" menu -->
+ <browser:menuItem
+ menu="add_connection"
+ for="zope.app.container.interfaces.IAdding"
+ title="Psycopg DA"
+ description="A PostgreSQL Database Adapter using the Psycopg driver"
+ action="AddPsycopgDA"
+ />
+
+</configure>
Added: psycopgda/trunk/psycopgda-configure.zcml
===================================================================
--- psycopgda/trunk/psycopgda-configure.zcml 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/psycopgda-configure.zcml 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1 @@
+<include package="psycopgda" />
Added: psycopgda/trunk/tests/__init__.py
===================================================================
--- psycopgda/trunk/tests/__init__.py 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/tests/__init__.py 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1 @@
+# make this directory a package
Added: psycopgda/trunk/tests/test_adapter.py
===================================================================
--- psycopgda/trunk/tests/test_adapter.py 2004-08-12 18:11:23 UTC (rev 27068)
+++ psycopgda/trunk/tests/test_adapter.py 2004-08-12 18:18:31 UTC (rev 27069)
@@ -0,0 +1,352 @@
+##############################################################################
+#
+# Copyright (c) 2002-2004 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+"""Unit tests for PsycopgDA.
+
+$Id$
+"""
+from unittest import TestCase, TestSuite, main, makeSuite
+from datetime import tzinfo, timedelta
+import psycopg
+
+class Stub:
+
+ def __init__(self, **kw):
+ self.__dict__.update(kw)
+
+class TZStub(tzinfo):
+
+ def __init__(self, h, m):
+ self.offset = h * 60 + m
+
+ def utcoffset(self, dt):
+ return timedelta(minutes=self.offset)
+
+ def dst(self, dt):
+ return 0
+
+ def tzname(self, dt):
+ return ''
+
+ def __repr__(self):
+ return 'tzinfo(%d)' % self.offset
+
+ def __reduce__(self):
+ return type(self), (), self.__dict__
+
+class PsycopgStub:
+
+ __shared_state = {} # 'Borg' design pattern
+
+ DATE = psycopg.DATE
+ TIME = psycopg.TIME
+ DATETIME = psycopg.DATETIME
+ INTERVAL = psycopg.INTERVAL
+
+ def __init__(self):
+ self.__dict__ = self.__shared_state
+ self.types = {}
+
+ def connect(self, connection_string):
+ self.last_connection_string = connection_string
+
+ def new_type(self, values, name, converter):
+ return Stub(name=name, values=values)
+
+ def register_type(self, type):
+ for typeid in type.values:
+ self.types[typeid] = type
+
+
+class TestPsycopgTypeConversion(TestCase):
+
+ def test_conv_date(self):
+ from psycopgda.adapter import _conv_date
+ from datetime import date
+ self.assertEquals(_conv_date(''), None)
+ self.assertEquals(_conv_date('2001-03-02'), date(2001, 3, 2))
+
+ def test_conv_time(self):
+ from psycopgda.adapter import _conv_time
+ from datetime import time
+ self.assertEquals(_conv_time(''), None)
+ self.assertEquals(_conv_time('23:17:57'),
+ time(23, 17, 57))
+ self.assertEquals(_conv_time('23:17:57.037'),
+ time(23, 17, 57, 37000))
+
+ def test_conv_timetz(self):
+ from psycopgda.adapter import _conv_timetz
+ from datetime import time
+ self.assertEquals(_conv_timetz(''), None)
+ self.assertEquals(_conv_timetz('12:44:01+01:00'),
+ time(12, 44, 01, 0, TZStub(1,0)))
+ self.assertEquals(_conv_timetz('12:44:01.037-00:30'),
+ time(12, 44, 01, 37000, TZStub(0,-30)))
+
+ def test_conv_timestamp(self):
+ from psycopgda.adapter import _conv_timestamp
+ from datetime import datetime
+ self.assertEquals(_conv_timestamp(''), None)
+ self.assertEquals(_conv_timestamp('2001-03-02 12:44:01'),
+ datetime(2001, 3, 2, 12, 44, 01))
+ self.assertEquals(_conv_timestamp('2001-03-02 12:44:01.037'),
+ datetime(2001, 3, 2, 12, 44, 01, 37000))
+ self.assertEquals(_conv_timestamp('2001-03-02 12:44:01.000001'),
+ datetime(2001, 3, 2, 12, 44, 01, 1))
+
+ def test_conv_timestamptz(self):
+ from psycopgda.adapter import _conv_timestamptz as c
+ from datetime import datetime
+ self.assertEquals(c(''), None)
+
+ self.assertEquals(c('2001-03-02 12:44:01+01:00'),
+ datetime(2001, 3, 2, 12, 44, 01, 0, TZStub(1,0)))
+ self.assertEquals(c('2001-03-02 12:44:01.037-00:30'),
+ datetime(2001, 3, 2, 12, 44, 01, 37000, TZStub(0,-30)))
+ self.assertEquals(c('2001-03-02 12:44:01.000001+12:00'),
+ datetime(2001, 3, 2, 12, 44, 01, 1, TZStub(12,0)))
+ self.assertEquals(c('2001-06-25 12:14:00-07'),
+ datetime(2001, 6, 25, 12, 14, 00, 0, TZStub(-7,0)))
+
+ def test_conv_interval(self):
+ from psycopgda.adapter import _conv_interval as c
+ from datetime import timedelta
+ self.assertEquals(c(''), None)
+ self.assertEquals(c('01:00'), timedelta(hours=1))
+ self.assertEquals(c('00:15'), timedelta(minutes=15))
+ self.assertEquals(c('00:00:47'), timedelta(seconds=47))
+ self.assertEquals(c('00:00:00.037'), timedelta(microseconds=37000))
+ self.assertEquals(c('00:00:00.111111'), timedelta(microseconds=111111))
+ self.assertEquals(c('1 day'), timedelta(days=1))
+ self.assertEquals(c('2 days'), timedelta(days=2))
+ self.assertEquals(c('374 days'), timedelta(days=374))
+ self.assertEquals(c('2 days 03:20:15.123456'),
+ timedelta(days=2, hours=3, minutes=20,
+ seconds=15, microseconds=123456))
+ # XXX There's a problem with years and months. Currently timedelta
+ # cannot represent them accurately
+ self.assertEquals(c('1 month'), '1 month')
+ self.assertEquals(c('2 months'), '2 months')
+ self.assertEquals(c('1 year'), '1 year')
+ self.assertEquals(c('3 years'), '3 years')
+ # Later we might be able to use
+ ## self.assertEquals(c('1 month'), timedelta(months=1))
+ ## self.assertEquals(c('2 months'), timedelta(months=2))
+ ## self.assertEquals(c('1 year'), timedelta(years=1))
+ ## self.assertEquals(c('3 years'), timedelta(years=3))
+
+ self.assertRaises(ValueError, c, '2 day')
+ self.assertRaises(ValueError, c, '2days')
+ self.assertRaises(ValueError, c, '123')
+
+
+class TestPsycopgAdapter(TestCase):
+
+ def setUp(self):
+ import psycopgda.adapter as adapter
+ self.real_psycopg = adapter.psycopg
+ adapter.psycopg = self.psycopg_stub = PsycopgStub()
+
+ def tearDown(self):
+ import psycopgda.adapter as adapter
+ adapter.psycopg = self.real_psycopg
+
+ def test_connection_factory(self):
+ from psycopgda.adapter import PsycopgAdapter
+ a = PsycopgAdapter('dbi://username:password@hostname:port/dbname;junk=ignored')
+ c = a._connection_factory()
+ args = self.psycopg_stub.last_connection_string.split()
+ args.sort()
+ self.assertEquals(args, ['dbname=dbname', 'host=hostname',
+ 'password=password', 'port=port',
+ 'user=username'])
+
+ def test_registerTypes(self):
+ import psycopgda.adapter as adapter
+ from psycopgda.adapter import PsycopgAdapter
+ a = PsycopgAdapter('dbi://')
+ a._registerTypes()
+ for typename in ('DATE', 'TIME', 'TIMETZ', 'TIMESTAMP',
+ 'TIMESTAMPTZ', 'INTERVAL'):
+ typeid = getattr(adapter, '%s_OID' % typename)
+ result = self.psycopg_stub.types.get(typeid, None)
+ if not result:
+ # comparing None with psycopg.type object segfaults
+ self.fail("did not register %s (%d): got None, not Z%s"
+ % (typename, typeid, typename))
+ else:
+ result_name = getattr(result, 'name', 'None')
+ self.assertEquals(result, getattr(adapter, typename),
+ "did not register %s (%d): got %s, not Z%s"
+ % (typename, typeid, result_name, typename))
+
+
+class TestISODateTime(TestCase):
+
+ # Test if date/time parsing functions accept a sensible subset of ISO-8601
+ # compliant date/time strings.
+ #
+ # Resources:
+ # http://www.cl.cam.ac.uk/~mgk25/iso-time.html
+ # http://www.mcs.vuw.ac.nz/technical/software/SGML/doc/iso8601/ISO8601.html
+ # http://www.w3.org/TR/NOTE-datetime
+ # http://www.ietf.org/rfc/rfc3339.txt
+
+ basic_dates = (('20020304', (2002, 03, 04)),
+ ('20000229', (2000, 02, 29)))
+
+ extended_dates = (('2002-03-04', (2002, 03, 04)),
+ ('2000-02-29', (2000, 02, 29)))
+
+ basic_times = (('12', (12, 0, 0)),
+ ('1234', (12, 34, 0)),
+ ('123417', (12, 34, 17)),
+ ('123417.5', (12, 34, 17.5)),
+ ('123417,5', (12, 34, 17.5)))
+
+ extended_times = (('12', (12, 0, 0)),
+ ('12:34', (12, 34, 0)),
+ ('12:34:17', (12, 34, 17)),
+ ('12:34:17.5', (12, 34, 17.5)),
+ ('12:34:17,5', (12, 34, 17.5)))
+
+ basic_tzs = (('Z', 0),
+ ('+02', 2*60),
+ ('+1130', 11*60+30),
+ ('-05', -5*60),
+ ('-0030', -30))
+
+ extended_tzs = (('Z', 0),
+ ('+02', 2*60),
+ ('+11:30', 11*60+30),
+ ('-05', -5*60),
+ ('-00:30', -30))
+
+ time_separators = (' ', 'T')
+
+ bad_dates = ('', 'foo', 'XXXXXXXX', 'XXXX-XX-XX', '2001-2-29',
+ '1990/13/14')
+
+ bad_times = ('', 'foo', 'XXXXXX', '12:34,5', '12:34:56,')
+
+ bad_timetzs = ('12+12 34', '15:45 +1234', '18:00-12:34:56', '18:00+123', '18:00Q')
+
+ bad_datetimes = ('', 'foo', '2002-03-0412:33')
+
+ bad_datetimetzs = ('', 'foo', '2002-03-04T12:33 +1200')
+
+ exception_type = ValueError
+
+ # We need the following funcions:
+ # parse_date -> (year, month, day)
+ # parse_time -> (hour, minute, second)
+ # parse_timetz -> (hour, minute, second, tzoffset)
+ # parse_datetime -> (year, month, day, hour, minute, second)
+ # parse_datetimetz -> (year, month, day, hour, minute, second, tzoffset)
+ # second can be a float, all other values are ints
+ # tzoffset is offset in minutes east of UTC
+
+ def setUp(self):
+ from psycopgda.adapter import parse_date, parse_time, \
+ parse_timetz, parse_datetime, parse_datetimetz
+ self.parse_date = parse_date
+ self.parse_time = parse_time
+ self.parse_timetz = parse_timetz
+ self.parse_datetime = parse_datetime
+ self.parse_datetimetz = parse_datetimetz
+
+ def test_basic_date(self):
+ for s, d in self.basic_dates:
+ self.assertEqual(self.parse_date(s), d)
+
+ def test_extended_date(self):
+ for s, d in self.extended_dates:
+ self.assertEqual(self.parse_date(s), d)
+
+ def test_bad_date(self):
+ for s in self.bad_dates:
+ self.assertRaises(self.exception_type, self.parse_date, s)
+
+ def test_basic_time(self):
+ for s, t in self.basic_times:
+ self.assertEqual(self.parse_time(s), t)
+
+ def test_extended_time(self):
+ for s, t in self.extended_times:
+ self.assertEqual(self.parse_time(s), t)
+
+ def test_bad_time(self):
+ for s in self.bad_times:
+ self.assertRaises(self.exception_type, self.parse_time, s)
+
+ def test_basic_timetz(self):
+ for s, t in self.basic_times:
+ for tz, off in self.basic_tzs:
+ self.assertEqual(self.parse_timetz(s+tz), t + (off,))
+
+ def test_extended_timetz(self):
+ for s, t in self.extended_times:
+ for tz, off in self.extended_tzs:
+ self.assertEqual(self.parse_timetz(s+tz), t + (off,))
+
+ def test_bad_timetzs(self):
+ for s in self.bad_timetzs:
+ self.assertRaises(self.exception_type, self.parse_timetz, s)
+
+ def test_basic_datetime(self):
+ for ds, d in self.basic_dates:
+ for ts, t in self.basic_times:
+ for sep in self.time_separators:
+ self.assertEqual(self.parse_datetime(ds+sep+ts), d + t)
+
+ def test_extended_datetime(self):
+ for ds, d in self.extended_dates:
+ for ts, t in self.extended_times:
+ for sep in self.time_separators:
+ self.assertEqual(self.parse_datetime(ds+sep+ts), d + t)
+
+ def test_bad_datetimes(self):
+ for s in self.bad_datetimes:
+ self.assertRaises(self.exception_type, self.parse_datetime, s)
+
+ def test_basic_datetimetz(self):
+ for ds, d in self.basic_dates:
+ for ts, t in self.basic_times:
+ for tz, off in self.basic_tzs:
+ for sep in self.time_separators:
+ self.assertEqual(self.parse_datetimetz(ds+sep+ts+tz),
+ d + t + (off,))
+
+ def test_extended_datetimetz(self):
+ for ds, d in self.extended_dates:
+ for ts, t in self.extended_times:
+ for tz, off in self.extended_tzs:
+ for sep in self.time_separators:
+ self.assertEqual(self.parse_datetimetz(ds+sep+ts+tz),
+ d + t + (off,))
+
+ def test_bad_datetimetzs(self):
+ for s in self.bad_datetimetzs:
+ self.assertRaises(self.exception_type, self.parse_datetimetz, s)
+
+
+def test_suite():
+ return TestSuite((
+ makeSuite(TestPsycopgTypeConversion),
+ makeSuite(TestPsycopgAdapter),
+ makeSuite(TestISODateTime),
+ ))
+
+if __name__=='__main__':
+ main(defaultTest='test_suite')
More information about the Zope-CVS
mailing list