Repository: apache/superset
File: tests/integration_tests/databases/api_tests.py (Python)

Code before:
                "{% if city_filter is defined %}\n"
                "    AND city = '{{ city_filter }}'\n{% endif %}\n"
                "LIMIT {{ limit | default(100) }}"
            ),
            "schema": None,
            "template_params": {},
        }

        example_db = get_example_database()
        if example_db.backend not in ("presto", "postgresql"):
            pytest.skip("Only presto and PG are implemented")

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/database/{example_db.id}/validate_sql/"
        rv = self.client.post(uri, json=request_payload)
        response = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        # Template was successfully rendered and validated
        # For valid SQL, expect either an empty list or warnings (not errors)
        result = response["result"]
        assert isinstance(result, list)
        # As a smoke check, ensure Jinja syntax was rendered
        # (no '{' or '%' in error messages)
        for error in result:
            assert "{" not in error["message"]
            assert "%" not in error["message"]

    @mock.patch.dict(
        "superset.config.SQL_VALIDATORS_BY_ENGINE",
        SQL_VALIDATORS_BY_ENGINE,
        clear=True,
    )
    def test_validate_sql_with_jinja_templates_and_params(self):
        """
        Database API: validate SQL with Jinja templates and parameters
        """
        request_payload = {
            "sql": (
                "SELECT *\nFROM birth_names\nWHERE 1=1\n"
                "{% if city_filter is defined %}\n"
                "    AND city = '{{ city_filter }}'\n"
                "{% endif %}\nLIMIT {{ limit }}"
            ),
            "schema": None,
            "template_params": {"city_filter": "New York", "limit": 50},
        }

        example_db = get_example_database()
        if example_db.backend not in ("presto", "postgresql"):
            pytest.skip("Only presto and PG are implemented")


Review comment (suggestion):
Agreed with @pizofreude here. This creates a tight coupling between the error message and the test. Let's check if we have any other way to determine jinja errors here 

Code after:
                "{% if city_filter is defined %}\n"
                "    AND city = '{{ city_filter }}'\n{% endif %}\n"
                "LIMIT {{ limit | default(100) }}"
            ),
            "schema": None,
            "template_params": {},
        }

        example_db = get_example_database()
        if example_db.backend not in ("presto", "postgresql"):
            pytest.skip("Only presto and PG are implemented")

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/database/{example_db.id}/validate_sql/"
        rv = self.client.post(uri, json=request_payload)
        response = json.loads(rv.data.decode("utf-8"))
        assert rv.status_code == 200
        # Template was successfully rendered and validated
        # so a valid query returns an empty result list
        result = response["result"]
        assert isinstance(result, list)
        assert len(result) == 0

    @mock.patch.dict(
        "superset.config.SQL_VALIDATORS_BY_ENGINE",
        SQL_VALIDATORS_BY_ENGINE,
        clear=True,
    )
    def test_validate_sql_with_jinja_templates_and_params(self):
        """
        Database API: validate SQL with Jinja templates and parameters
        """
        request_payload = {
            "sql": (
                "SELECT *\nFROM birth_names\nWHERE 1=1\n"
                "{% if city_filter is defined %}\n"
                "    AND city = '{{ city_filter }}'\n"
                "{% endif %}\nLIMIT {{ limit }}"
            ),
            "schema": None,
            "template_params": {"city_filter": "New York", "limit": 50},
        }

        example_db = get_example_database()
        if example_db.backend not in ("presto", "postgresql"):
            pytest.skip("Only presto and PG are implemented")

        self.login(ADMIN_USERNAME)
        uri = f"api/v1/database/{example_db.id}/validate_sql/"
        rv = self.client.post(uri, json=request_payload)
        response = json.loads(rv.data.decode("utf-8"))