Fix for #49

Author: pcisar

src/firebird/driver/core.py

@@ -53,6 +53,7 @@
 import sys
 import threading
 import weakref
+from urllib.parse import urlparse
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Mapping, Sequence
 from ctypes import addressof, byref, create_string_buffer, memmove, memset, pointer, string_at
@@ -1795,7 +1796,7 @@ def __init__(self, att: iAttachment, dsn: str, dpb: bytes | None=None, sql_diale
         self.__FIREBIRD_LIB__ = None
     def __del__(self):
         if not self.is_closed():
-            warn(f"Connection disposed without prior close()", ResourceWarning)
+            warn("Connection disposed without prior close()", ResourceWarning)
             self._close()
             self._close_internals()
             self._att.detach()
@@ -2179,6 +2180,75 @@ def _connect_helper(dsn: str, host: str, port: str, database: str, protocol: Net
         dsn += database
     return dsn
 
+def _is_dsn(value: str) -> bool:
+    """
+    Checks if the given string matches known patterns for Firebird DSNs.
+
+    This function analyzes the string for structures that are typical for
+    Firebird connection strings, based on how the firebird-driver might
+    construct them (per _connect_helper) or how the Firebird client
+    library generally interprets them.
+
+    Args:
+        value: The string to check.
+
+    Returns:
+        True if the string matches a DSN pattern, False otherwise.
+    """
+    if not isinstance(value, str) or not value.strip():
+        # Empty or whitespace-only strings are not DSNs
+        return False
+
+    # 1. Protocol-based DSNs (e.g., inet://localhost/employee)
+    #    These are directly produced by _connect_helper if a protocol is specified.
+    try:
+        parsed = urlparse(value)
+        if parsed.scheme in [p.name.lower() for p in NetProtocol]:
+            # A scheme-based DSN must have a network location (host/port) or a path (database part).
+            # parsed.path.lstrip('/') handles cases like "xnet:///path/to/db" where netloc is empty.
+            if parsed.netloc or (parsed.path and parsed.path.lstrip('/')):
+                return True
+    except ValueError:
+        # urlparse can raise ValueError for some malformed inputs (e.g. "::1")
+        # These are unlikely to be scheme-based DSNs in the firebird context.
+        pass
+
+    # 2. Windows Named Pipes (e.g., \\server\pipe_name or \\server@port\pipe_name)
+    #    These are indicated by starting with \\.
+    if value.startswith("\\\\"):
+        # Basic check: must have some content after \\, and not be just \\ or \\\
+        if len(value) > 2 and value[2] not in ['\\', '/']:
+            return True
+
+    # 3. Classic host:database or host/port:database syntax
+    #    (e.g., localhost:employee, server/3050:/data/db.fdb)
+    #    This pattern should not be confused with "C:\path" or "http://..."
+    colon_idx = value.find(':')
+    if colon_idx > 0 and "://" not in value[:colon_idx]: # Colon exists, not at start, and not part of a scheme
+        host_spec = value[:colon_idx]
+        # db_spec = value[colon_idx+1:] # Not strictly needed for this check
+
+        # Avoid misinterpreting "C:\path" as "host C, db \path".
+        # If host_spec is a single letter (like a drive) AND the char after colon is a path separator,
+        # it's more likely an absolute path. os.path.isabs handles these better.
+        is_windows_drive_abs_path_candidate = (
+            len(host_spec) == 1 and host_spec[0].isalpha() and
+            len(value) > colon_idx + 1 and value[colon_idx+1] in ('/', '\\')
+        )
+
+        if not is_windows_drive_abs_path_candidate:
+            # host_spec should not contain backslashes if it's a hostname.
+            # It can contain a forward slash for host/port.
+            if '\\' not in host_spec:
+                if '/' in host_spec: # Potential host/port:db
+                    # e.g., "server/3050:dbname"
+                    parts = host_spec.split('/', 1) # Split only on the first /
+                    if len(parts) == 2 and parts[0] and parts[1].isdigit(): # host part and port part (digits)
+                        return True
+                elif host_spec: # Plain host:db, ensure host_spec is not empty
+                    # e.g., "localhost:dbname", "localhost:/path/to/db", "localhost:C:relative_path_on_C"
+                    return True
+    return False
 def __make_connection(dsn: str, utf8filename: bool, dpb: bytes, sql_dialect: int, charset: str, # noqa: FBT001
                       crypt_callback: iCryptKeyCallbackImpl, *, create: bool) -> Connection:
     with a.get_api().master.get_dispatcher() as provider:
@@ -2244,13 +2314,14 @@ def connect(database: str | Path, *, user: str | None=None, password: str | None
     if isinstance(database, Path):
         database = str(database)
     db_config: DatabaseConfig = driver_config.get_database(database)
+    dsn: str | None = None
     if db_config is None:
         db_config = driver_config.db_defaults
-        # we'll assume that 'database' is 'dsn'
-        dsn = database
-        database = None
         srv_config = driver_config.server_defaults
-        srv_config.host.clear()
+        if _is_dsn(database):
+            dsn = database
+            database = None
+            srv_config.host.clear()
     else:
         database = db_config.database.value
         dsn = db_config.dsn.value
@@ -2328,11 +2399,11 @@ def create_database(database: str | Path, *, user: str | None=None, password: st
     db_config: DatabaseConfig = driver_config.get_database(database)
     if db_config is None:
         db_config = driver_config.db_defaults
-        # we'll assume that 'database' is 'dsn'
-        dsn = database
-        database = None
         srv_config = driver_config.server_defaults
-        srv_config.host.clear()
+        if _is_dsn(database):
+            dsn = database
+            database = None
+            srv_config.host.clear()
     else:
         database = db_config.database.value
         dsn = db_config.dsn.value
@@ -2537,7 +2608,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.close()
     def __del__(self):
         if self._tra is not None:
-            warn(f"Transaction disposed while active", ResourceWarning)
+            warn("Transaction disposed while active", ResourceWarning)
             self._finish()
     def __dead_con(self, obj: Connection) -> None: # noqa: ARG002
         self._connection = None
@@ -2947,7 +3018,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.free()
     def __del__(self):
         if self._in_meta or self._out_meta or self._istmt:
-            warn(f"Statement disposed without prior free()", ResourceWarning)
+            warn("Statement disposed without prior free()", ResourceWarning)
             self.free()
     def __dead_con(self, obj: Connection) -> None: # noqa: ARG002
         self._connection = None
@@ -3081,7 +3152,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.close()
     def __del__(self):
         if self._blob is not None:
-            warn(f"BlobReader disposed without prior close()", ResourceWarning)
+            warn("BlobReader disposed without prior close()", ResourceWarning)
             self.close()
     def flush(self) -> None:
         """Does nothing.
@@ -3286,7 +3357,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.close()
     def __del__(self):
         if self._result is not None or self._stmt is not None or self.__blob_readers:
-            warn(f"Cursor disposed without prior close()", ResourceWarning)
+            warn("Cursor disposed without prior close()", ResourceWarning)
             self.close()
     def __next__(self):
         if (row := self.fetchone()) is not None:
@@ -5541,7 +5612,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.close()
     def __del__(self):
         if self._svc is not None:
-            warn(f"Server disposed without prior close()", ResourceWarning)
+            warn("Server disposed without prior close()", ResourceWarning)
             self.close()
     def __next__(self):
         if (line := self.readline()) is not None:

tests/test_connection.py

@@ -27,7 +27,8 @@
 import firebird.driver as driver
 from firebird.driver.types import ImpData, ImpDataOld
 from firebird.driver import (NetProtocol, connect, Isolation, tpb,  DefaultAction,
-                             DbInfoCode, DbWriteMode, DbAccessMode, DbSpaceReservation)
+                             DbInfoCode, DbWriteMode, DbAccessMode, DbSpaceReservation,
+                             driver_config)
 
 def test_connect_helper():
     DB_LINUX_PATH = '/path/to/db/employee.fdb'
@@ -376,3 +377,177 @@ def test_db_info(db_connection, fb_vars, db_file):
     guid = con.info.get_info(DbInfoCode.DB_GUID)
     assert isinstance(guid, str)
     assert len(guid) == 38 # Example check for {GUID} format
+
+def test_connect_with_driver_config_server_defaults_local(driver_cfg, db_file, fb_vars):
+    """
+    Tests connect() using driver_config.server_defaults for a local connection.
+    The database alias registered for this test will have its 'server' attribute
+    set to None, which means it should pick up settings from server_defaults.
+    """
+    db_alias = "pytest_cfg_local_db"
+    db_path_str = str(db_file)
+
+    # Save original server_defaults to restore them, though driver_cfg fixture handles full reset
+    original_s_host = driver_config.server_defaults.host.value
+    original_s_port = driver_config.server_defaults.port.value
+    original_s_user = driver_config.server_defaults.user.value
+    original_s_password = driver_config.server_defaults.password.value
+
+    # Configure server_defaults for a local connection
+    driver_config.server_defaults.host.value = None
+    driver_config.server_defaults.port.value = None # Explicitly None for local
+    driver_config.server_defaults.user.value = fb_vars['user']
+    driver_config.server_defaults.password.value = fb_vars['password']
+
+    # Ensure the test-specific DB alias is clean if it exists from a prior failed run
+    if driver_config.get_database(db_alias):
+        driver_config.databases.value = [db_cfg for db_cfg in driver_config.databases.value if db_cfg.name != db_alias]
+
+    # Register a database alias that will use these server_defaults
+    test_db_config_entry = driver_config.register_database(db_alias)
+    test_db_config_entry.database.value = db_path_str
+    test_db_config_entry.server.value = None # Key: This tells driver to use server_defaults
+
+    # For a local connection (host=None, port=None), DSN is just the database path
+    expected_dsn = db_path_str
+
+    conn = None
+    try:
+        conn = driver.connect(db_alias, charset='UTF8')
+        assert conn._att is not None, "Connection attachment failed"
+        assert conn.dsn == expected_dsn, f"Expected DSN '{expected_dsn}', got '{conn.dsn}'"
+
+        # Verify connection is usable with a simple query
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1 FROM RDB$DATABASE")
+            assert cur.fetchone()[0] == 1, "Query failed on the connection"
+    finally:
+        if conn and not conn.is_closed():
+            conn.close()
+        # Restore original server_defaults values (driver_cfg also handles full reset)
+        driver_config.server_defaults.host.value = original_s_host
+        driver_config.server_defaults.port.value = original_s_port
+        driver_config.server_defaults.user.value = original_s_user
+        driver_config.server_defaults.password.value = original_s_password
+
+
+def test_connect_with_driver_config_server_defaults_remote(driver_cfg, db_file, fb_vars):
+    """
+    Tests connect() using driver_config.server_defaults for a remote-like connection.
+    This test relies on fb_vars providing a host (and optionally port) from conftest.py.
+    If no host is configured in fb_vars, this test variant is skipped.
+    """
+    db_alias = "pytest_cfg_remote_db"
+    db_path_str = str(db_file)
+
+    test_host = fb_vars.get('host')
+    test_port = fb_vars.get('port') # Can be None or empty string
+
+    if not test_host:
+        pytest.skip("Skipping remote server_defaults test as no host is configured in fb_vars. "
+                    "This test requires a configured host (and optionally port) for execution.")
+        return
+
+    # Save original server_defaults
+    original_s_host = driver_config.server_defaults.host.value
+    original_s_port = driver_config.server_defaults.port.value
+    original_s_user = driver_config.server_defaults.user.value
+    original_s_password = driver_config.server_defaults.password.value
+
+    # Configure server_defaults for a "remote" connection
+    driver_config.server_defaults.host.value = test_host
+    driver_config.server_defaults.port.value = str(test_port) if test_port else None
+    driver_config.server_defaults.user.value = fb_vars['user']
+    driver_config.server_defaults.password.value = fb_vars['password']
+
+    # Ensure the test-specific DB alias is clean
+    if driver_config.get_database(db_alias):
+        driver_config.databases.value = [db_cfg for db_cfg in driver_config.databases.value if db_cfg.name != db_alias]
+
+    test_db_config_entry = driver_config.register_database(db_alias)
+    test_db_config_entry.database.value = db_path_str
+    test_db_config_entry.server.value = None # Use server_defaults
+
+    # Determine expected DSN based on _connect_helper logic for non-protocol DSNs
+    if test_host.startswith("\\\\"): # Windows Named Pipes
+        if test_port:
+            expected_dsn = f"{test_host}@{test_port}\\{db_path_str}"
+        else:
+            expected_dsn = f"{test_host}\\{db_path_str}"
+    elif test_port: # TCP/IP with port
+        expected_dsn = f"{test_host}/{test_port}:{db_path_str}"
+    else: # TCP/IP without port (or other local-like with host)
+        expected_dsn = f"{test_host}:{db_path_str}"
+
+    conn = None
+    try:
+        conn = driver.connect(db_alias, charset='UTF8')
+        assert conn._att is not None, "Connection attachment failed"
+        assert conn.dsn == expected_dsn, f"Expected DSN '{expected_dsn}', got '{conn.dsn}'"
+
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1 FROM RDB$DATABASE")
+            assert cur.fetchone()[0] == 1, "Query failed on the connection"
+    finally:
+        if conn and not conn.is_closed():
+            conn.close()
+        # Restore original server_defaults
+        driver_config.server_defaults.host.value = original_s_host
+        driver_config.server_defaults.port.value = original_s_port
+        driver_config.server_defaults.user.value = original_s_user
+        driver_config.server_defaults.password.value = original_s_password
+
+def test_connect_with_driver_config_db_defaults_local(driver_cfg, db_file, fb_vars):
+    """
+    Tests connect() when db_defaults provides the database path, and
+    server_defaults provides local connection info (host=None, port=None).
+    Here, connect() is called with a DSN-like string that is *not* a registered alias.
+    """
+    db_path_str = str(db_file) # This will be our "DSN" to connect to
+
+    # Save original defaults
+    original_s_host = driver_config.server_defaults.host.value
+    original_s_port = driver_config.server_defaults.port.value
+    original_s_user = driver_config.server_defaults.user.value
+    original_s_password = driver_config.server_defaults.password.value
+    original_db_database = driver_config.db_defaults.database.value
+    original_db_server = driver_config.db_defaults.server.value
+
+
+    # Configure server_defaults for local connection
+    driver_config.server_defaults.host.value = None
+    driver_config.server_defaults.port.value = None
+    driver_config.server_defaults.user.value = fb_vars['user']
+    driver_config.server_defaults.password.value = fb_vars['password']
+
+    # Configure db_defaults (it won't be used for database path if DSN is absolute path)
+    # but it's good to ensure it's set to something known for the test.
+    # The key here is that if connect(db_path_str) is called and db_path_str is
+    # an absolute path, it's treated as the DSN. Server info then comes from
+    # server_defaults IF db_path_str is NOT a full DSN with host/port.
+    # If db_path_str is an absolute path, it's treated as the direct database target.
+    driver_config.db_defaults.database.value = "some_default_db_ignore" # Should not be used if DSN is absolute
+    driver_config.db_defaults.server.value = None # Use server_defaults
+
+    expected_dsn = db_path_str # For local connection with absolute path, DSN is the path
+
+    conn = None
+    try:
+        # Connect using the absolute path as the DSN
+        conn = driver.connect(db_path_str, charset='UTF8')
+        assert conn._att is not None, "Connection attachment failed"
+        assert conn.dsn == expected_dsn, f"Expected DSN '{expected_dsn}', got '{conn.dsn}'"
+
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1 FROM RDB$DATABASE")
+            assert cur.fetchone()[0] == 1, "Query failed on the connection"
+    finally:
+        if conn and not conn.is_closed():
+            conn.close()
+        # Restore originals
+        driver_config.server_defaults.host.value = original_s_host
+        driver_config.server_defaults.port.value = original_s_port
+        driver_config.server_defaults.user.value = original_s_user
+        driver_config.server_defaults.password.value = original_s_password
+        driver_config.db_defaults.database.value = original_db_database
+        driver_config.db_defaults.server.value = original_db_server