add tests for column defaults when inserting

encode · ckkz-it · Nov 12, 2020 · Nov 12, 2020 · Nov 13, 2020 · Nov 13, 2020
commit 71f4616160d5628e339d2dc5757cd42c576c88d5
diff --git a/databases/core.py b/databases/core.py
@@ -301,6 +301,11 @@ def _build_query(
         elif values:
             return query.values(**values)
 
+        # for case when `table.insert()` called without `.values()` it has to be
+        # called to produce `insert_prefetch` for compiled query
+        if query.__visit_name__ == "insert":
+            return query.values()
+
         return query
 
 

diff --git a/tests/test_databases.py b/tests/test_databases.py
@@ -70,6 +70,15 @@ def process_result_value(self, value, dialect):
     sqlalchemy.Column("price", sqlalchemy.Numeric(precision=30, scale=20)),
 )
 
+# Used to test column default values
+default_values = sqlalchemy.Table(
+    "default_values",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
+    sqlalchemy.Column("with_default", sqlalchemy.Integer, default=42),
+    sqlalchemy.Column("without_default", sqlalchemy.Integer),
+)
+
 
 @pytest.fixture(autouse=True, scope="module")
 def create_test_database():
@@ -651,6 +660,50 @@ async def test_json_field(database_url):
             assert results[0]["data"] == {"text": "hello", "boolean": True, "int": 1}
 
 
+@pytest.mark.parametrize("database_url", DATABASE_URLS)
+@async_adapter
+async def test_insert_with_default_values(database_url):
+    """
+    Test insert with column default values
+    """
+
+    async with Database(database_url) as database:
+        async with database.transaction(force_rollback=True):
+            # execute()
+            query = default_values.insert()
+            values = {"without_default": 1}
+            inserted_id = await database.execute(query, values)
+
+            # fetch_one()
+            query = default_values.select().where(default_values.c.id == inserted_id)
+            result = await database.fetch_one(query=query)
+            assert result["with_default"] == 42
+            assert result["without_default"] == values["without_default"]
+
+            # test without passing values and without calling `values()`
+            # execute()
+            query = default_values.insert()
+            inserted_id = await database.execute(query)
+
+            # fetch_one()
+            query = default_values.select().where(default_values.c.id == inserted_id)
+            result = await database.fetch_one(query=query)
+            assert result["with_default"] == 42
+            assert result["without_default"] is None
+
+            # test pass other than default value
+            # execute()
+            query = default_values.insert()
+            values = {"with_default": 5}
+            inserted_id = await database.execute(query, values)
+
+            # fetch_one()
+            query = default_values.select().where(default_values.c.id == inserted_id)
+            result = await database.fetch_one(query=query)
+            assert result["with_default"] == values["with_default"]
+            assert result["without_default"] is None
+
+
 @pytest.mark.parametrize("database_url", DATABASE_URLS)
 @async_adapter
 async def test_custom_field(database_url):
@@ -915,7 +968,6 @@ def test_global_connection_is_initialized_lazily(database_url):
     @async_adapter
     async def run_database_queries():
         async with database:
-
             async def db_lookup():
                 await database.fetch_one("SELECT pg_sleep(1)")