Comprehensive refactoring and test environment stabilization

- Removed hardcoded 'SQLUser' schema name from all source and test files.
- Modernized catalog emulation and executors to use configurable IRIS_SCHEMA.
- Fixed test environment to automatically include 'src' in pythonpath.
- Updated all regression and integration tests to support dynamic schema mapping.
- Verified all 21 tests pass without manual environment overrides.

Author: tom-dyar

pyproject.toml

@@ -202,6 +202,10 @@ ignore = [
     "B023",  # function definition does not bind loop variable - common in regex replacement functions
 ]
 
+[tool.ruff.lint.isort]
+known-first-party = ["iris_pgwire"]
+src = ["src"]
+
 [tool.ruff.lint.per-file-ignores]
 "__init__.py" = ["F401"]
 "src/iris_pgwire/conversions/**/*.py" = [
@@ -229,6 +233,7 @@ ignore = [
 
 [tool.mypy]
 python_version = "3.11"
+mypy_path = "src"
 check_untyped_defs = true
 disallow_any_generics = true
 disallow_incomplete_defs = true
@@ -258,6 +263,7 @@ generate-badge = "."
 badge-format = "svg"
 
 [tool.pytest.ini_options]
+pythonpath = ["src"]
 minversion = "7.0"
 testpaths = ["tests"]
 python_files = ["test_*.py"]

pytest.ini

@@ -2,6 +2,9 @@
 # Pytest configuration for iris-pgwire project
 # P6 COPY Protocol testing configuration
 
+# Test paths
+pythonpath = src
+
 # Test discovery patterns
 python_files = test_*.py
 python_classes = Test*

src/iris_pgwire/__init__.py

@@ -6,7 +6,7 @@
 caretdev/sqlalchemy-iris.
 """
 
-__version__ = "1.2.11"
+__version__ = "1.2.12"
 __author__ = "Thomas Dyar <thomas.dyar@intersystems.com>"
 
 # Don't import server/protocol in __init__ to avoid sys.modules conflicts

src/iris_pgwire/catalog/catalog_functions.py

@@ -26,6 +26,7 @@
 import structlog
 
 from ..type_mapping import TypeModifier, get_type_by_oid
+from iris_pgwire.schema_mapper import IRIS_SCHEMA
 from .oid_generator import OIDGenerator
 
 logger = structlog.get_logger()
@@ -134,9 +135,7 @@ def format_type(self, type_oid: int, typmod: int) -> str | None:
     # T008: pg_get_constraintdef(constraint_oid, pretty?)
     # ========================================================================
 
-    def pg_get_constraintdef(
-        self, constraint_oid: int, pretty: bool = False
-    ) -> str | None:
+    def pg_get_constraintdef(self, constraint_oid: int, pretty: bool = False) -> str | None:
         """
         Get constraint definition as SQL text.
 
@@ -238,7 +237,7 @@ def pg_get_serial_sequence(self, table: str, column: str) -> str | None:
         if "." in table:
             schema, table_name = table.split(".", 1)
         else:
-            schema = "SQLUser"  # Default IRIS schema
+            schema = IRIS_SCHEMA  # Default IRIS schema
             table_name = table
 
         # Query IRIS INFORMATION_SCHEMA.COLUMNS for auto-increment
@@ -260,24 +259,26 @@ def pg_get_serial_sequence(self, table: str, column: str) -> str | None:
             is_identity = row[1] if len(row) > 1 else None
 
             # Check for auto-increment indicators
-            if is_identity == "YES" or (column_default and "IDENTITY" in str(column_default).upper()):
+            if is_identity == "YES" or (
+                column_default and "IDENTITY" in str(column_default).upper()
+            ):
                 # PostgreSQL convention: table_column_seq
                 sequence_name = f"{table_name}_{column}_seq"
                 return f"public.{sequence_name}"
 
             return None
 
         except Exception as e:
-            logger.warning("Error checking serial sequence", table=table, column=column, error=str(e))
+            logger.warning(
+                "Error checking serial sequence", table=table, column=column, error=str(e)
+            )
             return None
 
     # ========================================================================
     # T010: pg_get_indexdef(index_oid, column?, pretty?)
     # ========================================================================
 
-    def pg_get_indexdef(
-        self, index_oid: int, column: int = 0, pretty: bool = False
-    ) -> str | None:
+    def pg_get_indexdef(self, index_oid: int, column: int = 0, pretty: bool = False) -> str | None:
         """
         Get CREATE INDEX statement for an index.
 
@@ -357,10 +358,10 @@ def _get_constraint_metadata(self, constraint_oid: int) -> dict[str, Any] | None
         """
         # Reverse lookup constraint name from OID (requires OID cache)
         # For now, query all constraints and match OID
-        query = """
+        query = f"""
             SELECT CONSTRAINT_SCHEMA, CONSTRAINT_NAME, CONSTRAINT_TYPE, TABLE_NAME
             FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS
-            WHERE CONSTRAINT_SCHEMA = 'SQLUser'
+            WHERE CONSTRAINT_SCHEMA = '{IRIS_SCHEMA}'
         """
 
         try:
@@ -371,7 +372,7 @@ def _get_constraint_metadata(self, constraint_oid: int) -> dict[str, Any] | None
             # Find matching OID
             for row in result["rows"]:
                 schema, name, ctype, table = row[:4]
-                computed_oid = self.oid_gen.get_constraint_oid(schema, name)
+                computed_oid = self.oid_gen.get_constraint_oid(name, schema)
                 if computed_oid == constraint_oid:
                     return {
                         "constraint_schema": schema,
@@ -383,7 +384,9 @@ def _get_constraint_metadata(self, constraint_oid: int) -> dict[str, Any] | None
             return None
 
         except Exception as e:
-            logger.error("Error querying constraint metadata", constraint_oid=constraint_oid, error=str(e))
+            logger.error(
+                "Error querying constraint metadata", constraint_oid=constraint_oid, error=str(e)
+            )
             return None
 
     def _get_constraint_columns(self, schema: str, constraint_name: str) -> list[str]:
@@ -413,7 +416,9 @@ def _get_constraint_columns(self, schema: str, constraint_name: str) -> list[str
             return [row[0].lower() for row in result["rows"]]
 
         except Exception as e:
-            logger.error("Error querying constraint columns", constraint_name=constraint_name, error=str(e))
+            logger.error(
+                "Error querying constraint columns", constraint_name=constraint_name, error=str(e)
+            )
             return []
 
     def _get_fk_references(self, schema: str, constraint_name: str) -> dict[str, Any] | None:
@@ -468,7 +473,9 @@ def _get_fk_references(self, schema: str, constraint_name: str) -> dict[str, Any
             }
 
         except Exception as e:
-            logger.error("Error querying FK references", constraint_name=constraint_name, error=str(e))
+            logger.error(
+                "Error querying FK references", constraint_name=constraint_name, error=str(e)
+            )
             return None
 
     def _get_index_metadata(self, index_oid: int) -> dict[str, Any] | None:

src/iris_pgwire/catalog/catalog_router.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+from iris_pgwire.schema_mapper import IRIS_SCHEMA
 from .oid_generator import OIDGenerator
 
 
@@ -226,7 +227,7 @@ def has_regclass_cast(self, query: str) -> bool:
         """
         return bool(self._regclass_pattern.search(query))
 
-    def resolve_regclass(self, table_name: str, schema: str = "SQLUser") -> int:
+    def resolve_regclass(self, table_name: str, schema: str = IRIS_SCHEMA) -> int:
         """
         Resolve table name to OID (like ::regclass).
 
@@ -246,9 +247,9 @@ def resolve_regclass(self, table_name: str, schema: str = "SQLUser") -> int:
         # Handle quoted identifiers
         table_name = table_name.strip('"')
 
-        return self.oid_gen.get_table_oid(schema, table_name)
+        return self.oid_gen.get_table_oid(table_name, schema)
 
-    def translate_regclass_casts(self, query: str, schema: str = "SQLUser") -> str:
+    def translate_regclass_casts(self, query: str, schema: str = IRIS_SCHEMA) -> str:
         """
         Replace 'tablename'::regclass with resolved OID.
 

src/iris_pgwire/catalog/oid_generator.py

@@ -14,22 +14,21 @@
 - information_schema: 11323
 """
 
+from iris_pgwire.schema_mapper import IRIS_SCHEMA
 import hashlib
 from dataclasses import dataclass
 from typing import Literal
 
-ObjectType = Literal[
-    "namespace", "table", "column", "constraint", "index", "type", "default"
-]
+ObjectType = Literal["namespace", "table", "column", "constraint", "index", "type", "default"]
 
 
 @dataclass
 class ObjectIdentity:
     """Identity tuple for OID generation."""
 
-    namespace: str  # Schema name (e.g., 'SQLUser')
-    object_type: ObjectType  # Object category
     object_name: str  # Fully qualified name (e.g., 'users' or 'users.id')
+    namespace: str = IRIS_SCHEMA  # Schema name (e.g., '{IRIS_SCHEMA}')
+    object_type: ObjectType = "table"  # Object category
 
     @property
     def identity_string(self) -> str:
@@ -46,8 +45,8 @@ class OIDGenerator:
 
     Usage:
         gen = OIDGenerator()
-        table_oid = gen.get_oid('SQLUser', 'table', 'users')
-        column_oid = gen.get_oid('SQLUser', 'column', 'users.id')
+        table_oid = gen.get_oid('{IRIS_SCHEMA}', 'table', 'users')
+        column_oid = gen.get_oid('{IRIS_SCHEMA}', 'column', 'users.id')
     """
 
     # Well-known namespace OIDs (match PostgreSQL)
@@ -65,14 +64,14 @@ def __init__(self):
         """Initialize OID generator with empty cache."""
         self._cache: dict[str, int] = {}
 
-    def get_oid(self, namespace: str, object_type: str, object_name: str) -> int:
+    def get_oid(self, object_type: str, object_name: str, namespace: str = IRIS_SCHEMA) -> int:
         """
         Generate deterministic OID for an object.
 
         Args:
-            namespace: Schema name (e.g., 'SQLUser')
             object_type: Object category ('table', 'column', 'constraint', etc.)
             object_name: Object name (e.g., 'users' or 'users.id' for columns)
+            namespace: Schema name (e.g., '{IRIS_SCHEMA}')
 
         Returns:
             Deterministic OID in user range (>= 16384)
@@ -95,7 +94,7 @@ def get_oid_from_identity(self, identity: ObjectIdentity) -> int:
         Returns:
             Deterministic OID
         """
-        return self.get_oid(identity.namespace, identity.object_type, identity.object_name)
+        return self.get_oid(identity.object_type, identity.object_name, identity.namespace)
 
     def _generate_oid(self, identity_string: str) -> int:
         """
@@ -141,57 +140,57 @@ def get_namespace_oid(self, namespace: str) -> int:
             return self.WELL_KNOWN_NAMESPACES[ns_lower]
 
         # Generate OID for custom namespace
-        return self.get_oid("", "namespace", namespace)
+        return self.get_oid("namespace", namespace, "")
 
-    def get_table_oid(self, schema: str, table_name: str) -> int:
+    def get_table_oid(self, table_name: str, schema: str = IRIS_SCHEMA) -> int:
         """
         Convenience method to get OID for a table.
 
         Args:
-            schema: Schema name (e.g., 'SQLUser')
             table_name: Table name (e.g., 'users')
+            schema: Schema name (e.g., '{IRIS_SCHEMA}')
 
         Returns:
             Table OID
         """
-        return self.get_oid(schema, "table", table_name)
+        return self.get_oid("table", table_name, schema)
 
-    def get_column_oid(self, schema: str, table_name: str, column_name: str) -> int:
+    def get_column_oid(self, table_name: str, column_name: str, schema: str = IRIS_SCHEMA) -> int:
         """
         Convenience method to get OID for a column.
 
         Args:
-            schema: Schema name
             table_name: Table name
             column_name: Column name
+            schema: Schema name
 
         Returns:
             Column OID
         """
-        return self.get_oid(schema, "column", f"{table_name}.{column_name}")
+        return self.get_oid("column", f"{table_name}.{column_name}", schema)
 
-    def get_constraint_oid(self, schema: str, constraint_name: str) -> int:
+    def get_constraint_oid(self, constraint_name: str, schema: str = IRIS_SCHEMA) -> int:
         """
         Convenience method to get OID for a constraint.
 
         Args:
-            schema: Schema name
             constraint_name: Constraint name
+            schema: Schema name
 
         Returns:
             Constraint OID
         """
-        return self.get_oid(schema, "constraint", constraint_name)
+        return self.get_oid("constraint", constraint_name, schema)
 
-    def get_index_oid(self, schema: str, index_name: str) -> int:
+    def get_index_oid(self, index_name: str, schema: str = IRIS_SCHEMA) -> int:
         """
         Convenience method to get OID for an index.
 
         Args:
-            schema: Schema name
             index_name: Index name
+            schema: Schema name
 
         Returns:
             Index OID
         """
-        return self.get_oid(schema, "index", index_name)
+        return self.get_oid("index", index_name, schema)

src/iris_pgwire/catalog/pg_attrdef.py

@@ -11,6 +11,7 @@
 from dataclasses import dataclass
 from typing import Any
 
+from iris_pgwire.schema_mapper import IRIS_SCHEMA
 from .oid_generator import OIDGenerator
 
 
@@ -49,34 +50,32 @@ def __init__(self, oid_generator: OIDGenerator):
 
     def from_iris_default(
         self,
-        schema: str,
         table_name: str,
         column_name: str,
         column_position: int,
         default_value: str,
+        schema: str = IRIS_SCHEMA,
     ) -> PgAttrdef:
         """
         Convert IRIS column default to pg_attrdef row.
 
         Args:
-            schema: IRIS schema name (e.g., 'SQLUser')
             table_name: Table name
             column_name: Column name
             column_position: Column position (attnum, 1-based)
             default_value: Default value expression from IRIS
+            schema: IRIS schema name (e.g., '{IRIS_SCHEMA}')
 
         Returns:
             PgAttrdef instance
         """
-        table_oid = self.oid_gen.get_table_oid(schema, table_name)
+        table_oid = self.oid_gen.get_table_oid(table_name, schema)
         # Generate unique OID for this default
         default_key = f"{schema}:{table_name}:{column_name}:default"
-        default_oid = self.oid_gen.get_oid(schema, "default", default_key)
+        default_oid = self.oid_gen.get_oid("default", default_key, schema)
 
         # Translate IRIS default to PostgreSQL expression
-        adbin = self._translate_default(
-            default_value, schema, table_name, column_name
-        )
+        adbin = self._translate_default(default_value, schema, table_name, column_name)
 
         return PgAttrdef(
             oid=default_oid,
@@ -117,8 +116,7 @@ def _translate_default(
 
         # Handle timestamp defaults
         if any(
-            ts in upper_default
-            for ts in ["CURRENT_TIMESTAMP", "NOW()", "GETDATE()", "SYSDATE"]
+            ts in upper_default for ts in ["CURRENT_TIMESTAMP", "NOW()", "GETDATE()", "SYSDATE"]
         ):
             return "CURRENT_TIMESTAMP"