test: Add end-to-end SQL test (#12)

* test: Add end-to-end SQL test

* Make black happy

* Remove unnecessary `commit`

* Select `id` column

* XFail on windows

* Add `pytest-github-actions-annotate-failures` to test env

Author: edgarrmondragon

pyproject.toml

@@ -56,7 +56,12 @@ Source = "https://github.com/edgarrmondragon/sqlean-driver"
 source = "vcs"
 
 [tool.hatch.envs.default]
-dependencies = ["coverage[toml]>=6.5", "pytest", "sqlalchemy=={env:SQLALCHEMY_VERSION:2.*}"]
+dependencies = [
+  "coverage[toml]>=6.5",
+  "pytest",
+  "pytest-github-actions-annotate-failures",
+  "sqlalchemy=={env:SQLALCHEMY_VERSION:2.*}",
+]
 [tool.hatch.envs.default.scripts]
 test = "pytest {args:tests}"
 test-cov = "coverage run -m pytest {args:tests}"

tests/test_driver.py

@@ -8,8 +8,9 @@
 from types import ModuleType
 
 import pytest
-from sqlalchemy import column, create_engine, func, select
+from sqlalchemy import Column, MetaData, Table, column, create_engine, func, insert, literal, select
 from sqlalchemy.exc import OperationalError
+from sqlalchemy.types import Integer, String
 
 if t.TYPE_CHECKING:
     from sqlalchemy.engine import Engine
@@ -119,3 +120,50 @@ def test_extensions(extensions: str, query: Select, expected: tuple):
     with engine.connect() as conn:
         result = conn.execute(query)
     assert result.fetchone() == expected
+
+
+@pytest.mark.xfail(
+    sys.platform == "win32",
+    reason="'ipaddr' extension not available on Windows",
+)
+def test_e2e_sql():
+    """Test that the SQL works."""
+    url = "sqlite+sqlean:///:memory:?extensions=all"
+    engine = create_engine(url)
+
+    metadata = MetaData()
+    table = Table(
+        "example",
+        metadata,
+        Column("id", Integer, primary_key=True),
+        Column("ip", String),
+        Column("network", String),
+        Column("family", Integer),
+    )
+
+    metadata.create_all(engine)
+
+    with engine.connect() as conn:
+        insert_sql = insert(table).from_select(
+            [
+                table.c.ip,
+                table.c.network,
+                table.c.family,
+            ],
+            select(
+                literal("192.168.1.1"),
+                func.ipnetwork("192.168.1.1").label("network"),
+                func.ipfamily("192.168.1.1").label("family"),
+            ),
+        )
+        conn.execute(insert_sql)
+
+        select_sql = select(
+            table.c.id,
+            table.c.ip,
+            table.c.network,
+            table.c.family,
+        )
+        result = conn.execute(select_sql)
+
+        assert result.fetchone() == (1, "192.168.1.1", "192.168.1.1/32", 4)