better test for table identifiers and nested CTEs

tests/test_custom_monitors.py

@@ -49,6 +49,30 @@ def test_add_row_limit(ql):
     assert len(rows) == 5
 
 
+@pytest.mark.template(func="add_row_limit_template")
+def test_add_row_limit_with_cte(ql):
+    """Add LIMIT to a query that contains a CTE.
+
+    The agent framework passes full queries (CTE + SELECT) to add_row_limit_template.
+    Implementations that wrap the query in a subquery (e.g. SELECT TOP N FROM ({{ query }}))
+    will break because SQL Server and others don't allow CTEs inside subqueries.
+    """
+    data = [{"val": i} for i in range(10)]
+    cte, alias = ql.make_data_source(data)
+    all_fields = ql.render(ql.templates.all_fields_expression_template)
+    from_clause = ql.render(ql.templates.add_from_clause_template, from_expression=alias)
+    select_clause = ql.render(ql.templates.add_select_clause_template, select_expressions=[all_fields])
+    full_query = f"{cte} {select_clause} {from_clause}"
+
+    limited_query = ql.render(
+        ql.templates.add_row_limit_template,
+        query=full_query,
+        limit=5,
+    )
+    rows = ql.execute(limited_query)
+    assert len(rows) == 5
+
+
 @pytest.mark.template(func="get_count_all_expression_template")
 def test_count_all_expression(ql):
     """COUNT(*) on CTE, verify row count."""

tests/test_metadata_collection.py

@@ -290,6 +290,28 @@ def test_fetch_tables_with_table_names_filter(connector, templates, database, sc
     )
 
 
+@pytest.mark.template(func="get_table_identifier_template")
+def test_table_identifier_is_queryable(connector, templates, tables):
+    """Verify that get_table_identifier_template produces an identifier the database accepts.
+
+    The agent uses this identifier in every monitor query (e.g. SELECT ... FROM <identifier>).
+    If the format is wrong (e.g. three-part naming on a database that doesn't support it),
+    every monitor will fail at runtime. This test catches that early by picking a real table
+    from metadata and executing a zero-row query against it.
+    """
+    # Pick a concrete table from metadata
+    sample = next((t for t in tables if t.table_type == "table"), tables[0])
+
+    identifier = templates.render_template(
+        templates.get_table_identifier_template,
+        _optional_vars={"database"},
+        database=sample.database_name,
+        schema=sample.schema_name,
+        table=sample.table_name,
+    )
+    connector.execute_and_fetch_all(f"SELECT * FROM {identifier} WHERE 1=0")
+
+
 @pytest.mark.capability("supports_volume_rows")
 def test_volume_rows(tables):
     supports_rows = any(

tests/test_ql_prerequisites.py

@@ -133,19 +133,6 @@ def test_negate_expression(ql):
     assert result == 0
 
 
-@pytest.mark.template(func="get_table_identifier_template")
-def test_get_table_identifier(ql):
-    """Verify db.schema.table formatting."""
-    result = ql.render(
-        ql.templates.get_table_identifier_template,
-        _optional_vars={"database": "my_db"},
-        schema="my_schema",
-        table="my_table",
-    )
-    assert "my_schema" in result
-    assert "my_table" in result
-
-
 @pytest.mark.template(func="all_fields_expression_template")
 def test_all_fields_expression(ql):
     """SELECT * returns all columns."""
@@ -868,6 +855,8 @@ def test_placeholder_templates_do_not_use_jinja_field(templates):
         if method is None:
             continue
         tpl_str = method()
+        if not tpl_str:
+            continue
         if jinja_field_re.search(tpl_str):
             violations.append(name)
     assert not violations, (