mirror of
https://github.com/apache/superset.git
synced 2024-09-18 19:49:37 -04:00
00c99c91ec
* feat: new models for SIP-68 * feat: new dataset models DB migration * Add shadow write (WIP) * Physical dataset shadow write (WIP) * Virtual dataset shadow write (WIP) * Update migration to populate models * Cascade delete columns * after_delete * Update hook * Add metric test * Do not rename tables * Small fixes * Fix migration * Fix tests * Fix more tests * Even more tests * Addressing comments (WIP) * Fix migration * Rebase and update head * Fix unit tests * Add external management cols * Small fixes
286 lines
8.7 KiB
Python
286 lines
8.7 KiB
Python
# 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.
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict, Generator, List, TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from superset import db, security_manager as sm
|
|
from superset.dashboards.filter_sets.consts import (
|
|
DESCRIPTION_FIELD,
|
|
JSON_METADATA_FIELD,
|
|
NAME_FIELD,
|
|
OWNER_ID_FIELD,
|
|
OWNER_TYPE_FIELD,
|
|
USER_OWNER_TYPE,
|
|
)
|
|
from superset.models.dashboard import Dashboard
|
|
from superset.models.filter_set import FilterSet
|
|
from tests.integration_tests.dashboards.filter_sets.consts import (
|
|
ADMIN_USERNAME_FOR_TEST,
|
|
DASHBOARD_OWNER_USERNAME,
|
|
FILTER_SET_OWNER_USERNAME,
|
|
REGULAR_USER,
|
|
)
|
|
from tests.integration_tests.dashboards.superset_factory_util import (
|
|
create_dashboard,
|
|
create_database,
|
|
create_datasource_table,
|
|
create_slice,
|
|
)
|
|
from tests.integration_tests.test_app import app
|
|
|
|
if TYPE_CHECKING:
|
|
from flask.ctx import AppContext
|
|
from flask.testing import FlaskClient
|
|
from flask_appbuilder.security.manager import BaseSecurityManager
|
|
from flask_appbuilder.security.sqla.models import (
|
|
PermissionView,
|
|
Role,
|
|
User,
|
|
ViewMenu,
|
|
)
|
|
from sqlalchemy.orm import Session
|
|
|
|
from superset.connectors.sqla.models import SqlaTable
|
|
from superset.models.core import Database
|
|
from superset.models.slice import Slice
|
|
|
|
|
|
security_manager: BaseSecurityManager = sm
|
|
|
|
|
|
@pytest.fixture(autouse=True, scope="module")
|
|
def test_users() -> Generator[Dict[str, int], None, None]:
|
|
usernames = [
|
|
ADMIN_USERNAME_FOR_TEST,
|
|
DASHBOARD_OWNER_USERNAME,
|
|
FILTER_SET_OWNER_USERNAME,
|
|
REGULAR_USER,
|
|
]
|
|
with app.app_context():
|
|
filter_set_role = build_filter_set_role()
|
|
admin_role: Role = security_manager.find_role("Admin")
|
|
usernames_to_ids = create_test_users(admin_role, filter_set_role, usernames)
|
|
yield usernames_to_ids
|
|
delete_users(usernames_to_ids)
|
|
|
|
|
|
def delete_users(usernames_to_ids: Dict[str, int]) -> None:
|
|
for username in usernames_to_ids.keys():
|
|
db.session.delete(security_manager.find_user(username))
|
|
db.session.commit()
|
|
|
|
|
|
def create_test_users(
|
|
admin_role: Role, filter_set_role: Role, usernames: List[str]
|
|
) -> Dict[str, int]:
|
|
users: List[User] = []
|
|
for username in usernames:
|
|
user = build_user(username, filter_set_role, admin_role)
|
|
users.append(user)
|
|
return {user.username: user.id for user in users}
|
|
|
|
|
|
def build_user(username: str, filter_set_role: Role, admin_role: Role) -> User:
|
|
roles_to_add = (
|
|
[admin_role] if username == ADMIN_USERNAME_FOR_TEST else [filter_set_role]
|
|
)
|
|
user: User = security_manager.add_user(
|
|
username, "test", "test", username, roles_to_add, password="general"
|
|
)
|
|
if not user:
|
|
user = security_manager.find_user(username)
|
|
if user is None:
|
|
raise Exception("Failed to build the user {}".format(username))
|
|
return user
|
|
|
|
|
|
def build_filter_set_role() -> Role:
|
|
filter_set_role: Role = security_manager.add_role("filter_set_role")
|
|
filterset_view_name: ViewMenu = security_manager.find_view_menu("FilterSets")
|
|
all_datasource_view_name: ViewMenu = security_manager.find_view_menu(
|
|
"all_datasource_access"
|
|
)
|
|
pvms: List[PermissionView] = security_manager.find_permissions_view_menu(
|
|
filterset_view_name
|
|
) + security_manager.find_permissions_view_menu(all_datasource_view_name)
|
|
for pvm in pvms:
|
|
security_manager.add_permission_role(filter_set_role, pvm)
|
|
return filter_set_role
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> Generator[FlaskClient[Any], None, None]:
|
|
with app.test_client() as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def dashboard(app_context) -> Generator[Dashboard, None, None]:
|
|
dashboard_owner_user = security_manager.find_user(DASHBOARD_OWNER_USERNAME)
|
|
database = create_database("test_database_filter_sets")
|
|
datasource = create_datasource_table(
|
|
name="test_datasource", database=database, owners=[dashboard_owner_user]
|
|
)
|
|
slice_ = create_slice(
|
|
datasource=datasource, name="test_slice", owners=[dashboard_owner_user]
|
|
)
|
|
dashboard = create_dashboard(
|
|
dashboard_title="test_dashboard",
|
|
published=True,
|
|
slices=[slice_],
|
|
owners=[dashboard_owner_user],
|
|
)
|
|
db.session.add(dashboard)
|
|
db.session.commit()
|
|
|
|
yield dashboard
|
|
|
|
db.session.delete(dashboard)
|
|
db.session.delete(slice_)
|
|
db.session.delete(datasource)
|
|
db.session.delete(database)
|
|
db.session.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
def dashboard_id(dashboard: Dashboard) -> Generator[int, None, None]:
|
|
yield dashboard.id
|
|
|
|
|
|
@pytest.fixture
|
|
def filtersets(
|
|
dashboard_id: int, test_users: Dict[str, int], dumped_valid_json_metadata: str
|
|
) -> Generator[Dict[str, List[FilterSet]], None, None]:
|
|
first_filter_set = FilterSet(
|
|
name="filter_set_1_of_" + str(dashboard_id),
|
|
dashboard_id=dashboard_id,
|
|
json_metadata=dumped_valid_json_metadata,
|
|
owner_id=dashboard_id,
|
|
owner_type="Dashboard",
|
|
)
|
|
second_filter_set = FilterSet(
|
|
name="filter_set_2_of_" + str(dashboard_id),
|
|
json_metadata=dumped_valid_json_metadata,
|
|
dashboard_id=dashboard_id,
|
|
owner_id=dashboard_id,
|
|
owner_type="Dashboard",
|
|
)
|
|
third_filter_set = FilterSet(
|
|
name="filter_set_3_of_" + str(dashboard_id),
|
|
json_metadata=dumped_valid_json_metadata,
|
|
dashboard_id=dashboard_id,
|
|
owner_id=test_users[FILTER_SET_OWNER_USERNAME],
|
|
owner_type="User",
|
|
)
|
|
fourth_filter_set = FilterSet(
|
|
name="filter_set_4_of_" + str(dashboard_id),
|
|
json_metadata=dumped_valid_json_metadata,
|
|
dashboard_id=dashboard_id,
|
|
owner_id=test_users[FILTER_SET_OWNER_USERNAME],
|
|
owner_type="User",
|
|
)
|
|
db.session.add(first_filter_set)
|
|
db.session.add(second_filter_set)
|
|
db.session.add(third_filter_set)
|
|
db.session.add(fourth_filter_set)
|
|
db.session.commit()
|
|
|
|
yield {
|
|
"Dashboard": [first_filter_set, second_filter_set],
|
|
FILTER_SET_OWNER_USERNAME: [third_filter_set, fourth_filter_set],
|
|
}
|
|
|
|
db.session.delete(first_filter_set)
|
|
db.session.delete(second_filter_set)
|
|
db.session.delete(third_filter_set)
|
|
db.session.delete(fourth_filter_set)
|
|
db.session.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
def filterset_id(filtersets: Dict[str, List[FilterSet]]) -> int:
|
|
return filtersets["Dashboard"][0].id
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_json_metadata() -> Dict[str, Any]:
|
|
return {"nativeFilters": {}}
|
|
|
|
|
|
@pytest.fixture
|
|
def dumped_valid_json_metadata(valid_json_metadata: Dict[str, Any]) -> str:
|
|
return json.dumps(valid_json_metadata)
|
|
|
|
|
|
@pytest.fixture
|
|
def exists_user_id() -> int:
|
|
return 1
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_filter_set_data_for_create(
|
|
dashboard_id: int, dumped_valid_json_metadata: str, exists_user_id: int
|
|
) -> Dict[str, Any]:
|
|
name = "test_filter_set_of_dashboard_" + str(dashboard_id)
|
|
return {
|
|
NAME_FIELD: name,
|
|
DESCRIPTION_FIELD: "description of " + name,
|
|
JSON_METADATA_FIELD: dumped_valid_json_metadata,
|
|
OWNER_TYPE_FIELD: USER_OWNER_TYPE,
|
|
OWNER_ID_FIELD: exists_user_id,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def valid_filter_set_data_for_update(
|
|
dashboard_id: int, dumped_valid_json_metadata: str, exists_user_id: int
|
|
) -> Dict[str, Any]:
|
|
name = "name_changed_test_filter_set_of_dashboard_" + str(dashboard_id)
|
|
return {
|
|
NAME_FIELD: name,
|
|
DESCRIPTION_FIELD: "changed description of " + name,
|
|
JSON_METADATA_FIELD: dumped_valid_json_metadata,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def not_exists_dashboard_id(dashboard_id: int) -> Generator[int, None, None]:
|
|
yield dashboard_id + 1
|
|
|
|
|
|
@pytest.fixture
|
|
def not_exists_user_id() -> int:
|
|
return 99999
|
|
|
|
|
|
@pytest.fixture()
|
|
def dashboard_based_filter_set_dict(
|
|
filtersets: Dict[str, List[FilterSet]]
|
|
) -> Dict[str, Any]:
|
|
return filtersets["Dashboard"][0].to_dict()
|
|
|
|
|
|
@pytest.fixture()
|
|
def user_based_filter_set_dict(
|
|
filtersets: Dict[str, List[FilterSet]]
|
|
) -> Dict[str, Any]:
|
|
return filtersets[FILTER_SET_OWNER_USERNAME][0].to_dict()
|