feat(rules): add W025 assertion-malformed

CHANGELOG.md

@@ -39,6 +39,17 @@ Four new rules added since v0.7.0 (W021, E009, T006, W024) and a meaningful cove
   Jinja preprocessing and the remaining six rules in the ADR are
   intentionally deferred to follow-up PRs.
 
+- **W025 `assertion-malformed`** - warns when an `-- @assert:` comment
+  carries a predicate that does not match the sql-sop assertion grammar.
+  sql-sop owns the syntax but does not execute the assertion; execution
+  is delegated to dbt tests, a downstream runner, or a future companion
+  tool. v1 grammar covers `row_count <op> <int>`, `unique(<col>)`,
+  `not_null(<col>)`, and `<col> <op> <literal>`. The rule anchors `--`
+  to start-of-line so `-- @assert:` mentioned inside a longer prose
+  comment does not produce a false positive. See the ADR proposal for
+  scope discipline (no execution, no DB connection, stays in the
+  rule-based hard line from `GOVERNANCE.md`).
+
 - **W021 `having-without-group-by`** - warns when `HAVING` appears
   without a preceding `GROUP BY` in the same query block. The query
   becomes a single implicit group, almost always a typo for `WHERE`.

README.md

@@ -33,8 +33,8 @@
 
 | | |
 |---|---|
-| Rules | 43 (10 errors, 28 warnings, 5 Python-source); 48 with `--contract` |
-| Tests | 210 |
+| Rules | 44 (10 errors, 29 warnings, 5 Python-source); 49 with `--contract` |
+| Tests | 277 |
 | Coverage | 86% |
 | Scan speed | 0.08s across 200 files |
 | PyPI downloads | 500+/month |
@@ -141,7 +141,7 @@
 # .pre-commit-config.yaml
 repos:
   - repo: https://github.com/Pawansingh3889/sql-guard
     rev: v0.7.0
     hooks:
       - id: sql-guard
         args: [--severity, error]  # only block on errors locally
@@ -248,6 +248,7 @@
 | W022 | `cross-join-explicit` | `FROM products CROSS JOIN regions` -- Cartesian product, confirm intent |
 | W023 | `scalar-udf-in-where` | `WHERE dbo.fn_X(col) = 1` -- row-by-row predicate evaluation |
 | W024 | `select-distinct-suspicious` | `SELECT DISTINCT a, b FROM x JOIN y ON ...` -- DISTINCT often masks a missing join condition or GROUP BY |
+| W025 | `assertion-malformed` | `-- @assert: <predicate>` comment whose predicate does not match the sql-sop grammar (`row_count <op> <int>`, `unique(<col>)`, `not_null(<col>)`, `<col> <op> <literal>`) |
 
 
 ### Structural (v0.3.0+, sqlparse-based)

sql_guard/rules/__init__.py

@@ -15,6 +15,7 @@
     UpdateWithoutWhere,
 )
 from sql_guard.rules.warnings import (
+    AssertionMalformed,
     CaseWithoutElse,
     CommentedOutCode,
     CountDistinctUnbounded,
@@ -112,6 +113,7 @@
     HavingWithoutGroupBy(),
     CrossJoinExplicit(),
     ScalarUdfInWhere(),
+    AssertionMalformed(),
     # Structural (S001-S003)
     ImplicitCrossJoin(),
     DeeplyNestedSubquery(),

sql_guard/rules/warnings.py

@@ -1,4 +1,4 @@
-"""Warning rules (W001-W023) -- advisory, don't block commits by default."""
+"""Warning rules (W001-W025) -- advisory, don't block commits by default."""
 
 from __future__ import annotations
 
@@ -797,3 +797,75 @@ def check_line(self, line: str, line_number: int, file: str) -> Finding | None:
                 suggestion="If intentional, suppress with '-- sql-guard: disable=W022' on the same line",
             )
         return None
+
+
+class AssertionMalformed(Rule):
+    """W025: ``-- @assert:`` comment with malformed predicate.
+
+    sql-sop defines a small predicate grammar so downstream tooling (or
+    you, reading the file) can mechanically pick up data assertions
+    embedded in SQL. The rule does not execute anything -- it only
+    checks that the predicate parses. Execution belongs in dbt tests,
+    a downstream runner, or a future companion tool.
+
+    Grammar (v1):
+        @assert: row_count <op> <int>
+               | unique(<col>)
+               | not_null(<col>)
+               | <col> <op> <literal>
+        op := = | != | < | <= | > | >=
+
+    Well-formed examples::
+
+        -- @assert: row_count > 0
+        -- @assert: unique(batch_id)
+        -- @assert: not_null(weight)
+        -- @assert: weight > 0
+
+    The rule only flags the malformed shape. Free-form text after
+    `-- @assert:` is the failure mode the rule is designed to catch.
+    """
+
+    id = "W025"
+    name = "assertion-malformed"
+    severity = "warning"
+    description = "-- @assert: predicate does not match the sql-sop assertion grammar"
+
+    # The `--` must open the comment at the start of the line (optionally
+    # indented). This avoids matching `-- @assert:` when it appears as a
+    # substring inside another `-- ...` comment (e.g. when the rule is
+    # documented in a SQL file).
+    _assert_line = Rule._compile(r"^\s*--\s*@assert\s*:\s*(.+?)\s*$")
+    # Column grammar for v1: a bare identifier, optionally qualified with
+    # one dot (e.g. `weight` or `orders.weight`). Two-dot forms such as
+    # `schema.table.column` fall through to the malformed branch.
+    # Quoted identifiers ("col", [col], `col`) are out of scope for v1 --
+    # see the W025 ADR for the rationale.
+    _well_formed = Rule._compile(
+        r"^("
+        r"row_count\s*(=|!=|<=|>=|<|>)\s*\d+"
+        r"|unique\s*\(\s*[a-zA-Z_]\w*(?:\.[a-zA-Z_]\w*)?\s*\)"
+        r"|not_null\s*\(\s*[a-zA-Z_]\w*(?:\.[a-zA-Z_]\w*)?\s*\)"
+        r"|[a-zA-Z_]\w*(?:\.[a-zA-Z_]\w*)?\s*(=|!=|<=|>=|<|>)\s*"
+        r"('[^']*'|\"[^\"]*\"|-?\d+(\.\d+)?)"
+        r")$"
+    )
+
+    def check_line(self, line: str, line_number: int, file: str) -> Finding | None:
+        match = self._assert_line.search(line)
+        if not match:
+            return None
+        predicate = match.group(1).strip()
+        if self._well_formed.match(predicate):
+            return None
+        return Finding(
+            rule_id=self.id,
+            severity=self.severity,
+            file=file,
+            line=line_number,
+            message=f"-- @assert: predicate is malformed: {predicate!r}",
+            suggestion=(
+                "Use one of: 'row_count <op> <int>', 'unique(<col>)', "
+                "'not_null(<col>)', '<col> <op> <literal>'"
+            ),
+        )

tests/fixtures/assertions.sql

@@ -0,0 +1,16 @@
+-- Fixture for W025 assertion-malformed.
+-- sql-sop defines a small predicate grammar for `-- @assert:` comments.
+-- Each block below shows the kind of malformed predicate W025 fires on.
+-- Well-formed cases live in the unit tests, not here.
+
+-- W025: predicate is just a bare identifier, no operator
+-- @assert: row_count
+
+-- W025: numeric operator with non-numeric right-hand side
+-- @assert: row_count > zero
+
+-- W025: unique() form without parentheses
+-- @assert: unique batch_id
+
+-- W025: freeform prose in place of a predicate
+-- @assert: weight is positive

tests/test_fluent.py

@@ -102,8 +102,8 @@ def test_scan_file_clean(self) -> None:
 class TestScanDir:
     def test_scan_dir(self) -> None:
         result = SqlGuard().scan_dir(FIXTURES)
-        # errors.sql, warnings.sql, clean.sql, contract_drift.sql
-        assert result.files_checked == 4
+        # errors.sql, warnings.sql, clean.sql, contract_drift.sql, assertions.sql
+        assert result.files_checked == 5
         assert len(result.findings) > 0
 
 

tests/test_new_rules.py

@@ -5,6 +5,7 @@
 from sql_guard.rules.errors import AlterAddNotNullNoDefault, DropColumn
 from sql_guard.rules.tsql import CreateIndexWithoutOnline
 from sql_guard.rules.warnings import (
+    AssertionMalformed,
     CountDistinctUnbounded,
     CrossJoinExplicit,
     LeadingWildcardLike,
@@ -470,3 +471,140 @@ def test_w024_message_mentions_join_or_grouping():
     assert finding is not None
     msg = finding.message.lower()
     assert "join" in msg or "grouping" in msg
+
+
+# W025 assertion-malformed ---------------------------------------------------
+
+
+def test_w025_passes_on_row_count_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: row_count > 0") is None
+
+
+def test_w025_passes_on_row_count_with_large_integer():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: row_count >= 100") is None
+
+
+def test_w025_passes_on_unique_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: unique(batch_id)") is None
+
+
+def test_w025_passes_on_not_null_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: not_null(weight)") is None
+
+
+def test_w025_passes_on_column_literal_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: weight > 0") is None
+
+
+def test_w025_passes_on_column_string_literal_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: status = 'paid'") is None
+
+
+def test_w025_flags_predicate_with_no_operator():
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: row_count")
+    assert finding is not None
+    assert finding.rule_id == "W025"
+    assert finding.severity == "warning"
+
+
+def test_w025_flags_non_numeric_rhs():
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: row_count > zero")
+    assert finding is not None
+    assert finding.rule_id == "W025"
+
+
+def test_w025_flags_unique_without_parens():
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: unique batch_id")
+    assert finding is not None
+    assert finding.rule_id == "W025"
+
+
+def test_w025_flags_freeform_text():
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: weight is positive")
+    assert finding is not None
+    assert finding.rule_id == "W025"
+
+
+def test_w025_does_not_fire_on_non_assert_comment():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- this is just a comment") is None
+
+
+def test_w025_does_not_fire_on_plain_sql():
+    rule = AssertionMalformed()
+    assert _line(rule, "SELECT * FROM users WHERE id = 1;") is None
+
+
+def test_w025_does_not_fire_on_assert_mentioned_inside_other_comment():
+    # `-- @assert:` appearing as a substring inside a longer comment is
+    # not an assertion -- it is prose describing the feature. The rule
+    # anchors `--` to start-of-line specifically to avoid this case.
+    rule = AssertionMalformed()
+    assert (
+        _line(
+            rule,
+            "-- sql-sop reads `-- @assert: row_count > 0` comments",
+        )
+        is None
+    )
+
+
+def test_w025_passes_on_qualified_unique_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: unique(orders.batch_id)") is None
+
+
+def test_w025_passes_on_qualified_not_null_predicate():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: not_null(production.weight)") is None
+
+
+def test_w025_passes_on_qualified_column_comparison():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: orders.total > 0") is None
+
+
+def test_w025_flags_double_qualified_column():
+    # `schema.table.column` is two dots; v1 supports one. Anything
+    # deeper falls through to malformed.
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: schema.table.column > 0")
+    assert finding is not None
+    assert finding.rule_id == "W025"
+
+
+def test_w025_tolerates_extra_spaces_after_dashes():
+    rule = AssertionMalformed()
+    assert _line(rule, "--   @assert: unique(id)") is None
+
+
+def test_w025_tolerates_extra_spaces_around_colon():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert:   unique(id)") is None
+
+
+def test_w025_tolerates_leading_indent_on_comment():
+    rule = AssertionMalformed()
+    assert _line(rule, "    -- @assert: unique(id)") is None
+
+
+def test_w025_tolerates_trailing_whitespace():
+    rule = AssertionMalformed()
+    assert _line(rule, "-- @assert: unique(id)   ") is None
+
+
+def test_w025_message_includes_offending_predicate():
+    rule = AssertionMalformed()
+    finding = _line(rule, "-- @assert: weight is positive")
+    assert finding is not None
+    assert "weight is positive" in finding.message

tests/test_rules.py

@@ -18,19 +18,19 @@
 
 class TestRuleRegistry:
     def test_all_rules_loaded(self) -> None:
-        assert len(ALL_RULES) == 42
+        assert len(ALL_RULES) == 43
 
     def test_11_errors(self) -> None:
         # 9 E-series + 2 T-series (T002 xp-cmdshell, T004 deprecated-outer-join).
         errors = [r for r in ALL_RULES if r.severity == "error"]
         assert len(errors) == 11
 
     def test_31_warnings(self) -> None:
-        # 24 W-series + 3 S-series + 4 T-series (T001 with-nolock,
+        # 25 W-series + 3 S-series + 4 T-series (T001 with-nolock,
         # T003 cursor-declaration, T005 create-index-without-online,
         # T006 select-into-without-typed-fields).
         warnings = [r for r in ALL_RULES if r.severity == "warning"]
-        assert len(warnings) == 31
+        assert len(warnings) == 32
 
     def test_unique_ids(self) -> None:
         ids = [r.id for r in ALL_RULES]
@@ -364,8 +364,8 @@ def test_no_findings_on_clean(self) -> None:
 class TestChecker:
     def test_files_checked_count(self) -> None:
         result = check([str(FIXTURES)])
-        # errors.sql, warnings.sql, clean.sql, contract_drift.sql
-        assert result.files_checked == 4
+        # errors.sql, warnings.sql, clean.sql, contract_drift.sql, assertions.sql
+        assert result.files_checked == 5
 
     def test_duration_tracked(self) -> None:
         result = check([str(FIXTURES)])