Merge branch 'validate-params' into 'enterprise'

fix(nav): validate query parameters

See merge request dkinternal/testgen/dataops-testgen!286

Author: ci bot

testgen/common/models/profiling_run.py

@@ -131,8 +131,13 @@ def select_minimal_where(
     @classmethod
     @st.cache_data(show_spinner=False)
     def select_summary(
-        cls, project_code: str, table_group_id: str | None = None, profiling_run_ids: list[str] | None = None
+        cls, project_code: str, table_group_id: str | UUID | None = None, profiling_run_ids: list[str] | None = None
     ) -> Iterable[ProfilingRunSummary]:
+        if (table_group_id and not is_uuid4(table_group_id)) or (
+            profiling_run_ids and not all(is_uuid4(run_id) for run_id in profiling_run_ids)
+        ):
+            return []
+
         query = f"""
         WITH profile_anomalies AS (
             SELECT profile_anomaly_results.profile_run_id,

testgen/common/models/scores.py

@@ -50,6 +50,7 @@
     "transform_level",
     "data_product",
 ]
+ScoreTypes = Literal["score", "cde_score"]
 
 
 class ScoreCategory(enum.Enum):
@@ -344,7 +345,7 @@ def get_score_card_issues(
 
         dq_dimension_filter = ""
         if group_by == "dq_dimension":
-            dq_dimension_filter = f" AND dq_dimension = '{value_}'"
+            dq_dimension_filter = " AND dq_dimension = :value"
 
         query = (
             read_template_sql_file(query_template_file, sub_directory="score_cards")
@@ -594,7 +595,7 @@ def filter(
         *,
         definition_id: str,
         category: Categories,
-        score_type: Literal["score", "cde_score"],
+        score_type: ScoreTypes,
     ) -> Iterable[Self]:
         items = []
         db_session = get_current_session()

testgen/common/models/test_run.py

@@ -125,6 +125,13 @@ def select_summary(
         test_suite_id: str | None = None,
         test_run_ids: list[str] | None = None,
     ) -> Iterable[TestRunSummary]:
+        if (
+            (table_group_id and not is_uuid4(table_group_id))
+            or (test_suite_id and not is_uuid4(test_suite_id))
+            or (test_run_ids and not all(is_uuid4(run_id) for run_id in test_run_ids))
+        ):
+            return []
+
         query = f"""
         WITH run_results AS (
             SELECT test_run_id,

testgen/common/models/test_suite.py

@@ -11,6 +11,7 @@
 from testgen.common.models import get_current_session
 from testgen.common.models.custom_types import NullIfEmptyString, YNString
 from testgen.common.models.entity import ENTITY_HASH_FUNCS, Entity, EntityMinimal
+from testgen.utils import is_uuid4
 
 
 @dataclass
@@ -83,7 +84,10 @@ def select_minimal_where(
 
     @classmethod
     @st.cache_data(show_spinner=False)
-    def select_summary(cls, project_code: str, table_group_id: str | None = None) -> Iterable[TestSuiteSummary]:
+    def select_summary(cls, project_code: str, table_group_id: str | UUID | None = None) -> Iterable[TestSuiteSummary]:
+        if table_group_id and not is_uuid4(table_group_id):
+            return []
+
         query = f"""
         WITH last_run AS (
             SELECT test_runs.test_suite_id,

testgen/ui/navigation/router.py

@@ -7,6 +7,7 @@
 
 import testgen.ui.navigation.page
 from testgen.common.mixpanel_service import MixpanelService
+from testgen.common.models.project import Project
 from testgen.ui.session import session
 from testgen.utils.singleton import Singleton
 
@@ -114,8 +115,8 @@ def navigate(self, /, to: str, with_args: dict = {}) -> None:  # noqa: B006
     def navigate_with_warning(self, warning: str, to: str, with_args: dict = {}) -> None:  # noqa: B006
         st.warning(warning)
         time.sleep(3)
-        self.navigate(to, with_args)
-
+        session.sidebar_project = session.sidebar_project or Project.select_where()[0].project_code
+        self.navigate(to, {"project_code": session.sidebar_project, **with_args})
 
     def set_query_params(self, with_args: dict) -> None:
         params = st.query_params

testgen/ui/views/data_catalog.py

@@ -426,7 +426,7 @@ def get_table_group_columns(table_group_id: str) -> list[dict]:
 
 
 def get_selected_item(selected: str, table_group_id: str) -> dict | None:
-    if not selected or not is_uuid4(table_group_id):
+    if not selected or "_" not in selected or not is_uuid4(table_group_id):
         return None
 
     item_type, item_id = selected.split("_", 2)

testgen/ui/views/profiling_runs.py

@@ -59,6 +59,7 @@ def render(self, project_code: str, table_group_id: str | None = None, **_kwargs
         with group_filter_column:
             table_groups = TableGroup.select_minimal_where(TableGroup.project_code == project_code)
             table_groups_df = to_dataframe(table_groups, TableGroupMinimal.columns())
+            table_groups_df["id"] = table_groups_df["id"].apply(lambda x: str(x))
             table_group_id = testgen.select(
                 options=table_groups_df,
                 value_column="id",

testgen/ui/views/score_details.py

@@ -1,4 +1,5 @@
 import logging
+import typing
 from io import BytesIO
 from typing import ClassVar
 
@@ -8,7 +9,14 @@
 from testgen.commands.run_refresh_score_cards_results import run_recalculate_score_card
 from testgen.common.mixpanel_service import MixpanelService
 from testgen.common.models import with_database_session
-from testgen.common.models.scores import ScoreCategory, ScoreDefinition, ScoreDefinitionBreakdownItem, SelectedIssue
+from testgen.common.models.scores import (
+    Categories,
+    ScoreCategory,
+    ScoreDefinition,
+    ScoreDefinitionBreakdownItem,
+    ScoreTypes,
+    SelectedIssue,
+)
 from testgen.ui.components import widgets as testgen
 from testgen.ui.components.widgets.download_dialog import FILE_DATA_TYPE, download_dialog, zip_multi_file_data
 from testgen.ui.navigation.page import Page
@@ -60,6 +68,9 @@ def render(
             ],
         )
 
+        if category not in typing.get_args(Categories):
+            category = None
+
         if not category and score_definition.category:
             category = score_definition.category.value
 
@@ -72,6 +83,8 @@ def render(
         with st.spinner(text="Loading data :gray[:small[(This might take a few minutes)]] ..."):
             user_can_edit = user_session_service.user_can_edit()
             score_card = format_score_card(score_definition.as_cached_score_card())
+            if score_type not in typing.get_args(ScoreTypes):
+                score_type = None
             if not score_type:
                 score_type = "cde_score" if score_card["cde_score"] and not score_card["score"] else "score"
             if not drilldown:

testgen/ui/views/score_explorer.py

@@ -105,7 +105,7 @@ def render(
                 score_definition.name = name
                 score_definition.total_score = total_score and total_score.lower() == "true"
                 score_definition.cde_score = cde_score and cde_score.lower() == "true"
-                score_definition.category = ScoreCategory(category) if category else None
+                score_definition.category = ScoreCategory(category) if category in [cat.value for cat in ScoreCategory] else None
 
                 if filters:
                     applied_filters: list[dict] = try_json(filters, default=[])

testgen/ui/views/table_groups.py

@@ -30,6 +30,7 @@ class TableGroupsPage(Page):
     can_activate: typing.ClassVar = [
         lambda: session.authentication_status,
         lambda: not user_session_service.user_has_catalog_role(),
+        lambda: "project_code" in st.query_params,
     ]
     menu_item = MenuItem(
         icon="table_view",
@@ -44,6 +45,9 @@ def render(self, project_code: str, connection_id: str | None = None, **_kwargs)
 
         user_can_edit = user_session_service.user_can_edit()
         project_summary = Project.get_summary(project_code)
+        if connection_id and not connection_id.isdigit():
+            connection_id = None
+
         if connection_id:
             table_groups = TableGroup.select_minimal_where(
                 TableGroup.project_code == project_code,