fix: Allow connection with schema=information_schema (#122)

With snowflake, it's sometimes (particularly with alembic) convenient to
connect with `schema='information_schema'`, so that all other schemas
are required explicitly (since there's no public schema, and the weird
way snowflake scoping works).

As-is without this PR, the duckdb `information_schema` is lowercased,
and the auto-use-schema code in fakesnow would unconditionally
`.upper()` the given schema, search for `INFORMATION_SCHEMA`, not find
it, and then you'd get `This session does not have a current schema.
Call 'USE SCHEMA'` when you wouldn't in real life.

It might perhaps be more ideal to somehow fully recreate the snowflake
information_schema and replace it, because there **are** some
differences with the duckdb version that i suspect will end up being
irreconcilable, but it's not obvious to me **how** to do this, nor how
much additional work it'd be so...just a thought.

---------

Co-authored-by: Oliver Mannion <[email protected]>

fakesnow/fakes.py

@@ -523,9 +523,14 @@ def __init__(
     ):
         self._duck_conn = duck_conn
         # upper case database and schema like snowflake unquoted identifiers
-        # NB: catalog names are not case-sensitive in duckdb but stored as cased in information_schema.schemata
+        # so they appear as upper-cased in information_schema
+        # catalog and schema names are not actually case-sensitive in duckdb even though
+        # they are as cased in information_schema.schemata, so when selecting from
+        # information_schema.schemata below we use upper-case to match any existing duckdb
+        # catalog or schemas like "information_schema"
         self.database = database and database.upper()
         self.schema = schema and schema.upper()
+
         self.database_set = False
         self.schema_set = False
         self.db_path = Path(db_path) if db_path else None
@@ -539,7 +544,7 @@ def __init__(
             and self.database
             and not duck_conn.execute(
                 f"""select * from information_schema.schemata
-                where catalog_name = '{self.database}'"""
+                where upper(catalog_name) = '{self.database}'"""
             ).fetchone()
         ):
             db_file = f"{self.db_path/self.database}.db" if self.db_path else ":memory:"
@@ -554,7 +559,7 @@ def __init__(
             and self.schema
             and not duck_conn.execute(
                 f"""select * from information_schema.schemata
-                where catalog_name = '{self.database}' and schema_name = '{self.schema}'"""
+                where upper(catalog_name) = '{self.database}' and upper(schema_name) = '{self.schema}'"""
             ).fetchone()
         ):
             duck_conn.execute(f"CREATE SCHEMA {self.database}.{self.schema}")
@@ -565,7 +570,7 @@ def __init__(
             and self.schema
             and duck_conn.execute(
                 f"""select * from information_schema.schemata
-                where catalog_name = '{self.database}' and schema_name = '{self.schema}'"""
+                where upper(catalog_name) = '{self.database}' and upper(schema_name) = '{self.schema}'"""
             ).fetchone()
         ):
             duck_conn.execute(f"SET schema='{self.database}.{self.schema}'")
@@ -576,7 +581,7 @@ def __init__(
             self.database
             and duck_conn.execute(
                 f"""select * from information_schema.schemata
-                where catalog_name = '{self.database}'"""
+                where upper(catalog_name) = '{self.database}'"""
             ).fetchone()
         ):
             duck_conn.execute(f"SET schema='{self.database}.main'")

tests/test_connect.py

@@ -83,6 +83,15 @@ def test_connect_db_path_reuse():
             assert cur.execute("select * from example").fetchall() == [(420,)]
 
 
+def test_connect_information_schema():
+    with fakesnow.patch(create_schema_on_connect=False):
+        conn = snowflake.connector.connect(database="db1", schema="information_schema")
+        assert conn.schema == "INFORMATION_SCHEMA"
+        with conn, conn.cursor() as cur:
+            # shouldn't fail
+            cur.execute("SELECT * FROM databases")
+
+
 def test_connect_without_database(_fakesnow_no_auto_create: None):
     with snowflake.connector.connect() as conn, conn.cursor() as cur:
         with pytest.raises(snowflake.connector.errors.ProgrammingError) as excinfo: