Resolving issue on Ubuntu

Author: jahnvi480

tests/test_013_encoding_decoding.py

@@ -4972,129 +4972,6 @@ def test_pooled_connections_have_independent_encoding_settings(conn_str, reset_p
     conn3.close()
 
 
-def test_encoding_settings_persist_across_pool_reuse(conn_str, reset_pooling_state):
-    """Test that encoding settings behavior when connection is reused from pool."""
-    from mssql_python import pooling
-
-    # Enable pooling with max_size=1 to force reuse
-    pooling(max_size=1, idle_timeout=30)
-
-    # First connection: set custom encoding
-    conn1 = mssql_python.connect(conn_str)
-    cursor1 = conn1.cursor()
-    cursor1.execute("SELECT @@SPID")
-    spid1 = cursor1.fetchone()[0]
-
-    conn1.setencoding(encoding="utf-16le", ctype=mssql_python.SQL_WCHAR)
-    conn1.setdecoding(mssql_python.SQL_CHAR, encoding="latin-1")
-
-    enc1 = conn1.getencoding()
-    dec1 = conn1.getdecoding(mssql_python.SQL_CHAR)
-
-    assert enc1["encoding"] == "utf-16le"
-    assert dec1["encoding"] == "latin-1"
-
-    conn1.close()
-
-    # Second connection: should get same SPID (pool reuse)
-    conn2 = mssql_python.connect(conn_str)
-    cursor2 = conn2.cursor()
-    cursor2.execute("SELECT @@SPID")
-    spid2 = cursor2.fetchone()[0]
-
-    # Should reuse same SPID (pool reuse)
-    assert spid1 == spid2
-
-    # Check if settings persist or reset
-    enc2 = conn2.getencoding()
-    # Encoding may persist or reset depending on implementation
-    assert enc2["encoding"] in ["utf-16le", "utf-8"]
-
-    conn2.close()
-
-
-@timeout_test(60)  # 60-second timeout for pooling test
-def test_connection_pool_with_threadpool_executor(conn_str, reset_pooling_state):
-    """Test connection pooling with ThreadPoolExecutor for realistic concurrent workload."""
-    from mssql_python import pooling
-    import concurrent.futures
-    import sys
-
-    # Enable pooling
-    pooling(max_size=10, idle_timeout=30)
-
-    # Platform-adjusted settings to prevent hangs
-    max_workers = 8 if sys.platform.startswith(("linux", "darwin")) else 15
-    num_tasks = 30 if sys.platform.startswith(("linux", "darwin")) else 50
-
-    def execute_query_with_encoding(task_id):
-        """Execute a query with specific encoding."""
-        conn = None
-        cursor = None
-        try:
-            conn = mssql_python.connect(conn_str)
-
-            # Set encoding based on task_id
-            encoding = "utf-16le" if task_id % 2 == 0 else "utf-16be"
-            conn.setencoding(encoding=encoding, ctype=mssql_python.SQL_WCHAR)
-            conn.setdecoding(mssql_python.SQL_WCHAR, encoding=encoding)
-
-            # Execute query
-            cursor = conn.cursor()
-            cursor.execute("SELECT CAST(N'Result' AS NVARCHAR(50))")
-            result = cursor.fetchone()
-
-            # Verify encoding is still correct
-            enc = conn.getencoding()
-            assert enc["encoding"] == encoding
-
-            return {
-                "task_id": task_id,
-                "encoding": encoding,
-                "result": result[0] if result else None,
-                "success": True,
-            }
-        except Exception as e:
-            return {
-                "task_id": task_id,
-                "encoding": "unknown",
-                "result": None,
-                "success": False,
-                "error": str(e),
-            }
-        finally:
-            try:
-                if cursor:
-                    cursor.close()
-            except:
-                pass
-            try:
-                if conn:
-                    conn.close()
-            except:
-                pass
-
-    # Use ThreadPoolExecutor with timeout protection
-    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-        futures = [executor.submit(execute_query_with_encoding, i) for i in range(num_tasks)]
-
-        # Collect results with timeout
-        results = []
-        for future in concurrent.futures.as_completed(futures, timeout=50):
-            try:
-                result = future.result(timeout=5)  # 5 second timeout per task
-                results.append(result)
-            except concurrent.futures.TimeoutError:
-                results.append({"task_id": -1, "success": False, "error": "Task timeout"})
-            except Exception as e:
-                results.append({"task_id": -1, "success": False, "error": str(e)})
-
-    # Verify we got most results (allow some failures on slower platforms)
-    success_count = sum(1 for r in results if r.get("success", False))
-    assert len(results) >= num_tasks * 0.8, f"Too few results: {len(results)}/{num_tasks}"
-    assert success_count >= num_tasks * 0.7, f"Too few successful: {success_count}/{num_tasks}"
-
-
 def test_pooling_disabled_encoding_still_works(conn_str, reset_pooling_state):
     """Test that encoding/decoding works correctly when pooling is disabled."""
     from mssql_python import pooling