cowork-bot: surface silently-dropped SQL constraints and malformed type overrides as warnings

Author: DevForge Engineer

src/schemaforge/generators/json_schema_generator.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import json
+import warnings
 from typing import Any
 
 from ..ir import Column, ColumnType, Schema, Table
@@ -130,7 +131,15 @@ def _column_to_prop(self, col: Column) -> dict[str, Any]:
                         prop = _json.loads(overridden)
                         return self._add_base_annotations(prop, col)
                     except (json.JSONDecodeError, ValueError):
-                        pass
+                        # A malformed override is a silent-failure trap: the
+                        # generator would fall back to a plain type string with
+                        # no indication the JSON override was discarded.
+                        warnings.warn(
+                            f"json_schema generator: type override for column "
+                            f"'{col.name}' looks like JSON but failed to parse; "
+                            f"using it as a plain type instead",
+                            stacklevel=2,
+                        )
                 prop["type"] = overridden
                 return self._add_base_annotations(prop, col)
 

src/schemaforge/parsers/sql_parser.py

@@ -4,6 +4,7 @@
 
 import contextlib
 import re
+import warnings
 from typing import Any
 
 from ..ir import Column, ColumnType, EnumType, Index, Schema, Table
@@ -146,10 +147,17 @@ def _parse_create_table(self, stmt: str) -> Table | None:
                     table.indexes.append(idx)
             elif upper.startswith("PRIMARY KEY"):
                 pass  # PK handled via column constraints
-            elif upper.startswith("CONSTRAINT"):
-                pass  # Foreign keys, etc.
-            elif upper.startswith("FOREIGN KEY") or upper.startswith("CHECK"):
-                pass
+            elif upper.startswith("CONSTRAINT") or upper.startswith(
+                "FOREIGN KEY"
+            ) or upper.startswith("CHECK"):
+                # Silent drops here are a correctness trap: schemas that differ
+                # only in FK/CHECK constraints would compare as equivalent.
+                # Surface the loss instead of swallowing it.
+                warnings.warn(
+                    f"SQL parser: ignored unsupported table constraint in "
+                    f"'{table.name}': {defn[:60]}",
+                    stacklevel=3,
+                )
             else:
                 col = self._parse_column_def(defn)
                 if col:

tests/test_silent_failures.py

@@ -0,0 +1,86 @@
+"""Silent-failure regression tests: unsupported constructs must warn, not vanish."""
+
+from __future__ import annotations
+
+import pytest
+import warnings
+from schemaforge.generators.json_schema_generator import JSONSchemaGenerator
+from schemaforge.parsers.sql_parser import SQLParser
+from schemaforge.type_config import TypeConfig
+
+
+def test_sql_parser_warns_on_foreign_key_constraint() -> None:
+    sql = """
+    CREATE TABLE orders (
+        id INTEGER PRIMARY KEY,
+        customer_id INTEGER,
+        FOREIGN KEY (customer_id) REFERENCES customers(id)
+    );
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        SQLParser().parse(sql)
+    assert any("FOREIGN KEY" in str(w.message) for w in caught)
+
+
+def test_sql_parser_warns_on_named_check_constraint() -> None:
+    sql = """
+    CREATE TABLE products (
+        price NUMERIC,
+        CONSTRAINT positive_price CHECK (price >= 0)
+    );
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        SQLParser().parse(sql)
+    assert any("CHECK" in str(w.message) for w in caught)
+
+
+def test_sql_parser_no_warning_for_plain_table() -> None:
+    sql = """
+    CREATE TABLE users (
+        id INTEGER PRIMARY KEY,
+        email VARCHAR(255) NOT NULL
+    );
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        SQLParser().parse(sql)
+    assert not caught
+
+
+def test_json_schema_generator_warns_on_malformed_override() -> None:
+    from schemaforge.ir import Column, ColumnType
+
+    class MalformedOverrideCfg(TypeConfig):
+        """Returns an override that looks like JSON but fails to parse."""
+
+        def get_override(self, col, fmt, type_args=None):  # noqa: D102
+            return "{broken json}"
+
+    col = Column(name="email", type=ColumnType.STRING, nullable=True)
+    gen = JSONSchemaGenerator(type_config=MalformedOverrideCfg())
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        gen._column_to_prop(col)
+    assert any("failed to parse" in str(w.message) for w in caught)
+
+
+def _make_column(name: str):
+    from schemaforge.ir import Column, ColumnType
+
+    return Column(name=name, type=ColumnType.STRING, nullable=True)
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "CREATE TABLE t (a INT, CONSTRAINT fk_a FOREIGN KEY (a) REFERENCES o(a));",
+        "CREATE TABLE t (a INT, FOREIGN KEY (a) REFERENCES o(a));",
+    ],
+)
+def test_warning_includes_table_name(sql: str) -> None:
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        SQLParser().parse(sql)
+    assert any("'t'" in str(w.message) for w in caught)