add unit tests

tests/integration_tests/base_tests.py

@@ -46,6 +46,7 @@
 from superset.views.base_api import BaseSupersetModelRestApi
 
 FAKE_DB_NAME = "fake_db_100"
+DEFAULT_PASSWORD = "general"
 test_client = app.test_client()
 
 
@@ -132,7 +133,7 @@ def create_user_with_roles(
  username,
  f"{username}@superset.com",
  security_manager.find_role("Gamma"), # it needs a role
- password="general",
+ password=DEFAULT_PASSWORD,
  )
  db.session.commit()
  user_to_create = security_manager.find_user(username)
@@ -164,14 +165,17 @@ def temporary_user(
  # user is automatically logged out and deleted after the test
  """
  username = username or f"temp_user_{shortid()}"
- temp_user = ab_models.User(username=username, email=f"{username}@temp.com")
+ temp_user = ab_models.User(
+ username=username, email=f"{username}@temp.com", active=True
+ )
  if clone_user:
  temp_user.roles = clone_user.roles
  temp_user.first_name = clone_user.first_name
  temp_user.last_name = clone_user.last_name
+ temp_user.password = clone_user.password
  else:
- temp_user.first_name = "temp"
- temp_user.last_name = "temp"
+ temp_user.first_name = temp_user.last_name = username
+ password = DEFAULT_PASSWORD
 
  if clone_user:
  temp_user.roles = clone_user.roles
@@ -196,11 +200,13 @@ def temporary_user(
  # Add the temp user to the session and commit to apply changes for the test
  db.session.add(temp_user)
  db.session.commit()
- previous_g_user = g.user
+ previous_g_user = g.user if hasattr(g, "user") else None
  try:
  if login:
- self.login(username=temp_user.username)
- g.user = temp_user
+ resp = self.login(username=temp_user.username)
+ print(resp)
+ else:
+ g.user = temp_user
  yield temp_user
  finally:
  # Revert changes after the test
@@ -265,7 +271,7 @@ def get_or_create(self, cls, criteria, **kwargs):
  db.session.commit()
  return obj
 
- def login(self, username, password="general"):
+ def login(self, username, password=DEFAULT_PASSWORD):
  return login(self.client, username, password)
 
  def get_slice(self, slice_name: str) -> Slice:

tests/integration_tests/charts/api_tests.py

@@ -1293,6 +1293,40 @@ def test_user_gets_none_filtered_energy_slices(self):
  data = json.loads(rv.data.decode("utf-8"))
  self.assertEqual(data["count"], 0)
 
+ @pytest.mark.usefixtures("load_energy_charts")
+ def test_user_gets_all_charts(self):
+ # test filtering on datasource_name
+ gamma_user = security_manager.find_user(username="gamma")
+
+ def count_charts():
+ uri = "api/v1/chart/"
+ rv = self.client.get(uri, "get_list")
+ print(rv.data)
+ self.assertEqual(rv.status_code, 200)
+ data = rv.get_json()
+ return data["count"]
+
+ with self.temporary_user(gamma_user, login=True) as user:
+ self.assertEqual(count_charts(), 0)
+
+ all_db_pvm = ("all_database_access", "all_database_access")
+ with self.temporary_user(
+ gamma_user, extra_pvms=[all_db_pvm], login=True
+ ) as user:
+ self.login(username=user.username)
+ assert count_charts() > 0
+
+ all_db_pvm = ("all_datasource_access", "all_datasource_access")
+ with self.temporary_user(
+ gamma_user, extra_pvms=[all_db_pvm], login=True
+ ) as user:
+ self.login(username=user.username)
+ assert count_charts() > 0
+
+ # Back to normal
+ with self.temporary_user(gamma_user, login=True) as user_gamma:
+ self.assertEqual(count_charts(), 0)
+
  @pytest.mark.usefixtures("create_charts")
  def test_get_charts_favorite_filter(self):
  """

tests/integration_tests/datasets/api_tests.py

@@ -185,6 +185,40 @@ def create_dataset_import(self) -> BytesIO:
  buf.seek(0)
  return buf
 
+ @pytest.mark.usefixtures("load_energy_table_with_slice")
+ def test_user_gets_all_datasets(self):
+ # test filtering on datasource_name
+ gamma_user = security_manager.find_user(username="gamma")
+
+ def count_datasets():
+ uri = "api/v1/chart/"
+ rv = self.client.get(uri, "get_list")
+ print(rv.data)
+ self.assertEqual(rv.status_code, 200)
+ data = rv.get_json()
+ return data["count"]
+
+ with self.temporary_user(gamma_user, login=True) as user:
+ assert count_datasets() == 0
+
+ all_db_pvm = ("all_database_access", "all_database_access")
+ with self.temporary_user(
+ gamma_user, extra_pvms=[all_db_pvm], login=True
+ ) as user:
+ self.login(username=user.username)
+ assert count_datasets() > 0
+
+ all_db_pvm = ("all_datasource_access", "all_datasource_access")
+ with self.temporary_user(
+ gamma_user, extra_pvms=[all_db_pvm], login=True
+ ) as user:
+ self.login(username=user.username)
+ assert count_datasets() > 0
+
+ # Back to normal
+ with self.temporary_user(gamma_user, login=True) as user_gamma:
+ assert count_datasets() == 0
+
  def test_get_dataset_list(self):
  """
  Dataset API: Test get dataset list

tests/integration_tests/test_app.py

@@ -39,3 +39,4 @@ def login(
  data=dict(username=username, password=password),
  ).get_data(as_text=True)
  assert "User confirmation needed" not in resp
+ return resp