- Fix for PYFB-17

fdb/fbcore.py

@@ -1824,6 +1824,9 @@ def __Tuple2XSQLDA(self, xsqlda, parameters):
                 # Set the null flag whether sqlvar definition allows it or not,
                 # to give BEFORE triggers to act on value without
                 # our interference.
+                if (sqlvar.sqltype & 1) == 0:
+                    # NULLs were not allowed, so set it allowed or FB will complain
+                    sqlvar.sqltype += 1
                 sqlvar.sqlind = ctypes.pointer(ISC_SHORT(-1))
                 sqlvar.sqldata = None
             else:

test/testfdb.py

@@ -727,114 +727,43 @@ def send_events(command_list):
         events.close()
         assert repr(e) == "{'insert_1': 1}"
 
-class TestWork(unittest.TestCase):
+class TestBugs(unittest.TestCase):
     def setUp(self):
         self.cwd = os.getcwd()
         self.dbpath = os.path.join(self.cwd,'test')
-        self.dbfile = os.path.join(self.dbpath,'fbwork.fdb')
+        self.dbfile = os.path.join(self.dbpath,'fbbugs.fdb')
+        if os.path.exists(self.dbfile):
+            os.remove(self.dbfile)
         self.con = fdb.create_database("CREATE DATABASE '%s' USER 'SYSDBA' PASSWORD 'masterkey'" % self.dbfile)
     def tearDown(self):
         self.con.drop_database()
         self.con.close()
-    #def test_functional_fkey_unique_insert_07(self):
-        ## SetUp
-        #c = self.con.cursor()
-        #c.execute("""CREATE TABLE MASTER_TABLE (
-    #ID     INTEGER PRIMARY KEY,
-    #UF     INTEGER UNIQUE,
-    #INT_F  INTEGER
-#);""")
-        #c.execute("""CREATE TABLE DETAIL_TABLE (
-    #ID    INTEGER PRIMARY KEY,
-    #FKEY  INTEGER
-#);""")
-        #self.con.commit()
-        #c.execute("ALTER TABLE DETAIL_TABLE ADD CONSTRAINT FK_DETAIL_TABLE FOREIGN KEY (FKEY) REFERENCES MASTER_TABLE (UF);")
-        #self.con.commit()
-        #c.execute("INSERT INTO MASTER_TABLE (ID, UF, INT_F) VALUES (1, 1, 10);")
-        #self.con.commit()
-        #c.close()
-        ## Test
-        #TPB_master = (
-              #chr(fdb.isc_tpb_write)
-            #+ chr(fdb.isc_tpb_read_committed) + chr(fdb.isc_tpb_rec_version)
-            #+ chr(fdb.isc_tpb_nowait)
-                          #)
-        #TPB_detail = (
-              #chr(fdb.isc_tpb_write)
-            #+ chr(fdb.isc_tpb_read_committed) + chr(fdb.isc_tpb_rec_version)
-            #+ chr(fdb.isc_tpb_nowait)
-                          #)
-        #self.con.begin(tpb=TPB_master)
-        #c = self.con.cursor()
-        #c.execute('DELETE FROM MASTER_TABLE WHERE ID=1')        
-        #self.con.commit()
-        ##Create second connection for change detail table
-        #con_detail = fdb.connect(dsn=self.dbfile,user='SYSDBA',password='masterkey')        
+    def test_pyfb_17(self):
+        create_table = """
+        Create Table table1  (
+            ID Integer,
+            sort Integer NOT Null
+        );
+        """
 
-        #try:
-            #con_detail.begin(tpb=TPB_detail)
-            #cd = con_detail.cursor()
-            #cd.execute("INSERT INTO DETAIL_TABLE (ID, FKEY) VALUES (1,1)")
-            #con_detail.commit()
-        #except Exception, e:
-            ##for msg in e: print msg
-            #print e[0]   
-        #else:
-            #con_detail.close()
-    #def test_functional_trigger_database_connect_04(self):
-        ## SetUp
-        #cmds = ["create table LOG (ID integer, MSG varchar(100));",
-                #"create generator LOGID;",
-                #"create exception CONNECTERROR 'Exception in ON CONNECT trigger';",
-                #"create role TEST;",
-                #"grant TEST to PUBLIC;",
-                #"""create trigger LOG_BI for LOG active before insert position 0
-#as
-#begin
-  #if (new.ID is null) then
-    #new.ID = gen_id(LOGID,1);
-#end""", """create trigger ONCONNECT_1 on connect position 0
-#as
-#begin
-  #insert into LOG (MSG) values ('Connect T1 as ' || current_role);
-#end""","""create trigger ONCONNECT_2 on connect position 0
-#as
-#begin
-  #insert into LOG (MSG) values ('Connect T2 as ' || current_role);
-  #if (current_role ='TEST') then
-      #exception CONNECTERROR;
-#end""","""create trigger ONCONNECT_3 on connect position 20
-#as
-#begin
-  #insert into LOG (MSG) values ('Connect T3 as ' || current_role);
-#end""",'COMMIT']
-        #c = self.con.cursor()
-        #for cmd in cmds:
-            #if cmd == 'COMMIT':
-                #self.con.commit()
-            #else:
-                #c.execute(cmd)
-        #c.close()
-        #self.con.close()
-        ## Test
-        #self.con = fdb.connect(dsn=self.dbfile,user='SYSDBA',password='masterkey')
-        #try:
-            #con2 = fdb.connect(dsn=self.dbfile,user='SYSDBA',password='masterkey',role='TEST')
-        #except Exception,e:
-            #for msg in e: print msg
-        #else:
-            #con2.close()
-
-        #c = self.con.cursor()
-        #c.execute('select * from LOG')
-        #printData(c)        
-    def test_functional_intfunc_string_ascii_val_01(self):
-        # SetUp
-        c = self.con.cursor()
-        c.execute("select ascii_val('') from rdb$database")
-        c.fetchall()
+        create_trigger = """CREATE TRIGGER BIU_Trigger FOR table1
+        ACTIVE BEFORE INSERT OR UPDATE POSITION 0
+        as
+        begin
+          if (new.sort IS NULL) then
+          begin
+            new.sort = 1;
+          end
+        end
+        """
 
+        cur = self.con.cursor()
+        cur.execute(create_table)
+        cur.execute(create_trigger)
+        self.con.commit()
+        # PYFB-17: fails with fdb, passes with kinterbasdb
+        cur.execute('insert into table1 (ID, sort) values(1, ?)', (None, ))
+
 
 if __name__ == '__main__':
     unittest.main()