fix UT

superset/views/datasource/schemas.py

@@ -36,10 +36,10 @@ class ExternalMetadataParams(TypedDict):
 
 
 class ExternalMetadataSchema(Schema):
-    datasource_type = fields.Str()
-    database_name = fields.Str()
+    datasource_type = fields.Str(required=True)
+    database_name = fields.Str(required=True)
     schema_name = fields.Str(allow_none=True)
-    table_name = fields.Str()
+    table_name = fields.Str(required=True)
 
     # pylint: disable=no-self-use,unused-argument
     @post_load
@@ -49,6 +49,6 @@ def normalize(
         return ExternalMetadataParams(
             datasource_type=data["datasource_type"],
             database_name=data["database_name"],
-            schema_name=data["schema_name"] or "",
+            schema_name=data.get("schema_name", ""),
             table_name=data["table_name"],
         )

superset/views/datasource/views.py

@@ -24,11 +24,16 @@
 from flask_appbuilder.security.decorators import has_access_api
 from flask_babel import _
 from marshmallow import ValidationError
+from sqlalchemy.exc import NoSuchTableError
+from sqlalchemy.orm.exc import NoResultFound
 
 from superset import app, db, event_logger
 from superset.connectors.connector_registry import ConnectorRegistry
 from superset.connectors.sqla.utils import get_physical_table_metadata
-from superset.datasets.commands.exceptions import DatasetForbiddenError
+from superset.datasets.commands.exceptions import (
+    DatasetForbiddenError,
+    DatasetNotFoundError,
+)
 from superset.exceptions import SupersetException, SupersetSecurityException
 from superset.models.core import Database
 from superset.typing import FlaskResponse
@@ -170,6 +175,6 @@ def external_metadata_by_name(self, **kwargs: Any) -> FlaskResponse:
                     table_name=params["table_name"],
                     schema_name=params["schema_name"],
                 )
-        except SupersetException as ex:
-            return json_error_response(str(ex), status=400)
+        except (NoResultFound, NoSuchTableError):
+            raise DatasetNotFoundError
         return self.json_response(external_metadata)

tests/integration_tests/datasource_tests.py

@@ -19,6 +19,7 @@
 from contextlib import contextmanager
 from unittest import mock
 
+import prison
 import pytest
 
 from superset import app, ConnectorRegistry, db
@@ -90,11 +91,15 @@ def test_external_metadata_for_virtual_table(self):
     def test_external_metadata_by_name_for_physical_table(self):
         self.login(username="admin")
         tbl = self.get_table(name="birth_names")
-        # empty schema need to be represented by undefined
-        url = (
-            f"/datasource/external_metadata_by_name/table/"
-            f"{tbl.database.database_name}/undefined/{tbl.table_name}/"
+        params = prison.dumps(
+            {
+                "datasource_type": "table",
+                "database_name": tbl.database.database_name,
+                "schema_name": tbl.schema,
+                "table_name": tbl.table_name,
+            }
         )
+        url = f"/datasource/external_metadata_by_name/?q={params}"
         resp = self.get_json_resp(url)
         col_names = {o.get("name") for o in resp}
         self.assertEqual(
@@ -112,33 +117,69 @@ def test_external_metadata_by_name_for_virtual_table(self):
         session.add(table)
         session.commit()
 
-        table = self.get_table(name="dummy_sql_table")
-        # empty schema need to be represented by undefined
-        url = (
-            f"/datasource/external_metadata_by_name/table/"
-            f"{table.database.database_name}/undefined/{table.table_name}/"
+        tbl = self.get_table(name="dummy_sql_table")
+        params = prison.dumps(
+            {
+                "datasource_type": "table",
+                "database_name": tbl.database.database_name,
+                "schema_name": tbl.schema,
+                "table_name": tbl.table_name,
+            }
         )
+        url = f"/datasource/external_metadata_by_name/?q={params}"
         resp = self.get_json_resp(url)
         assert {o.get("name") for o in resp} == {"intcol", "strcol"}
-        session.delete(table)
+        session.delete(tbl)
         session.commit()
 
     def test_external_metadata_by_name_from_sqla_inspector(self):
         self.login(username="admin")
         example_database = get_example_database()
         with create_test_table_context(example_database):
-            url = (
-                f"/datasource/external_metadata_by_name/table/"
-                f"{example_database.database_name}/undefined/test_table/"
+            params = prison.dumps(
+                {
+                    "datasource_type": "table",
+                    "database_name": example_database.database_name,
+                    "table_name": "test_table",
+                }
             )
+            url = f"/datasource/external_metadata_by_name/?q={params}"
             resp = self.get_json_resp(url)
             col_names = {o.get("name") for o in resp}
             self.assertEqual(col_names, {"first", "second"})
 
-        url = (
-            f"/datasource/external_metadata_by_name/table/" f"foobar/undefined/foobar/"
+        # No databases found
+        params = prison.dumps(
+            {"datasource_type": "table", "database_name": "foo", "table_name": "bar",}
+        )
+        url = f"/datasource/external_metadata_by_name/?q={params}"
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, DatasetNotFoundError.status)
+        self.assertEqual(
+            json.loads(resp.data.decode("utf-8")).get("error"),
+            DatasetNotFoundError.message,
+        )
+
+        # No table found
+        params = prison.dumps(
+            {
+                "datasource_type": "table",
+                "database_name": example_database.database_name,
+                "table_name": "fooooooooobarrrrrr",
+            }
+        )
+        url = f"/datasource/external_metadata_by_name/?q={params}"
+        resp = self.client.get(url)
+        self.assertEqual(resp.status_code, DatasetNotFoundError.status)
+        self.assertEqual(
+            json.loads(resp.data.decode("utf-8")).get("error"),
+            DatasetNotFoundError.message,
         )
-        resp = self.get_json_resp(url, raise_on_error=False)
+
+        # invalid query params
+        params = prison.dumps({"datasource_type": "table",})
+        url = f"/datasource/external_metadata_by_name/?q={params}"
+        resp = self.get_json_resp(url)
         self.assertIn("error", resp)
 
     def test_external_metadata_for_virtual_table_template_params(self):