[Osaas-commits] r12 - in trunk: . server/osaas server/test
scm-commit@wald.intevation.org
scm-commit at wald.intevation.org
Tue Aug 28 15:56:11 CEST 2007
Author: bh
Date: 2007-08-28 15:56:11 +0200 (Tue, 28 Aug 2007)
New Revision: 12
Added:
trunk/server/test/test_dbbackend.py
Modified:
trunk/ChangeLog
trunk/server/osaas/dbbackend.py
trunk/server/osaas/server.py
Log:
* server/osaas/dbbackend.py (DBField, Insertion, DBBackend): New
classes implementing the backend.
(dbhandler): Removed this dummy implementation
* server/osaas/server.py (OSAASServer.__init__): Do not reference
the removed dbhandler function anymore.
* server/test/test_dbbackend.py: New. tests for the db backend
Modified: trunk/ChangeLog
===================================================================
--- trunk/ChangeLog 2007-08-28 12:55:37 UTC (rev 11)
+++ trunk/ChangeLog 2007-08-28 13:56:11 UTC (rev 12)
@@ -1,5 +1,16 @@
2007-08-28 Bernhard Herzog <bh at intevation.de>
+ * server/osaas/dbbackend.py (DBField, Insertion, DBBackend): New
+ classes implementing the backend.
+ (dbhandler): Removed this dummy implementation
+
+ * server/osaas/server.py (OSAASServer.__init__): Do not reference
+ the removed dbhandler function anymore.
+
+ * server/test/test_dbbackend.py: New. tests for the db backend
+
+2007-08-28 Bernhard Herzog <bh at intevation.de>
+
* server/osaas/run.py (OSAASProgramMixin.add_db_options): New
method to add database options
(OSAASServerProgram.create_option_parser): Call add_db_options
Modified: trunk/server/osaas/dbbackend.py
===================================================================
--- trunk/server/osaas/dbbackend.py 2007-08-28 12:55:37 UTC (rev 11)
+++ trunk/server/osaas/dbbackend.py 2007-08-28 13:56:11 UTC (rev 12)
@@ -7,5 +7,85 @@
"""Database backend of OSAAS"""
-def dbhandler(record):
- pass
+try:
+ import cx_Oracle as dbapi
+except ImportError:
+ class dbapi(object):
+
+ def connect(*args):
+ raise RuntimeError("Trying to call DBAPI methods"
+ " without a DBAPI module")
+ connect = staticmethod(connect)
+
+class DBField(object):
+
+ def __init__(self, parameter, column, column_type):
+ self.parameter = parameter
+ self.column = column
+ self.column_type = column_type
+
+
+class Insertion(object):
+
+ def __init__(self, table, fields):
+ self.table = table
+ self.fields = fields
+
+
+class DBBackend(object):
+
+ def __init__(self, connection_parameters, rules):
+ self.connection = None
+ self.connection_parameters = connection_parameters
+ self.rules = rules
+
+ def ensure_connection(self):
+ if self.connection is None:
+ self.connection = self.connect(*self.connection_parameters)
+
+ def connect(self, *args):
+ return dbapi.connect(*args)
+
+ def handle_record(self, record):
+ self.ensure_connection()
+ rule = self.rules[0]
+ names = ", ".join([f.column for f in rule.fields])
+ statement = ("INSERT INTO %s (ID, %s) VALUES (%s.nextval, %s)"
+ % (rule.table, names, rule.table + "_seq",
+ ", ".join([":%s" % f.column for f in rule.fields])))
+ values = {}
+ for f in rule.fields:
+ values[f.column] = getattr(record, f.parameter)
+ cursor = self.connection.cursor()
+ try:
+ cursor.execute(statement, values)
+ except:
+ self.connection.rollback()
+ raise
+ else:
+ self.connection.commit()
+
+ def create_tables(self):
+ """Creates the tables and sequences for the rules"""
+ self.ensure_connection()
+ cursor = self.connection.cursor()
+ try:
+ for rule in self.rules:
+ cursor.execute("CREATE SEQUENCE %s" % (rule.table + "_seq"))
+
+ lines = []
+ lines.append("CREATE TABLE %s" % rule.table)
+ lines.append("(")
+ fieldlines = [" ID INT"]
+ for field in rule.fields:
+ fieldlines.append(" %s %s"
+ % (field.column, field.column_type))
+ lines.append(",\n".join(fieldlines))
+ lines.append(")")
+ statement = "\n".join(lines)
+ cursor.execute(statement)
+ except:
+ self.connection.rollback()
+ raise
+ else:
+ self.connection.commit()
Modified: trunk/server/osaas/server.py
===================================================================
--- trunk/server/osaas/server.py 2007-08-28 12:55:37 UTC (rev 11)
+++ trunk/server/osaas/server.py 2007-08-28 13:56:11 UTC (rev 12)
@@ -14,7 +14,6 @@
from osaas.http.httpserver import HTTPRequestHandler
from osaas.http.threadedserver import ThreadedHTTPServer
from osaas.http.threadpool import ThreadPool
-from dbbackend import dbhandler
from recordfilter import default_filter
from userdb import UserDB
from formparser import parse_formdata, FormParserException
@@ -116,7 +115,7 @@
class OSAASServer(ThreadedHTTPServer):
def __init__(self, serveraddress, RequestHandlerClass=OSAASRequestHandler,
- dbhandler=dbhandler, userdbfile=None,
+ dbhandler=lambda r: None, userdbfile=None,
**kw):
ThreadedHTTPServer.__init__(self, serveraddress, RequestHandlerClass,
**kw)
Added: trunk/server/test/test_dbbackend.py
===================================================================
--- trunk/server/test/test_dbbackend.py 2007-08-28 12:55:37 UTC (rev 11)
+++ trunk/server/test/test_dbbackend.py 2007-08-28 13:56:11 UTC (rev 12)
@@ -0,0 +1,94 @@
+# Copyright (C) 2007 by Intevation GmbH
+# Authors:
+# Bernhard Herzog <bh at intevation.de>
+#
+# This program is free software under the GPL (>=v2)
+# Read the file COPYING coming with the software for details.
+
+import unittest
+
+from osaas.dbbackend import DBBackend, Insertion, DBField
+from osaas.formparser import Record
+
+
+class MockConnection(object):
+
+ def __init__(self, testcase):
+ self.testcase = testcase
+
+ def cursor(self):
+ return MockCursor(self.testcase)
+
+ def commit(self):
+ self.testcase.method_called("Connection.commit")
+
+
+class MockCursor(object):
+
+ def __init__(self, testcase):
+ self.testcase = testcase
+
+ def execute(self, statement, *args, **kw):
+ self.testcase.failUnless(isinstance(statement, str))
+ self.testcase.method_called("Cursor.execute", statement,
+ *args, **kw)
+
+class TestBackend(DBBackend):
+
+ def __init__(self, testcase, *args):
+ self.testcase = testcase
+ DBBackend.__init__(self, *args)
+
+ def connect(self, *args):
+ self.testcase.assertEquals(args, self.connection_parameters)
+ return MockConnection(self.testcase)
+
+
+class TestDBBackend(unittest.TestCase):
+
+ def setUp(self):
+ self.method_log = []
+
+ def method_called(self, name, *args, **kw):
+ entry = [name]
+ if args:
+ entry.append(args)
+ if kw:
+ entry.append(args)
+ self.method_log.append(tuple(entry))
+
+ def test_create_tables(self):
+ backend = TestBackend(self,
+ ("someuser", "aPassword", "mydns"),
+ [Insertion("mytable",
+ [DBField("REQUEST", "req",
+ "character varying(11)")])])
+ backend.create_tables()
+
+ self.assertEquals(self.method_log,
+ [('Cursor.execute', ('CREATE SEQUENCE mytable_seq',)),
+ ("Cursor.execute", ("CREATE TABLE mytable\n"
+ "(\n"
+ " ID INT,\n"
+ " req character varying(11)\n"
+ ")",)),
+ ("Connection.commit",)])
+
+
+ def test_handle_record(self):
+ backend = TestBackend(self,
+ ("someuser", "aPassword", "mydns"),
+ [Insertion("mytable",
+ [DBField("REQUEST", "req",
+ "character varying(11)")])])
+ record = Record()
+ record.REQUEST = "SERVICE=WMS&REQUEST=GetMap"
+ backend.handle_record(record)
+
+ self.assertEquals(self.method_log,
+ [("Cursor.execute",
+ ("INSERT INTO mytable (ID, req)"
+ " VALUES (mytable_seq.nextval, :req)",
+ {"req": "SERVICE=WMS&REQUEST=GetMap"})),
+ ("Connection.commit",)])
+
Property changes on: trunk/server/test/test_dbbackend.py
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ native
More information about the Osaas-commits
mailing list