bito-code-review[bot] commented on code in PR #35018:
URL: https://github.com/apache/superset/pull/35018#discussion_r2353218311
##########
superset/daos/chart.py:
##########
@@ -35,10 +35,24 @@
logger = logging.getLogger(__name__)
+# Custom filterable fields for charts
+CHART_CUSTOM_FIELDS = {
+ "tags": ["eq", "in_", "like"],
+ "viz_type": ["eq", "in_", "like"],
+ "datasource_name": ["eq", "in_", "like"],
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Invalid field in filters</b></div>
<div id="fix">
The 'tags' field cannot be included in CHART_CUSTOM_FIELDS because it's a
relationship field (not a direct database column). The
get_filterable_columns_and_operators method only handles direct columns and
hybrid properties. Including 'tags' will cause runtime errors when the
filtering system tries to apply column-based filters on a relationship field.
Remove 'tags' from the custom fields and implement proper relationship
filtering through ChartFilter if needed.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
# Custom filterable fields for charts
CHART_CUSTOM_FIELDS = {
"viz_type": ["eq", "in_", "like"],
"datasource_name": ["eq", "in_", "like"],
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/daos/dashboard.py:
##########
@@ -45,10 +45,24 @@
logger = logging.getLogger(__name__)
+# Custom filterable fields for dashboards
+DASHBOARD_CUSTOM_FIELDS = {
+ "tags": ["eq", "in_", "like"],
+ "owner": ["eq", "in_"],
+ "published": ["eq"],
+}
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Field name mismatch</b></div>
<div id="fix">
The custom field mapping uses `"owner"` but the Dashboard model defines the
relationship as `"owners"` (plural). This mismatch will cause filtering by
owner to fail since the filter mechanism won't find a field named "owner".
Change to `"owners"` to match the actual relationship name defined in the
Dashboard model.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
DASHBOARD_CUSTOM_FIELDS = {
"tags": ["eq", "in_", "like"],
"owners": ["eq", "in_"],
"published": ["eq"],
}
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/daos/base.py:
##########
@@ -83,80 +202,167 @@ def find_by_id_or_uuid(
return None
@classmethod
- def find_by_id(
- cls,
- model_id: str | int,
- skip_base_filter: bool = False,
- ) -> T | None:
+ def _apply_base_filter(
+ cls, query: Any, skip_base_filter: bool = False, data_model: Any = None
+ ) -> Any:
"""
- Find a model by id, if defined applies `base_filter`
+ Apply the base_filter to the query if it exists and skip_base_filter
is False.
"""
- query = db.session.query(cls.model_cls)
if cls.base_filter and not skip_base_filter:
- data_model = SQLAInterface(cls.model_cls, db.session)
+ if data_model is None:
+ data_model = SQLAInterface(cls.model_cls, db.session)
query = cls.base_filter( # pylint: disable=not-callable
cls.id_column_name, data_model
).apply(query, None)
- id_column = getattr(cls.model_cls, cls.id_column_name)
+ return query
+
+ @classmethod
+ def _convert_value_for_column(cls, column: Any, value: Any) -> Any:
+ """
+ Convert a value to the appropriate type for a given SQLAlchemy column.
+
+ Args:
+ column: SQLAlchemy column object
+ value: Value to convert
+
+ Returns:
+ Converted value or None if conversion fails
+ """
+ if (
+ hasattr(column.type, "python_type")
+ and column.type.python_type == uuid_lib.UUID
+ ):
+ if isinstance(value, str):
+ try:
+ return uuid_lib.UUID(value)
+ except (ValueError, AttributeError):
+ return None
+ return value
+
+ @classmethod
+ def _find_by_column(
+ cls,
+ column_name: str,
+ value: str | int,
+ skip_base_filter: bool = False,
+ ) -> T | None:
+ """
+ Private method to find a model by any column value.
+
+ Args:
+ column_name: Name of the column to search by
+ value: Value to search for
+ skip_base_filter: Whether to skip base filtering
+
+ Returns:
+ Model instance or None if not found
+ """
+ query = db.session.query(cls.model_cls)
+ query = cls._apply_base_filter(query, skip_base_filter)
+
+ if not hasattr(cls.model_cls, column_name):
+ return None
+
+ column = getattr(cls.model_cls, column_name)
+ converted_value = cls._convert_value_for_column(column, value)
+ if converted_value is None:
+ return None
+
try:
- return query.filter(id_column == model_id).one_or_none()
+ return query.filter(column == converted_value).one_or_none()
except StatementError:
# can happen if int is passed instead of a string or similar
return None
+ @classmethod
+ def find_by_id(
+ cls,
+ model_id: str | int,
+ skip_base_filter: bool = False,
+ id_column: str | None = None,
+ ) -> T | None:
+ """
+ Find a model by ID using specified or default ID column.
+
+ Args:
+ model_id: ID value to search for
+ skip_base_filter: Whether to skip base filtering
+ id_column: Column name to use (defaults to cls.id_column_name)
+
+ Returns:
+ Model instance or None if not found
+ """
+ column = id_column or cls.id_column_name
+ return cls._find_by_column(column, model_id, skip_base_filter)
+
@classmethod
def find_by_ids(
cls,
- model_ids: list[str] | list[int],
+ model_ids: Sequence[str | int],
skip_base_filter: bool = False,
+ id_column: str | None = None,
) -> list[T]:
"""
Find a List of models by a list of ids, if defined applies
`base_filter`
+
+ :param model_ids: List of IDs to find
+ :param skip_base_filter: If true, skip applying the base filter
+ :param id_column: Optional column name to use for ID lookup
+ (defaults to id_column_name)
"""
- id_col = getattr(cls.model_cls, cls.id_column_name, None)
+ column = id_column or cls.id_column_name
+ id_col = getattr(cls.model_cls, column, None)
if id_col is None or not model_ids:
return []
- query = db.session.query(cls.model_cls).filter(id_col.in_(model_ids))
- if cls.base_filter and not skip_base_filter:
- data_model = SQLAInterface(cls.model_cls, db.session)
- query = cls.base_filter( # pylint: disable=not-callable
- cls.id_column_name, data_model
- ).apply(query, None)
+
+ # Convert IDs to appropriate types based on column type
+ converted_ids: list[str | int | uuid_lib.UUID] = []
+ for id_val in model_ids:
+ converted_value = cls._convert_value_for_column(id_col, id_val)
+ if converted_value is not None:
+ converted_ids.append(converted_value)
+
+ # Validate type consistency for better error handling
+ if len(converted_ids) > 1:
+ types_found = set(map(type, converted_ids))
+ if len(types_found) > 1:
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>ID conversion data loss</b></div>
<div id="fix">
The ID conversion logic in `find_by_ids` silently drops invalid ID
conversions by filtering out None values, which can lead to silent data loss.
When `_convert_value_for_column` returns None for failed conversions (like
invalid UUID strings), these IDs are removed from the query entirely, causing
valid-looking requests to return incomplete results. This affects downstream
callers like `ChartDAO.find_by_ids`, `DashboardDAO.find_by_ids`, and all DAO
classes that inherit from this base.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
# Convert IDs to appropriate types based on column type
converted_ids: list[str | int | uuid_lib.UUID] = []
for id_val in model_ids:
converted_value = cls._convert_value_for_column(id_col, id_val)
if converted_value is None:
# Preserve original value if conversion fails to avoid
silent data loss
converted_ids.append(id_val)
else:
converted_ids.append(converted_value)
# Validate type consistency for better error handling
if len(converted_ids) > 1:
types_found = set(map(type, converted_ids))
if len(types_found) > 1:
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
tests/integration_tests/dao/base_dao_test.py:
##########
@@ -0,0 +1,1610 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Integration tests for BaseDAO functionality.
+
+This module contains comprehensive integration tests for the BaseDAO class and
its
+subclasses, covering database operations, CRUD methods, flexible column
support,
+column operators, and error handling.
+
+Tests use an in-memory SQLite database for isolation and to replicate the unit
test
+environment behavior. User model deletions are avoided due to circular
dependency
+constraints with self-referential foreign keys.
+"""
+
+import datetime
+import time
+import uuid
+
+import pytest
+from flask_appbuilder.models.filters import BaseFilter
+from flask_appbuilder.security.sqla.models import User
+from sqlalchemy import Column, DateTime, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm.session import Session
+
+from superset.daos.base import BaseDAO, ColumnOperator, ColumnOperatorEnum
+from superset.daos.chart import ChartDAO
+from superset.daos.dashboard import DashboardDAO
+from superset.daos.database import DatabaseDAO
+from superset.daos.user import UserDAO
+from superset.extensions import db
+from superset.models.core import Database
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+
+# Create a test model for comprehensive testing
+Base = declarative_base()
+
+
+class ExampleModel(Base): # type: ignore
+ __tablename__ = "example_model"
+ id = Column(Integer, primary_key=True)
+ uuid = Column(String(36), unique=True, nullable=False)
+ slug = Column(String(100), unique=True)
+ name = Column(String(100))
+ code = Column(String(50), unique=True)
+ created_on = Column(DateTime, default=datetime.datetime.utcnow)
+
+
+class ExampleModelDAO(BaseDAO[ExampleModel]):
+ model_cls = ExampleModel
+ id_column_name = "id"
+ base_filter = None
+
+
[email protected](autouse=True)
+def mock_g_user(app_context):
+ """Mock the flask g.user for security context."""
+ # Within app context, we can safely mock g
+ from flask import g
+
+ mock_user = User()
+ mock_user.id = 1
+ mock_user.username = "test_user"
+
+ # Set g.user directly instead of patching
+ g.user = mock_user
+ yield
+
+ # Clean up
+ if hasattr(g, "user"):
+ delattr(g, "user")
+
+
+# =============================================================================
+# Integration Tests - These tests use the actual database
+# =============================================================================
+
+
+def test_column_operator_enum_complete_coverage(user_with_data: Session) ->
None:
+ """
+ Test that every single ColumnOperatorEnum operator is covered by tests.
+ This ensures we have comprehensive test coverage for all operators.
+ """
+ # Simply verify that we can create queries with all operators
+ for operator in ColumnOperatorEnum:
+ column_operator = ColumnOperator(
+ col="username", opr=operator, value="test_value"
+ )
+ # Just check it doesn't raise an error
+ assert column_operator.opr == operator
+
+
+def test_find_by_id_with_default_column(app_context: Session) -> None:
+ """Test find_by_id with default 'id' column."""
+ # Create a user to test with
+ user = User(
+ username="test_find_by_id",
+ first_name="Test",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Find by numeric id
+ found = UserDAO.find_by_id(user.id, skip_base_filter=True)
+ assert found is not None
+ assert found.id == user.id
+ assert found.username == "test_find_by_id"
+
+ # Test with non-existent id
+ not_found = UserDAO.find_by_id(999999, skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_uuid_column(app_context: Session) -> None:
+ """Test find_by_id with custom uuid column."""
+ # Create a dashboard with uuid
+ dashboard = Dashboard(
+ dashboard_title="Test UUID Dashboard",
+ slug="test-uuid-dashboard",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by uuid string using the uuid column
+ found = DashboardDAO.find_by_id(
+ str(dashboard.uuid), id_column="uuid", skip_base_filter=True
+ )
+ assert found is not None
+ assert found.uuid == dashboard.uuid
+ assert found.dashboard_title == "Test UUID Dashboard"
+
+ # Find by numeric id (should still work)
+ found_by_id = DashboardDAO.find_by_id(dashboard.id, skip_base_filter=True)
+ assert found_by_id is not None
+ assert found_by_id.id == dashboard.id
+
+ # Test with non-existent uuid
+ not_found = DashboardDAO.find_by_id(str(uuid.uuid4()),
skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_slug_column(app_context: Session) -> None:
+ """Test find_by_id with slug column fallback."""
+ # Create a dashboard with slug
+ dashboard = Dashboard(
+ dashboard_title="Test Slug Dashboard",
+ slug="test-slug-dashboard",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by slug using the slug column
+ found = DashboardDAO.find_by_id(
+ "test-slug-dashboard", id_column="slug", skip_base_filter=True
+ )
+ assert found is not None
+ assert found.slug == "test-slug-dashboard"
+ assert found.dashboard_title == "Test Slug Dashboard"
+
+ # Test with non-existent slug
+ not_found = DashboardDAO.find_by_id("non-existent-slug",
skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_invalid_column(app_context: Session) -> None:
+ """Test find_by_id returns None when column doesn't exist."""
+ # This should return None gracefully
+ result = UserDAO.find_by_id("not_a_valid_id", skip_base_filter=True)
+ assert result is None
+
+
+def test_find_by_id_skip_base_filter(app_context: Session) -> None:
+ """Test find_by_id with skip_base_filter parameter."""
+ # Create users with different active states
+ active_user = User(
+ username="active_user",
+ first_name="Active",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ inactive_user = User(
+ username="inactive_user",
+ first_name="Inactive",
+ last_name="User",
+ email="[email protected]",
+ active=False,
+ )
+ db.session.add_all([active_user, inactive_user])
+ db.session.commit()
+
+ # Without skipping base filter (if one exists)
+ found_active = UserDAO.find_by_id(active_user.id, skip_base_filter=False)
+ assert found_active is not None
+
+ # With skipping base filter
+ found_active_skip = UserDAO.find_by_id(active_user.id,
skip_base_filter=True)
+ assert found_active_skip is not None
+
+ # Both should find the user since UserDAO might not have a base filter
+ assert found_active.id == active_user.id
+ assert found_active_skip.id == active_user.id
+
+
+def test_find_by_ids_with_default_column(app_context: Session) -> None:
+ """Test find_by_ids with default 'id' column."""
+ # Create multiple users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"test_find_by_ids_{i}",
+ first_name=f"Test{i}",
+ last_name="User",
+ email=f"test{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ # Find by multiple ids
+ ids = [user.id for user in users]
+ found = UserDAO.find_by_ids(ids, skip_base_filter=True)
+ assert len(found) == 3
+ found_ids = [u.id for u in found]
+ assert set(found_ids) == set(ids)
+
+ # Test with mix of existent and non-existent ids
+ mixed_ids = [users[0].id, 999999, users[1].id]
+ found_mixed = UserDAO.find_by_ids(mixed_ids, skip_base_filter=True)
+ assert len(found_mixed) == 2
+
+ # Test with empty list
+ found_empty = UserDAO.find_by_ids([], skip_base_filter=True)
+ assert found_empty == []
+
+
+def test_find_by_ids_with_uuid_column(app_context: Session) -> None:
+ """Test find_by_ids with uuid column."""
+ # Create multiple dashboards
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Test UUID Dashboard {i}",
+ slug=f"test-uuid-dashboard-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by multiple uuids
+ uuids = [str(dashboard.uuid) for dashboard in dashboards]
+ found = DashboardDAO.find_by_ids(uuids, id_column="uuid",
skip_base_filter=True)
+ assert len(found) == 3
+ found_uuids = [str(d.uuid) for d in found]
+ assert set(found_uuids) == set(uuids)
+
+ # Test with mix of ids and uuids - search separately by column
+ found_by_id = DashboardDAO.find_by_ids([dashboards[0].id],
skip_base_filter=True)
+ found_by_uuid = DashboardDAO.find_by_ids(
+ [str(dashboards[1].uuid)], id_column="uuid", skip_base_filter=True
+ )
+ assert len(found_by_id) == 1
+ assert len(found_by_uuid) == 1
+
+
+def test_find_by_ids_with_slug_column(app_context: Session) -> None:
+ """Test find_by_ids with slug column."""
+ # Create multiple dashboards
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Test Slug Dashboard {i}",
+ slug=f"test-slug-dashboard-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by multiple slugs
+ slugs = [dashboard.slug for dashboard in dashboards]
+ found = DashboardDAO.find_by_ids(slugs, id_column="slug",
skip_base_filter=True)
+ assert len(found) == 3
+ found_slugs = [d.slug for d in found]
+ assert set(found_slugs) == set(slugs)
+
+
+def test_find_by_ids_with_invalid_column(app_context: Session) -> None:
+ """Test find_by_ids returns empty list when column doesn't exist."""
+ # This should return empty list gracefully
+ result = UserDAO.find_by_ids(["not_a_valid_id"], skip_base_filter=True)
+ assert result == []
+
+
+def test_find_by_ids_skip_base_filter(app_context: Session) -> None:
+ """Test find_by_ids with skip_base_filter parameter."""
+ # Create users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"test_skip_filter_{i}",
+ first_name=f"Test{i}",
+ last_name="User",
+ email=f"test{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ ids = [user.id for user in users]
+
+ # Without skipping base filter
+ found_no_skip = UserDAO.find_by_ids(ids, skip_base_filter=False)
+ assert len(found_no_skip) == 3
+
+ # With skipping base filter
+ found_skip = UserDAO.find_by_ids(ids, skip_base_filter=True)
+ assert len(found_skip) == 3
+
+
+def test_base_dao_create_with_item(app_context: Session) -> None:
+ """Test BaseDAO.create with an item parameter."""
+ # Create a user item
+ user = User(
+ username="created_with_item",
+ first_name="Created",
+ last_name="Item",
+ email="[email protected]",
+ active=True,
+ )
+
+ # Create using the item
+ created = UserDAO.create(item=user)
+ assert created is not None
+ assert created.username == "created_with_item"
+ assert created.first_name == "Created"
+
+ # Verify it's in the session
+ assert created in db.session
+
+ # Commit and verify it persists
+ db.session.commit()
+
+ # Find it again to ensure it was saved
+ found = UserDAO.find_by_id(created.id, skip_base_filter=True)
+ assert found is not None
+ assert found.username == "created_with_item"
+
+
+def test_base_dao_create_with_attributes(app_context: Session) -> None:
+ """Test BaseDAO.create with attributes parameter."""
+ # Create using attributes dict
+ attributes = {
+ "username": "created_with_attrs",
+ "first_name": "Created",
+ "last_name": "Attrs",
+ "email": "[email protected]",
+ "active": True,
+ }
+
+ created = UserDAO.create(attributes=attributes)
+ assert created is not None
+ assert created.username == "created_with_attrs"
+ assert created.email == "[email protected]"
+
+ # Commit and verify
+ db.session.commit()
+ found = UserDAO.find_by_id(created.id, skip_base_filter=True)
+ assert found is not None
+ assert found.username == "created_with_attrs"
+
+
+def test_base_dao_create_with_both_item_and_attributes(app_context: Session)
-> None:
+ """Test BaseDAO.create with both item and attributes (override
behavior)."""
+ # Create a user item
+ user = User(
+ username="item_username",
+ first_name="Item",
+ last_name="User",
+ email="[email protected]",
+ active=False,
+ )
+
+ # Override some attributes
+ attributes = {
+ "username": "override_username",
+ "active": True,
+ }
+
+ created = UserDAO.create(item=user, attributes=attributes)
+ assert created is not None
+ assert created.username == "override_username" # Should be overridden
+ assert created.active is True # Should be overridden
+ assert created.first_name == "Item" # Should keep original
+ assert created.last_name == "User" # Should keep original
+
+ db.session.commit()
+
+
+def test_base_dao_update_with_item(app_context: Session) -> None:
+ """Test BaseDAO.update with an item parameter."""
+ # Create a user first
+ user = User(
+ username="update_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Update the user
+ user.first_name = "Updated"
+ updated = UserDAO.update(item=user)
+ assert updated is not None
+ assert updated.first_name == "Updated"
+
+ db.session.commit()
+
+ # Verify the update persisted
+ found = UserDAO.find_by_id(user.id, skip_base_filter=True)
+ assert found is not None
+ assert found.first_name == "Updated"
+
+
+def test_base_dao_update_with_attributes(app_context: Session) -> None:
+ """Test BaseDAO.update with attributes parameter."""
+ # Create a user first
+ user = User(
+ username="update_attrs_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Update using attributes
+ attributes = {"first_name": "Updated", "last_name": "Attr User"}
+ updated = UserDAO.update(item=user, attributes=attributes)
+ assert updated is not None
+ assert updated.first_name == "Updated"
+ assert updated.last_name == "Attr User"
+
+ db.session.commit()
+
+
+def test_base_dao_update_detached_item(app_context: Session) -> None:
+ """Test BaseDAO.update with a detached item."""
+ # Create a user first
+ user = User(
+ username="detached_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ user_id = user.id
+
+ # Expunge to detach from session
+ db.session.expunge(user)
+
+ # Update the detached user
+ user.first_name = "Updated Detached"
+ updated = UserDAO.update(item=user)
+ assert updated is not None
+ assert updated.first_name == "Updated Detached"
+
+ db.session.commit()
+
+ # Verify the update persisted
+ found = UserDAO.find_by_id(user_id, skip_base_filter=True)
+ assert found is not None
+ assert found.first_name == "Updated Detached"
+
+
+def test_base_dao_delete_single_item(app_context: Session) -> None:
+ """Test BaseDAO.delete with a single item."""
+ # Create a dashboard instead of user to avoid circular dependencies
+ dashboard = Dashboard(
+ dashboard_title="Delete Test",
+ slug="delete-test",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ dashboard_id = dashboard.id
+
+ DashboardDAO.delete([dashboard])
+ db.session.commit()
+
+ # Verify it's gone
+ found = DashboardDAO.find_by_id(dashboard_id, skip_base_filter=True)
+ assert found is None
+
+
+def test_base_dao_delete_multiple_items(app_context: Session) -> None:
+ """Test BaseDAO.delete with multiple items."""
+ # Create multiple dashboards instead of users to avoid circular
dependencies
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Delete Multi {i}",
+ slug=f"delete-multi-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ dashboard_ids = [dashboard.id for dashboard in dashboards]
+
+ DashboardDAO.delete(dashboards)
+ db.session.commit()
+
+ # Verify they're all gone
+ for dashboard_id in dashboard_ids:
+ found = DashboardDAO.find_by_id(dashboard_id, skip_base_filter=True)
+ assert found is None
+
+
+def test_base_dao_delete_empty_list(app_context: Session) -> None:
+ """Test BaseDAO.delete with empty list."""
+ # Should not raise any errors
+ UserDAO.delete([])
+ db.session.commit()
+ # Just ensuring no exception is raised
+
+
+def test_base_dao_find_all(app_context: Session) -> None:
+ """Test BaseDAO.find_all method."""
+ # Create some users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"find_all_{i}",
+ first_name=f"Find{i}",
+ last_name="All",
+ email=f"findall{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ # Find all users
+ all_users = UserDAO.find_all()
+ assert len(all_users) >= 3 # At least our 3 users
+
+ # Check our users are in the results
+ usernames = [u.username for u in all_users]
+ for user in users:
+ assert user.username in usernames
+
+
+def test_base_dao_find_one_or_none(app_context: Session) -> None:
+ """Test BaseDAO.find_one_or_none method."""
+ # Create users with specific criteria
+ user1 = User(
+ username="unique_username_123",
+ first_name="Unique",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ user2 = User(
+ username="another_user",
+ first_name="Another",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ user3 = User(
+ username="third_user",
+ first_name="Another", # Same first name as user2
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add_all([user1, user2, user3])
+ db.session.commit()
+
+ # Find one with unique criteria
+ found = UserDAO.find_one_or_none(username="unique_username_123")
+ assert found is not None
+ assert found.username == "unique_username_123"
+
+ # Find none with non-existent criteria
+ not_found = UserDAO.find_one_or_none(username="non_existent_user")
+ assert not_found is None
+
+ db.session.commit()
+
+
+def test_base_dao_list_returns_results(user_with_data: Session) -> None:
+ """Test that BaseDAO.list returns results and total."""
+ results, total = UserDAO.list()
+ assert isinstance(results, list)
+ assert isinstance(total, int)
+ assert total >= 1 # At least the fixture user
+
+
+def test_base_dao_list_with_column_operators(user_with_data: Session) -> None:
+ """Test BaseDAO.list with column operators."""
+ column_operators = [
+ ColumnOperator(col="username", opr=ColumnOperatorEnum.eq,
value="testuser")
+ ]
+ results, total = UserDAO.list(column_operators=column_operators)
+ assert total == 1
+ assert results[0].username == "testuser"
+
+
+def test_base_dao_list_with_non_matching_column_operator(
+ user_with_data: Session,
+) -> None:
+ """Test BaseDAO.list with non-matching column operator."""
+ column_operators = [
+ ColumnOperator(
+ col="username", opr=ColumnOperatorEnum.eq, value="nonexistentuser"
+ )
+ ]
+ results, total = UserDAO.list(column_operators=column_operators)
+ assert total == 0
+ assert results == []
+
+
+def test_base_dao_count_returns_value(user_with_data: Session) -> None:
+ """Test that BaseDAO.count returns correct count."""
+ count = UserDAO.count()
+ assert isinstance(count, int)
+ assert count >= 1 # At least the fixture user
+
+
+def test_base_dao_count_with_column_operators(user_with_data: Session) -> None:
+ """Test BaseDAO.count with column operators."""
+ # Count with matching operator
+ column_operators = [
+ ColumnOperator(col="username", opr=ColumnOperatorEnum.eq,
value="testuser")
+ ]
+ count = UserDAO.count(column_operators=column_operators)
+ assert count == 1
+
+ # Count with non-matching operator
+ column_operators = [
+ ColumnOperator(
+ col="username", opr=ColumnOperatorEnum.eq, value="nonexistentuser"
+ )
+ ]
+ count = UserDAO.count(column_operators=column_operators)
+ assert count == 0
+
+
+def test_base_dao_list_ordering(user_with_data: Session) -> None:
+ """Test BaseDAO.list with ordering."""
+ # Create additional users with predictable names
+ users = []
+ for i in range(3):
+ user = User(
+ # Let database auto-generate IDs to avoid conflicts
+ username=f"order_test_{i}",
+ first_name=f"Order{i}",
+ last_name="Test",
+ email=f"order{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ user_with_data.add(user)
+ user_with_data.commit()
+
+ # Test ascending order
+ results_asc, _ = UserDAO.list(
+ order_column="username", order_direction="asc", page_size=100
+ )
+ usernames_asc = [r.username for r in results_asc]
+ # Check that our test users are in order
+ assert usernames_asc.index("order_test_0") <
usernames_asc.index("order_test_1")
+ assert usernames_asc.index("order_test_1") <
usernames_asc.index("order_test_2")
+
+ # Test descending order
+ results_desc, _ = UserDAO.list(
+ order_column="username", order_direction="desc", page_size=100
+ )
+ usernames_desc = [r.username for r in results_desc]
+ # Check that our test users are in reverse order
+ assert usernames_desc.index("order_test_2") <
usernames_desc.index("order_test_1")
+ assert usernames_desc.index("order_test_1") <
usernames_desc.index("order_test_0")
+
+ for user in users:
+ user_with_data.delete(user)
+ user_with_data.commit()
+
+
+def test_base_dao_list_paging(user_with_data: Session) -> None:
+ """Test BaseDAO.list with paging."""
+ # Create additional users for paging
+ users = []
+ for i in range(10):
+ user = User(
+ id=300 + i,
+ username=f"page_test_{i}",
+ first_name=f"Page{i}",
+ last_name="Test",
+ email=f"page{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ user_with_data.add(user)
+ user_with_data.commit()
+
+ # Test first page
+ page1_results, page1_total = UserDAO.list(page=0, page_size=5,
order_column="id")
+ assert len(page1_results) <= 5
+ assert page1_total >= 10 # At least our 10 users
+
+ # Test second page
+ page2_results, page2_total = UserDAO.list(page=1, page_size=5,
order_column="id")
+ assert len(page2_results) <= 5
+ assert page2_total >= 10
+
+ # Results should be different
+ page1_ids = [r.id for r in page1_results]
+ page2_ids = [r.id for r in page2_results]
+ assert set(page1_ids).isdisjoint(set(page2_ids)) # No overlap
+
+ for user in users:
+ user_with_data.delete(user)
+ user_with_data.commit()
+
+
+def test_base_dao_list_search(user_with_data: Session) -> None:
+ """Test BaseDAO.list with search."""
+ # Create users with searchable names
+ users = []
+ for i in range(3):
+ user = User(
+ id=400 + i,
+ username=f"searchable_{i}",
+ first_name=f"Searchable{i}",
+ last_name="User",
+ email=f"search{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ user_with_data.add(user)
+
+ # Create some non-matching users
+ for i in range(2):
+ user = User(
+ id=410 + i,
+ username=f"other_{i}",
+ first_name=f"Other{i}",
+ last_name="Person",
+ email=f"other{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ user_with_data.add(user)
+
+ user_with_data.commit()
+
+ # Search for "searchable"
+ results, total = UserDAO.list(
+ search="searchable", search_columns=["username", "first_name"]
+ )
+ assert total >= 3 # At least our 3 searchable users
+
+ # Verify all results contain "searchable"
+ for result in results:
+ assert (
+ "searchable" in result.username.lower()
+ or "searchable" in result.first_name.lower()
+ )
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Search test too brittle</b></div>
<div id="fix">
The search test assertions are too brittle when existing database users
match the search criteria. The test should verify that the created searchable
users are found rather than asserting all results contain the search term. This
prevents false failures in production environments with existing data.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
assert total >= 3 # At least our 3 searchable users
# Verify search functionality works - count how many of our test users
are found
searchable_test_users = [r for r in results if "searchable" in
(r.username.lower() + " " + r.first_name.lower())]
assert len(searchable_test_users) >= 3 # Our created test users should
be found
# Verify the search actually filtered results (total should be
reasonable)
assert total > 0 # Search returned some results
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
superset/daos/dataset.py:
##########
@@ -35,8 +35,20 @@
logger = logging.getLogger(__name__)
+# Custom filterable fields for datasets
+DATASET_CUSTOM_FIELDS = {
+ "schema": ["eq", "in_"],
+}
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Duplicate schema field definition</b></div>
<div id="fix">
The custom field "schema" in DATASET_CUSTOM_FIELDS conflicts with the
existing database column "schema" in SqlaTable. Since SqlaTable already has a
schema column (line 1141 in SqlaTable), this creates a duplicate key that will
override the actual database column definition. The base
get_filterable_columns_and_operators() method will automatically include the
schema field with proper type detection and operator mapping. Remove "schema"
from DATASET_CUSTOM_FIELDS to prevent this conflict.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
# Custom filterable fields for datasets
DATASET_CUSTOM_FIELDS = {}
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
##########
tests/integration_tests/dao/base_dao_test.py:
##########
@@ -0,0 +1,1610 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Integration tests for BaseDAO functionality.
+
+This module contains comprehensive integration tests for the BaseDAO class and
its
+subclasses, covering database operations, CRUD methods, flexible column
support,
+column operators, and error handling.
+
+Tests use an in-memory SQLite database for isolation and to replicate the unit
test
+environment behavior. User model deletions are avoided due to circular
dependency
+constraints with self-referential foreign keys.
+"""
+
+import datetime
+import time
+import uuid
+
+import pytest
+from flask_appbuilder.models.filters import BaseFilter
+from flask_appbuilder.security.sqla.models import User
+from sqlalchemy import Column, DateTime, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm.session import Session
+
+from superset.daos.base import BaseDAO, ColumnOperator, ColumnOperatorEnum
+from superset.daos.chart import ChartDAO
+from superset.daos.dashboard import DashboardDAO
+from superset.daos.database import DatabaseDAO
+from superset.daos.user import UserDAO
+from superset.extensions import db
+from superset.models.core import Database
+from superset.models.dashboard import Dashboard
+from superset.models.slice import Slice
+
+# Create a test model for comprehensive testing
+Base = declarative_base()
+
+
+class ExampleModel(Base): # type: ignore
+ __tablename__ = "example_model"
+ id = Column(Integer, primary_key=True)
+ uuid = Column(String(36), unique=True, nullable=False)
+ slug = Column(String(100), unique=True)
+ name = Column(String(100))
+ code = Column(String(50), unique=True)
+ created_on = Column(DateTime, default=datetime.datetime.utcnow)
+
+
+class ExampleModelDAO(BaseDAO[ExampleModel]):
+ model_cls = ExampleModel
+ id_column_name = "id"
+ base_filter = None
+
+
[email protected](autouse=True)
+def mock_g_user(app_context):
+ """Mock the flask g.user for security context."""
+ # Within app context, we can safely mock g
+ from flask import g
+
+ mock_user = User()
+ mock_user.id = 1
+ mock_user.username = "test_user"
+
+ # Set g.user directly instead of patching
+ g.user = mock_user
+ yield
+
+ # Clean up
+ if hasattr(g, "user"):
+ delattr(g, "user")
+
+
+# =============================================================================
+# Integration Tests - These tests use the actual database
+# =============================================================================
+
+
+def test_column_operator_enum_complete_coverage(user_with_data: Session) ->
None:
+ """
+ Test that every single ColumnOperatorEnum operator is covered by tests.
+ This ensures we have comprehensive test coverage for all operators.
+ """
+ # Simply verify that we can create queries with all operators
+ for operator in ColumnOperatorEnum:
+ column_operator = ColumnOperator(
+ col="username", opr=operator, value="test_value"
+ )
+ # Just check it doesn't raise an error
+ assert column_operator.opr == operator
+
+
+def test_find_by_id_with_default_column(app_context: Session) -> None:
+ """Test find_by_id with default 'id' column."""
+ # Create a user to test with
+ user = User(
+ username="test_find_by_id",
+ first_name="Test",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Find by numeric id
+ found = UserDAO.find_by_id(user.id, skip_base_filter=True)
+ assert found is not None
+ assert found.id == user.id
+ assert found.username == "test_find_by_id"
+
+ # Test with non-existent id
+ not_found = UserDAO.find_by_id(999999, skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_uuid_column(app_context: Session) -> None:
+ """Test find_by_id with custom uuid column."""
+ # Create a dashboard with uuid
+ dashboard = Dashboard(
+ dashboard_title="Test UUID Dashboard",
+ slug="test-uuid-dashboard",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by uuid string using the uuid column
+ found = DashboardDAO.find_by_id(
+ str(dashboard.uuid), id_column="uuid", skip_base_filter=True
+ )
+ assert found is not None
+ assert found.uuid == dashboard.uuid
+ assert found.dashboard_title == "Test UUID Dashboard"
+
+ # Find by numeric id (should still work)
+ found_by_id = DashboardDAO.find_by_id(dashboard.id, skip_base_filter=True)
+ assert found_by_id is not None
+ assert found_by_id.id == dashboard.id
+
+ # Test with non-existent uuid
+ not_found = DashboardDAO.find_by_id(str(uuid.uuid4()),
skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_slug_column(app_context: Session) -> None:
+ """Test find_by_id with slug column fallback."""
+ # Create a dashboard with slug
+ dashboard = Dashboard(
+ dashboard_title="Test Slug Dashboard",
+ slug="test-slug-dashboard",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by slug using the slug column
+ found = DashboardDAO.find_by_id(
+ "test-slug-dashboard", id_column="slug", skip_base_filter=True
+ )
+ assert found is not None
+ assert found.slug == "test-slug-dashboard"
+ assert found.dashboard_title == "Test Slug Dashboard"
+
+ # Test with non-existent slug
+ not_found = DashboardDAO.find_by_id("non-existent-slug",
skip_base_filter=True)
+ assert not_found is None
+
+
+def test_find_by_id_with_invalid_column(app_context: Session) -> None:
+ """Test find_by_id returns None when column doesn't exist."""
+ # This should return None gracefully
+ result = UserDAO.find_by_id("not_a_valid_id", skip_base_filter=True)
+ assert result is None
+
+
+def test_find_by_id_skip_base_filter(app_context: Session) -> None:
+ """Test find_by_id with skip_base_filter parameter."""
+ # Create users with different active states
+ active_user = User(
+ username="active_user",
+ first_name="Active",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ inactive_user = User(
+ username="inactive_user",
+ first_name="Inactive",
+ last_name="User",
+ email="[email protected]",
+ active=False,
+ )
+ db.session.add_all([active_user, inactive_user])
+ db.session.commit()
+
+ # Without skipping base filter (if one exists)
+ found_active = UserDAO.find_by_id(active_user.id, skip_base_filter=False)
+ assert found_active is not None
+
+ # With skipping base filter
+ found_active_skip = UserDAO.find_by_id(active_user.id,
skip_base_filter=True)
+ assert found_active_skip is not None
+
+ # Both should find the user since UserDAO might not have a base filter
+ assert found_active.id == active_user.id
+ assert found_active_skip.id == active_user.id
+
+
+def test_find_by_ids_with_default_column(app_context: Session) -> None:
+ """Test find_by_ids with default 'id' column."""
+ # Create multiple users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"test_find_by_ids_{i}",
+ first_name=f"Test{i}",
+ last_name="User",
+ email=f"test{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ # Find by multiple ids
+ ids = [user.id for user in users]
+ found = UserDAO.find_by_ids(ids, skip_base_filter=True)
+ assert len(found) == 3
+ found_ids = [u.id for u in found]
+ assert set(found_ids) == set(ids)
+
+ # Test with mix of existent and non-existent ids
+ mixed_ids = [users[0].id, 999999, users[1].id]
+ found_mixed = UserDAO.find_by_ids(mixed_ids, skip_base_filter=True)
+ assert len(found_mixed) == 2
+
+ # Test with empty list
+ found_empty = UserDAO.find_by_ids([], skip_base_filter=True)
+ assert found_empty == []
+
+
+def test_find_by_ids_with_uuid_column(app_context: Session) -> None:
+ """Test find_by_ids with uuid column."""
+ # Create multiple dashboards
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Test UUID Dashboard {i}",
+ slug=f"test-uuid-dashboard-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by multiple uuids
+ uuids = [str(dashboard.uuid) for dashboard in dashboards]
+ found = DashboardDAO.find_by_ids(uuids, id_column="uuid",
skip_base_filter=True)
+ assert len(found) == 3
+ found_uuids = [str(d.uuid) for d in found]
+ assert set(found_uuids) == set(uuids)
+
+ # Test with mix of ids and uuids - search separately by column
+ found_by_id = DashboardDAO.find_by_ids([dashboards[0].id],
skip_base_filter=True)
+ found_by_uuid = DashboardDAO.find_by_ids(
+ [str(dashboards[1].uuid)], id_column="uuid", skip_base_filter=True
+ )
+ assert len(found_by_id) == 1
+ assert len(found_by_uuid) == 1
+
+
+def test_find_by_ids_with_slug_column(app_context: Session) -> None:
+ """Test find_by_ids with slug column."""
+ # Create multiple dashboards
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Test Slug Dashboard {i}",
+ slug=f"test-slug-dashboard-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ # Find by multiple slugs
+ slugs = [dashboard.slug for dashboard in dashboards]
+ found = DashboardDAO.find_by_ids(slugs, id_column="slug",
skip_base_filter=True)
+ assert len(found) == 3
+ found_slugs = [d.slug for d in found]
+ assert set(found_slugs) == set(slugs)
+
+
+def test_find_by_ids_with_invalid_column(app_context: Session) -> None:
+ """Test find_by_ids returns empty list when column doesn't exist."""
+ # This should return empty list gracefully
+ result = UserDAO.find_by_ids(["not_a_valid_id"], skip_base_filter=True)
+ assert result == []
+
+
+def test_find_by_ids_skip_base_filter(app_context: Session) -> None:
+ """Test find_by_ids with skip_base_filter parameter."""
+ # Create users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"test_skip_filter_{i}",
+ first_name=f"Test{i}",
+ last_name="User",
+ email=f"test{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ ids = [user.id for user in users]
+
+ # Without skipping base filter
+ found_no_skip = UserDAO.find_by_ids(ids, skip_base_filter=False)
+ assert len(found_no_skip) == 3
+
+ # With skipping base filter
+ found_skip = UserDAO.find_by_ids(ids, skip_base_filter=True)
+ assert len(found_skip) == 3
+
+
+def test_base_dao_create_with_item(app_context: Session) -> None:
+ """Test BaseDAO.create with an item parameter."""
+ # Create a user item
+ user = User(
+ username="created_with_item",
+ first_name="Created",
+ last_name="Item",
+ email="[email protected]",
+ active=True,
+ )
+
+ # Create using the item
+ created = UserDAO.create(item=user)
+ assert created is not None
+ assert created.username == "created_with_item"
+ assert created.first_name == "Created"
+
+ # Verify it's in the session
+ assert created in db.session
+
+ # Commit and verify it persists
+ db.session.commit()
+
+ # Find it again to ensure it was saved
+ found = UserDAO.find_by_id(created.id, skip_base_filter=True)
+ assert found is not None
+ assert found.username == "created_with_item"
+
+
+def test_base_dao_create_with_attributes(app_context: Session) -> None:
+ """Test BaseDAO.create with attributes parameter."""
+ # Create using attributes dict
+ attributes = {
+ "username": "created_with_attrs",
+ "first_name": "Created",
+ "last_name": "Attrs",
+ "email": "[email protected]",
+ "active": True,
+ }
+
+ created = UserDAO.create(attributes=attributes)
+ assert created is not None
+ assert created.username == "created_with_attrs"
+ assert created.email == "[email protected]"
+
+ # Commit and verify
+ db.session.commit()
+ found = UserDAO.find_by_id(created.id, skip_base_filter=True)
+ assert found is not None
+ assert found.username == "created_with_attrs"
+
+
+def test_base_dao_create_with_both_item_and_attributes(app_context: Session)
-> None:
+ """Test BaseDAO.create with both item and attributes (override
behavior)."""
+ # Create a user item
+ user = User(
+ username="item_username",
+ first_name="Item",
+ last_name="User",
+ email="[email protected]",
+ active=False,
+ )
+
+ # Override some attributes
+ attributes = {
+ "username": "override_username",
+ "active": True,
+ }
+
+ created = UserDAO.create(item=user, attributes=attributes)
+ assert created is not None
+ assert created.username == "override_username" # Should be overridden
+ assert created.active is True # Should be overridden
+ assert created.first_name == "Item" # Should keep original
+ assert created.last_name == "User" # Should keep original
+
+ db.session.commit()
+
+
+def test_base_dao_update_with_item(app_context: Session) -> None:
+ """Test BaseDAO.update with an item parameter."""
+ # Create a user first
+ user = User(
+ username="update_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Update the user
+ user.first_name = "Updated"
+ updated = UserDAO.update(item=user)
+ assert updated is not None
+ assert updated.first_name == "Updated"
+
+ db.session.commit()
+
+ # Verify the update persisted
+ found = UserDAO.find_by_id(user.id, skip_base_filter=True)
+ assert found is not None
+ assert found.first_name == "Updated"
+
+
+def test_base_dao_update_with_attributes(app_context: Session) -> None:
+ """Test BaseDAO.update with attributes parameter."""
+ # Create a user first
+ user = User(
+ username="update_attrs_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ # Update using attributes
+ attributes = {"first_name": "Updated", "last_name": "Attr User"}
+ updated = UserDAO.update(item=user, attributes=attributes)
+ assert updated is not None
+ assert updated.first_name == "Updated"
+ assert updated.last_name == "Attr User"
+
+ db.session.commit()
+
+
+def test_base_dao_update_detached_item(app_context: Session) -> None:
+ """Test BaseDAO.update with a detached item."""
+ # Create a user first
+ user = User(
+ username="detached_test",
+ first_name="Original",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add(user)
+ db.session.commit()
+
+ user_id = user.id
+
+ # Expunge to detach from session
+ db.session.expunge(user)
+
+ # Update the detached user
+ user.first_name = "Updated Detached"
+ updated = UserDAO.update(item=user)
+ assert updated is not None
+ assert updated.first_name == "Updated Detached"
+
+ db.session.commit()
+
+ # Verify the update persisted
+ found = UserDAO.find_by_id(user_id, skip_base_filter=True)
+ assert found is not None
+ assert found.first_name == "Updated Detached"
+
+
+def test_base_dao_delete_single_item(app_context: Session) -> None:
+ """Test BaseDAO.delete with a single item."""
+ # Create a dashboard instead of user to avoid circular dependencies
+ dashboard = Dashboard(
+ dashboard_title="Delete Test",
+ slug="delete-test",
+ published=True,
+ )
+ db.session.add(dashboard)
+ db.session.commit()
+
+ dashboard_id = dashboard.id
+
+ DashboardDAO.delete([dashboard])
+ db.session.commit()
+
+ # Verify it's gone
+ found = DashboardDAO.find_by_id(dashboard_id, skip_base_filter=True)
+ assert found is None
+
+
+def test_base_dao_delete_multiple_items(app_context: Session) -> None:
+ """Test BaseDAO.delete with multiple items."""
+ # Create multiple dashboards instead of users to avoid circular
dependencies
+ dashboards = []
+ for i in range(3):
+ dashboard = Dashboard(
+ dashboard_title=f"Delete Multi {i}",
+ slug=f"delete-multi-{i}",
+ published=True,
+ )
+ dashboards.append(dashboard)
+ db.session.add(dashboard)
+ db.session.commit()
+
+ dashboard_ids = [dashboard.id for dashboard in dashboards]
+
+ DashboardDAO.delete(dashboards)
+ db.session.commit()
+
+ # Verify they're all gone
+ for dashboard_id in dashboard_ids:
+ found = DashboardDAO.find_by_id(dashboard_id, skip_base_filter=True)
+ assert found is None
+
+
+def test_base_dao_delete_empty_list(app_context: Session) -> None:
+ """Test BaseDAO.delete with empty list."""
+ # Should not raise any errors
+ UserDAO.delete([])
+ db.session.commit()
+ # Just ensuring no exception is raised
+
+
+def test_base_dao_find_all(app_context: Session) -> None:
+ """Test BaseDAO.find_all method."""
+ # Create some users
+ users = []
+ for i in range(3):
+ user = User(
+ username=f"find_all_{i}",
+ first_name=f"Find{i}",
+ last_name="All",
+ email=f"findall{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ db.session.add(user)
+ db.session.commit()
+
+ # Find all users
+ all_users = UserDAO.find_all()
+ assert len(all_users) >= 3 # At least our 3 users
+
+ # Check our users are in the results
+ usernames = [u.username for u in all_users]
+ for user in users:
+ assert user.username in usernames
+
+
+def test_base_dao_find_one_or_none(app_context: Session) -> None:
+ """Test BaseDAO.find_one_or_none method."""
+ # Create users with specific criteria
+ user1 = User(
+ username="unique_username_123",
+ first_name="Unique",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ user2 = User(
+ username="another_user",
+ first_name="Another",
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ user3 = User(
+ username="third_user",
+ first_name="Another", # Same first name as user2
+ last_name="User",
+ email="[email protected]",
+ active=True,
+ )
+ db.session.add_all([user1, user2, user3])
+ db.session.commit()
+
+ # Find one with unique criteria
+ found = UserDAO.find_one_or_none(username="unique_username_123")
+ assert found is not None
+ assert found.username == "unique_username_123"
+
+ # Find none with non-existent criteria
+ not_found = UserDAO.find_one_or_none(username="non_existent_user")
+ assert not_found is None
+
+ db.session.commit()
+
+
+def test_base_dao_list_returns_results(user_with_data: Session) -> None:
+ """Test that BaseDAO.list returns results and total."""
+ results, total = UserDAO.list()
+ assert isinstance(results, list)
+ assert isinstance(total, int)
+ assert total >= 1 # At least the fixture user
+
+
+def test_base_dao_list_with_column_operators(user_with_data: Session) -> None:
+ """Test BaseDAO.list with column operators."""
+ column_operators = [
+ ColumnOperator(col="username", opr=ColumnOperatorEnum.eq,
value="testuser")
+ ]
+ results, total = UserDAO.list(column_operators=column_operators)
+ assert total == 1
+ assert results[0].username == "testuser"
+
+
+def test_base_dao_list_with_non_matching_column_operator(
+ user_with_data: Session,
+) -> None:
+ """Test BaseDAO.list with non-matching column operator."""
+ column_operators = [
+ ColumnOperator(
+ col="username", opr=ColumnOperatorEnum.eq, value="nonexistentuser"
+ )
+ ]
+ results, total = UserDAO.list(column_operators=column_operators)
+ assert total == 0
+ assert results == []
+
+
+def test_base_dao_count_returns_value(user_with_data: Session) -> None:
+ """Test that BaseDAO.count returns correct count."""
+ count = UserDAO.count()
+ assert isinstance(count, int)
+ assert count >= 1 # At least the fixture user
+
+
+def test_base_dao_count_with_column_operators(user_with_data: Session) -> None:
+ """Test BaseDAO.count with column operators."""
+ # Count with matching operator
+ column_operators = [
+ ColumnOperator(col="username", opr=ColumnOperatorEnum.eq,
value="testuser")
+ ]
+ count = UserDAO.count(column_operators=column_operators)
+ assert count == 1
+
+ # Count with non-matching operator
+ column_operators = [
+ ColumnOperator(
+ col="username", opr=ColumnOperatorEnum.eq, value="nonexistentuser"
+ )
+ ]
+ count = UserDAO.count(column_operators=column_operators)
+ assert count == 0
+
+
+def test_base_dao_list_ordering(user_with_data: Session) -> None:
+ """Test BaseDAO.list with ordering."""
+ # Create additional users with predictable names
+ users = []
+ for i in range(3):
+ user = User(
+ # Let database auto-generate IDs to avoid conflicts
+ username=f"order_test_{i}",
+ first_name=f"Order{i}",
+ last_name="Test",
+ email=f"order{i}@example.com",
+ active=True,
+ )
+ users.append(user)
+ user_with_data.add(user)
+ user_with_data.commit()
+
+ # Test ascending order
+ results_asc, _ = UserDAO.list(
+ order_column="username", order_direction="asc", page_size=100
+ )
+ usernames_asc = [r.username for r in results_asc]
+ # Check that our test users are in order
+ assert usernames_asc.index("order_test_0") <
usernames_asc.index("order_test_1")
+ assert usernames_asc.index("order_test_1") <
usernames_asc.index("order_test_2")
+
+ # Test descending order
+ results_desc, _ = UserDAO.list(
+ order_column="username", order_direction="desc", page_size=100
+ )
+ usernames_desc = [r.username for r in results_desc]
+ # Check that our test users are in reverse order
+ assert usernames_desc.index("order_test_2") <
usernames_desc.index("order_test_1")
+ assert usernames_desc.index("order_test_1") <
usernames_desc.index("order_test_0")
+
+ for user in users:
+ user_with_data.delete(user)
+ user_with_data.commit()
+
+
+def test_base_dao_list_paging(user_with_data: Session) -> None:
+ """Test BaseDAO.list with paging."""
+ # Create additional users for paging
+ users = []
+ for i in range(10):
+ user = User(
+ id=300 + i,
Review Comment:
<div>
<div id="suggestion">
<div id="issue"><b>Primary key violation risk</b></div>
<div id="fix">
The test is manually assigning IDs (300+i) to User objects, which conflicts
with the database's auto-increment primary key mechanism. This will cause
primary key violations when the database already contains users in this ID
range. Remove the manual ID assignment and let the database auto-generate
unique IDs.
</div>
<details>
<summary>
<b>Code suggestion</b>
</summary>
<blockquote>Check the AI-generated fix before applying</blockquote>
<div id="code">
```suggestion
user = User(
```
</div>
</details>
</div>
<small><i>Code Review Run <a
href=https://github.com/apache/superset/pull/35018#issuecomment-3299702064>#b918c4</a></i></small>
</div>
---
Should Bito avoid suggestions like this for future reviews? (<a
href=https://alpha.bito.ai/home/ai-agents/review-rules>Manage Rules</a>)
- [ ] Yes, avoid them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
