Commits

Mike Bayer  committed 916ce82

- fixed compiler bug in mssql
- marked as unsupported for mssql all two-phase and nested transcation tests
- marked as unsupported for mssql various transactional/session tests which require two connections looking at uncommitted/external data at the same time (ms-sql cant handle it)
- put better explicit closeout step in unitofwork.py tests to appease ms-sqls hard locking

  • Participants
  • Parent commits 97d8137

Comments (0)

Files changed (6)

File lib/sqlalchemy/ansisql.py

         if stack:
             self.stack.append(stack)
         try:
-            return self.traverse_single(obj, **kwargs)
+            x = self.traverse_single(obj, **kwargs)
+            if x is None:
+                raise "hi " + repr(obj)
+            return x
         finally:
             if stack:
                 self.stack.pop(-1)

File lib/sqlalchemy/databases/mssql.py

                           }
     def visit_function(self, func):
         func.name = self.function_rewrites.get(func.name, func.name)
-        super(MSSQLCompiler, self).visit_function(func)            
+        return super(MSSQLCompiler, self).visit_function(func)            
 
     def for_update_clause(self, select):
         # "FOR UPDATE" is only allowed on "DECLARE CURSOR" which SQLAlchemy doesn't use

File test/dialect/alltests.py

         'dialect.mysql',
         'dialect.postgres',
         'dialect.oracle',
+        'dialect.mssql',
         )
     alltests = unittest.TestSuite()
     for name in modules_to_test:

File test/dialect/mssql.py

+import testbase
+import re
+from sqlalchemy import *
+from sqlalchemy.databases import mssql
+from testlib import *
+
+msdialect = mssql.MSSQLDialect()
+
+# TODO: migrate all MS-SQL tests here
+
+class CompileTest(AssertMixin):
+    def _test(self, statement, expected, **params):
+        if len(params):
+            res = str(statement.compile(dialect=msdialect, parameters=params))
+        else:
+            res = str(statement.compile(dialect=msdialect))
+        res = re.sub(r'\n', '', res)
+
+        assert res == expected, res
+        
+    def test_insert(self):
+        t = table('sometable', column('somecolumn'))
+        self._test(t.insert(), "INSERT INTO sometable (somecolumn) VALUES (:somecolumn)")
+
+    def test_update(self):
+        t = table('sometable', column('somecolumn'))
+	self._test(t.update(t.c.somecolumn==7), "UPDATE sometable SET somecolumn=:somecolumn WHERE sometable.somecolumn = :sometable_somecolumn", somecolumn=10)
+
+    def test_count(self):
+        t = table('sometable', column('somecolumn'))
+	self._test(t.count(), "SELECT count(sometable.somecolumn) AS tbl_row_count FROM sometable")
+    
+if __name__ == "__main__":
+    testbase.main()

File test/engine/transaction.py

         assert len(result.fetchall()) == 0
         connection.close()
     
-    @testing.unsupported('sqlite', 'mssql') # TEMP: test causes mssql to hang
+    @testing.supported('postgres', 'mysql', 'oracle')
     @testing.exclude('mysql', '<', (5, 0, 3))
     def testnestedsubtransactionrollback(self):
         connection = testbase.db.connect()
         )
         connection.close()
 
-    @testing.unsupported('sqlite', 'mssql') # TEMP: test causes mssql to hang
+    @testing.supported('postgres', 'mysql', 'oracle')
     @testing.exclude('mysql', '<', (5, 0, 3))
     def testnestedsubtransactioncommit(self):
         connection = testbase.db.connect()
         )
         connection.close()
 
-    @testing.unsupported('sqlite', 'mssql') # TEMP: test causes mssql to hang
+    @testing.supported('postgres', 'mysql', 'oracle')
     @testing.exclude('mysql', '<', (5, 0, 3))
     def testrollbacktosubtransaction(self):
         connection = testbase.db.connect()

File test/orm/unitofwork.py

         metadata.create_all()
     
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         users_table.delete().execute()
-        UnitOfWorkTest.tearDown(self)
         
     def tearDownAll(self):
         metadata.drop_all()
         UnitOfWorkTest.tearDownAll(self)
-        
+    
     @testing.unsupported('mssql') # TEMP: test causes mssql to hang
     def test_update(self):
         class User(object):
         assert u.name == 'test2'
         assert u.counter == 2
     
+    @testing.unsupported('mssql')
     def test_insert(self):
         class User(object):
             pass
     def setUpAll(self):
         UnitOfWorkTest.setUpAll(self)
         tables.create()
+
     def tearDownAll(self):
         tables.drop()
         UnitOfWorkTest.tearDownAll(self)
+
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         tables.delete()
-        UnitOfWorkTest.tearDown(self)
 
     def testonetomany_1(self):
         """test basic save of one to many."""
         )
 
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         tables.delete()
-        UnitOfWorkTest.tearDown(self)
 
     def testbasic(self):
         # save two users
     def setUpAll(self):
         UnitOfWorkTest.setUpAll(self)
         tables.create()
+
     def tearDownAll(self):
         tables.drop()
         UnitOfWorkTest.tearDownAll(self)
+
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         tables.delete()
-        UnitOfWorkTest.tearDown(self)
     
     def testm2oonetoone(self):
         # TODO: put assertion in here !!!
     def setUpAll(self):
         UnitOfWorkTest.setUpAll(self)
         tables.create()
+
     def tearDownAll(self):
         tables.drop()
         UnitOfWorkTest.tearDownAll(self)
+
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         tables.delete()
-        UnitOfWorkTest.tearDown(self)
 
     def testmanytomany(self):
         items = orderitems
         meta.create_all()
 
     def tearDown(self):
+        UnitOfWorkTest.tearDown(self)
         meta.drop_all()
-        UnitOfWorkTest.tearDown(self)
     
     def testbackwardsnonmatch(self):
         m = mapper(Address, addresses, properties = dict(