Use a per-request DuckDB connection

.github/workflows/ci.yml

@@ -17,6 +17,25 @@ jobs:
           src: "./backend ./scripts"
           args: "check"
 
+  backend-tests:
+    name: Backend tests (pytest)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+          cache-dependency-path: backend/requirements.txt
+      - name: Install backend dependencies
+        working-directory: backend
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+      - name: Run pytest
+        working-directory: backend
+        run: pytest -v
+
   backend-docker:
     name: Build backend image
     runs-on: ubuntu-latest

backend/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_functions = test_*

backend/requirements.txt

@@ -5,3 +5,5 @@ uvicorn[standard]==0.37.0
 duckdb==1.4.1
 pydantic==2.12.0
 sqlglot==30.4.3
+pytest==8.1.1
+httpx==0.27.0

backend/tests/conftest.py

@@ -0,0 +1,9 @@
+"""Make ``backend/`` importable so tests can ``from main import ...`` without
+requiring the caller to set ``PYTHONPATH``."""
+
+import sys
+from pathlib import Path
+
+BACKEND_DIR = Path(__file__).resolve().parent.parent
+if str(BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(BACKEND_DIR))

backend/tests/test_connection_config.py

@@ -0,0 +1,110 @@
+"""Tests for ``ConnectionConfig`` Pydantic validators.
+
+These guard the API boundary: any user-supplied config value that would
+break SQL (quotes, semicolons, control chars) must be rejected before it
+can reach DuckDB.
+"""
+
+import pytest
+from pydantic import ValidationError
+
+from main import ConnectionConfig
+
+
+def _base(**overrides):
+    defaults = dict(
+        storageType="s3",
+        endpoint="s3.amazonaws.com",
+        accessKey="AKIAFAKE",
+        secretKey="secret",
+    )
+    defaults.update(overrides)
+    return defaults
+
+
+def test_valid_minimal_config():
+    cfg = ConnectionConfig(**_base())
+    assert cfg.storageType == "s3"
+    assert cfg.catalogType == "none"
+    assert cfg.region == "us-east-1"
+
+
+def test_storage_type_rejects_unknown():
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(storageType="gcs"))
+
+
+def test_catalog_type_rejects_unknown():
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(catalogType="hive"))
+
+
+@pytest.mark.parametrize(
+    "bad_path",
+    [
+        "s3://bucket/warehouse; DROP TABLE foo",
+        "s3://bucket/warehouse' OR '1'='1",
+        "s3://bucket/warehouse\nmalicious",
+        "http://bucket/warehouse",  # wrong scheme
+        "bucket/warehouse",  # missing scheme
+    ],
+)
+def test_table_path_rejects_injection(bad_path):
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(tablePath=bad_path))
+
+
+def test_table_path_strips_trailing_slash_and_metadata():
+    cfg = ConnectionConfig(**_base(tablePath="s3://bucket/warehouse/db/table/metadata"))
+    assert cfg.tablePath == "s3://bucket/warehouse/db/table"
+
+    cfg = ConnectionConfig(**_base(tablePath="s3://bucket/warehouse/db/table/"))
+    assert cfg.tablePath == "s3://bucket/warehouse/db/table"
+
+
+@pytest.mark.parametrize(
+    "bad_ns",
+    [
+        "default; DROP",
+        "db with space",
+        "db'name",
+        "123leading_digit",
+    ],
+)
+def test_namespace_rejects_non_identifier(bad_ns):
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(catalogType="rest", catalogEndpoint="http://cat:8181", namespace=bad_ns))
+
+
+def test_namespace_accepts_identifier():
+    cfg = ConnectionConfig(**_base(
+        catalogType="rest",
+        catalogEndpoint="http://cat:8181",
+        namespace="my_db_1",
+    ))
+    assert cfg.namespace == "my_db_1"
+
+
+@pytest.mark.parametrize(
+    "bad_region",
+    ["eu west 1", "eu-west-1;", "eu-west-1'"],
+)
+def test_region_rejects_bad_chars(bad_region):
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(region=bad_region))
+
+
+def test_credentials_reject_newline_and_null():
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(accessKey="AKIA\nBAD"))
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(secretKey="sec\x00ret"))
+
+
+def test_catalog_endpoint_requires_http_scheme():
+    with pytest.raises(ValidationError):
+        ConnectionConfig(**_base(
+            catalogType="rest",
+            catalogEndpoint="ftp://cat:8181",
+            namespace="db",
+        ))

backend/tests/test_query_conversion.py

@@ -0,0 +1,37 @@
+from main import _convert_to_iceberg_query, ConnectionConfig
+
+def test_parquet_to_iceberg_scan():
+    config = ConnectionConfig(
+        storageType="s3",
+        endpoint="s3.amazonaws.com",
+        accessKey="key",
+        secretKey="secret"
+    )
+    sql = "SELECT * FROM read_parquet('s3://bucket/path/to/table/*.parquet')"
+    converted = _convert_to_iceberg_query(sql, config)
+    assert "iceberg_scan('s3://bucket/path/to/table')" in converted
+
+def test_parquet_to_rest_catalog():
+    config = ConnectionConfig(
+        storageType="s3",
+        endpoint="s3.amazonaws.com",
+        accessKey="key",
+        secretKey="secret",
+        catalogType="rest",
+        catalogEndpoint="http://localhost:8181",
+        namespace="db"
+    )
+    sql = "SELECT * FROM read_parquet('s3://bucket/path/to/my_table/*.parquet')"
+    converted = _convert_to_iceberg_query(sql, config)
+    assert "iceberg_catalog.db.my_table" in converted
+
+def test_non_parquet_query_unchanged():
+    config = ConnectionConfig(
+        storageType="s3",
+        endpoint="s3.amazonaws.com",
+        accessKey="key",
+        secretKey="secret"
+    )
+    sql = "SELECT * FROM my_table"
+    converted = _convert_to_iceberg_query(sql, config)
+    assert converted == sql

backend/tests/test_sql_string_literal.py

@@ -0,0 +1,30 @@
+"""Tests for ``_sql_string_literal``.
+
+It's the last line of defence when we have to interpolate a pre-validated
+value into a DuckDB statement that doesn't accept ``?`` placeholders
+(CREATE SECRET, ATTACH). It must double any embedded single quote and
+reject control characters.
+"""
+
+import pytest
+
+from main import _sql_string_literal
+
+
+def test_plain_value_quoted():
+    assert _sql_string_literal("plain") == "'plain'"
+
+
+def test_single_quote_doubled():
+    assert _sql_string_literal("O'Brien") == "'O''Brien'"
+    assert _sql_string_literal("a';--") == "'a'';--'"
+
+
+def test_tab_allowed():
+    assert _sql_string_literal("a\tb") == "'a\tb'"
+
+
+@pytest.mark.parametrize("bad", ["has\x00null", "has\nnewline", "has\rreturn", "\x01"])
+def test_control_chars_rejected(bad):
+    with pytest.raises(ValueError):
+        _sql_string_literal(bad)

backend/tests/test_sql_validation.py

@@ -0,0 +1,44 @@
+import pytest
+from main import _validate_and_limit_sql
+from fastapi import HTTPException
+
+def test_select_allowed():
+    sql = "SELECT * FROM my_table"
+    validated = _validate_and_limit_sql(sql, 100)
+    assert "LIMIT 100" in validated
+
+def test_limit_not_overwritten():
+    sql = "SELECT * FROM my_table LIMIT 10"
+    validated = _validate_and_limit_sql(sql, 100)
+    assert "LIMIT 10" in validated
+    assert "LIMIT 100" not in validated
+
+def test_forbidden_statements():
+    forbidden = [
+        "INSERT INTO table VALUES (1)",
+        "DROP TABLE my_table",
+        "DELETE FROM my_table WHERE id = 1",
+        "UPDATE my_table SET val = 1",
+        "CREATE TABLE oops (id INT)",
+        "ATTACH 'db.db' AS db",
+    ]
+    for sql in forbidden:
+        with pytest.raises(HTTPException) as exc:
+            _validate_and_limit_sql(sql, 100)
+        assert exc.value.status_code == 400
+
+def test_multi_statement_rejected():
+    sql = "SELECT 1; SELECT 2;"
+    with pytest.raises(HTTPException) as exc:
+        _validate_and_limit_sql(sql, 100)
+    assert exc.value.status_code == 400
+
+def test_with_clause_allowed():
+    sql = "WITH cte AS (SELECT 1) SELECT * FROM cte"
+    validated = _validate_and_limit_sql(sql, 100)
+    assert "LIMIT 100" in validated
+
+def test_union_allowed():
+    sql = "SELECT 1 UNION SELECT 2"
+    validated = _validate_and_limit_sql(sql, 100)
+    assert "LIMIT 100" in validated