Check for SQLite header in database_exists

conftest.py

@@ -73,7 +73,7 @@ def sqlite_none_database_dsn():
 
 
 @pytest.fixture
-def sqlite_file_dsn():
+def sqlite_file_dsn(db_name):
     return 'sqlite:///{0}.db'.format(db_name)
 
 

sqlalchemy_utils/functions/database.py

@@ -458,6 +458,15 @@ def get_scalar_result(engine, sql):
         engine.dispose()
         return result
 
+    def sqlite_file_exists(database):
+        if not os.path.isfile(database) or os.path.getsize(database) < 100:
+            return False
+
+        with open(database, 'rb') as f:
+            header = f.read(100)
+
+        return header[:16] == b'SQLite format 3\x00'
+
     url = copy(make_url(url))
     database = url.database
     if url.drivername.startswith('postgres'):
@@ -478,7 +487,7 @@ def get_scalar_result(engine, sql):
 
     elif engine.dialect.name == 'sqlite':
         if database:
-            return database == ':memory:' or os.path.exists(database)
+            return database == ':memory:' or sqlite_file_exists(database)
         else:
             # The default SQLAlchemy database is in memory,
             # and :memory is not required, thus we should support that use-case
@@ -567,7 +576,8 @@ def create_database(url, encoding='utf8', template=None):
 
     elif engine.dialect.name == 'sqlite' and database != ':memory:':
         if database:
-            open(database, 'w').close()
+            engine.execute("CREATE TABLE DB(id int);")
+            engine.execute("DROP TABLE DB;")
 
     else:
         text = 'CREATE DATABASE {0}'.format(quote(engine, database))

tests/functions/test_database.py

@@ -35,7 +35,10 @@ def test_exists_memory_none_database(self, sqlite_none_database_dsn):
 
 @pytest.mark.usefixtures('sqlite_file_dsn')
 class TestDatabaseSQLiteFile(DatabaseTest):
-    pass
+    def test_existing_non_sqlite_file(self, dsn):
+        database = sa.engine.url.make_url(dsn).database
+        open(database, 'w').close()
+        self.test_create_and_drop(dsn)
 
 
 @pytest.mark.skipif('pymysql is None')