Fail Neo4jOperator argument mistakes at Dag parse time

providers/neo4j/src/airflow/providers/neo4j/operators/neo4j.py

@@ -57,6 +57,10 @@ def __init__(
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
+        if sql is not None and cypher is not None:
+            raise ValueError("Cannot provide both `sql` and `cypher`. Use `cypher` only.")
+        if cypher is None and sql is None:
+            raise ValueError("Parameter `cypher` is required.")
         self.neo4j_conn_id = neo4j_conn_id
         self.cypher = cypher
         self.sql = sql
@@ -70,10 +74,10 @@ def execute(self, context: Context) -> None:
                 AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
-            if cypher is not None:
-                raise ValueError("Cannot provide both `sql` and `cypher`. Use `cypher` only.")
             cypher = self.sql
         if cypher is None:
+            # The constructor only sees whether the argument was passed; a passed field can still
+            # render to None, and the hook needs a query string.
             raise ValueError("Parameter `cypher` is required.")
 
         self.log.info("Executing: %s", cypher)

providers/neo4j/tests/unit/neo4j/operators/test_neo4j.py

@@ -66,15 +66,17 @@ def test_neo4j_operator_sql_param_is_deprecated(self, mock_hook):
             op.execute(mock.MagicMock())
         mock_hook.return_value.run.assert_called_once_with(cypher, None)
 
-    def test_neo4j_operator_both_sql_and_cypher_raises_on_execute(self):
-        op = Neo4jOperator(task_id="basic_neo4j", sql="a", cypher="b")
+    def test_neo4j_operator_both_sql_and_cypher_raises(self):
+        with pytest.raises(ValueError, match="Cannot provide both `sql` and `cypher`"):
+            Neo4jOperator(task_id="basic_neo4j", sql="a", cypher="b")
 
-        with pytest.warns(AirflowProviderDeprecationWarning):
-            with pytest.raises(ValueError, match="Cannot provide both `sql` and `cypher`"):
-                op.execute(mock.MagicMock())
+    def test_neo4j_operator_missing_cypher_raises(self):
+        with pytest.raises(ValueError, match="Parameter `cypher` is required."):
+            Neo4jOperator(task_id="basic_neo4j")
 
-    def test_neo4j_operator_missing_cypher_raises_on_execute(self):
-        op = Neo4jOperator(task_id="basic_neo4j")
+    def test_neo4j_operator_cypher_rendering_to_none_raises_on_execute(self):
+        op = Neo4jOperator(task_id="basic_neo4j", cypher="{{ missing }}")
+        op.cypher = None
 
         with pytest.raises(ValueError, match="Parameter `cypher` is required."):
             op.execute(mock.MagicMock())