[pypy-svn] r69632 - in pypy/trunk/pypy/module/oracle: . test

[afa at codespeak.net]

Author: afa
Date: Wed Nov 25 13:24:06 2009
New Revision: 69632

Modified:
   pypy/trunk/pypy/module/oracle/__init__.py
   pypy/trunk/pypy/module/oracle/interp_lob.py
   pypy/trunk/pypy/module/oracle/interp_variable.py
   pypy/trunk/pypy/module/oracle/test/test_lobvar.py
Log:
Implement ExternalLob.trim() + tests for CLOB


Modified: pypy/trunk/pypy/module/oracle/__init__.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/__init__.py	(original)
+++ pypy/trunk/pypy/module/oracle/__init__.py	Wed Nov 25 13:24:06 2009
@@ -18,6 +18,7 @@
         'FIXED_CHAR': 'interp_variable.VT_FixedChar',
         'CURSOR': 'interp_variable.VT_Cursor',
         'BLOB': 'interp_variable.VT_BLOB',
+        'CLOB': 'interp_variable.VT_CLOB',
         'Variable': 'interp_variable.W_Variable',
         'Timestamp': 'interp_error.get(space).w_DateTimeType',
         'Date': 'interp_error.get(space).w_DateType',

Modified: pypy/trunk/pypy/module/oracle/interp_lob.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/interp_lob.py	(original)
+++ pypy/trunk/pypy/module/oracle/interp_lob.py	Wed Nov 25 13:24:06 2009
@@ -2,6 +2,9 @@
 from pypy.interpreter.typedef import TypeDef
 from pypy.interpreter.gateway import ObjSpace
 from pypy.interpreter.gateway import interp2app
+from pypy.interpreter.error import OperationError
+
+from pypy.module.oracle.interp_error import get
 
 class W_ExternalLob(Wrappable):
     def __init__(self, var, pos):
@@ -26,6 +29,11 @@
         return space.wrap(self.lobVar.getLength(space, self.pos))
     size.unwrap_spec=['self', ObjSpace]
 
+    def trim(self, space, newSize=0):
+        self._verify(space)
+        self.lobVar.trim(space, self.pos, newSize)
+    trim.unwrap_spec=['self', ObjSpace, int]
+
     def desc_str(self, space):
         return self.read(space, offset=1, amount=-1)
     desc_str.unwrap_spec=['self', ObjSpace]
@@ -36,6 +44,8 @@
                       unwrap_spec=W_ExternalLob.read.unwrap_spec),
     size = interp2app(W_ExternalLob.size,
                       unwrap_spec=W_ExternalLob.size.unwrap_spec),
+    trim = interp2app(W_ExternalLob.trim,
+                      unwrap_spec=W_ExternalLob.trim.unwrap_spec),
     __str__ = interp2app(W_ExternalLob.desc_str,
                          unwrap_spec=W_ExternalLob.desc_str.unwrap_spec),
     )

Modified: pypy/trunk/pypy/module/oracle/interp_variable.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/interp_variable.py	(original)
+++ pypy/trunk/pypy/module/oracle/interp_variable.py	Wed Nov 25 13:24:06 2009
@@ -958,18 +958,7 @@
 
     def setValueProc(self, space, pos, w_value):
         self.ensureTemporary(space, pos)
-
-        # trim the current value
-        status = roci.OCILobTrim(
-            self.connection.handle,
-            self.environment.errorHandle,
-            self.getDataptr(pos)[0],
-            0)
-        self.environment.checkForError(
-            status,
-            "LobVar_SetValue(): trim")
-
-        # set the value
+        self.trim(space, pos, 0)
         self.write(space, pos, w_value, 1)
 
     def getLength(self, space, pos):
@@ -988,6 +977,16 @@
         finally:
             lltype.free(lengthptr, flavor='raw')
 
+    def trim(self, space, pos, newSize):
+        status = roci.OCILobTrim(
+            self.connection.handle,
+            self.environment.errorHandle,
+            self.getDataptr(pos)[0],
+            newSize)
+        self.environment.checkForError(
+            status,
+            "LobVar_Trim()")
+
     def read(self, space, pos, offset, amount):
         # modify the arguments
         if offset <= 0:

Modified: pypy/trunk/pypy/module/oracle/test/test_lobvar.py
==============================================================================
--- pypy/trunk/pypy/module/oracle/test/test_lobvar.py	(original)
+++ pypy/trunk/pypy/module/oracle/test/test_lobvar.py	Wed Nov 25 13:24:06 2009
@@ -30,5 +30,31 @@
         assert lob.size() == len(longString)
         assert lob.read() == longString
 
-class AppTestLob(LobTests, OracleTestBase):
+    def test_trim(self):
+        inputType = getattr(oracle, self.lobType)
+
+        cur = self.cnx.cursor()
+        try:
+            cur.execute("drop table pypy_temp_lobtable")
+        except oracle.DatabaseError:
+            pass
+        cur.execute("create table pypy_temp_lobtable "
+                    "(lobcol %s)" % self.lobType)
+
+        longString = "X" * 75000
+        cur.setinputsizes(lob=inputType)
+        cur.execute("insert into pypy_temp_lobtable values (:lob)",
+                    lob=longString)
+        cur.execute("select lobcol from pypy_temp_lobtable")
+        lob, = cur.fetchone()
+        assert lob.size() == 75000
+        lob.trim(25000)
+        assert lob.size() == 25000
+        lob.trim()
+        assert lob.size() == 0
+
+class AppTestBlob(LobTests, OracleTestBase):
     lobType = "BLOB"
+
+class AppTestClob(LobTests, OracleTestBase):
+    lobType = "CLOB"