[DOP-29475] Add table name validation

Author: dolfinus

syncmaster/schemas/v1/transfers/db.py

@@ -2,7 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-from pydantic import BaseModel
+import re
+from typing import ClassVar
+
+from pydantic import BaseModel, Field, field_validator
 
 from syncmaster.schemas.v1.connection_types import (
     CLICKHOUSE_TYPE,
@@ -16,7 +19,16 @@
 
 
 class DBTransfer(BaseModel):
-    table_name: str
+    TABLE_NAME_PATTERN: ClassVar[str] = r"^[\w\d]+\.[\w\d]+$"
+    table_name: str = Field(description="Table name", json_schema_extra={"pattern": TABLE_NAME_PATTERN})
+
+    # make error message more user friendly
+    @field_validator("table_name", mode="before")
+    @classmethod
+    def _table_name_is_qualified(cls, value):
+        if not re.match(cls.TABLE_NAME_PATTERN, value):
+            raise ValueError("Table name should be in format myschema.mytable")
+        return value
 
 
 class HiveTransferSourceOrTarget(DBTransfer):
@@ -46,6 +58,17 @@ class MySQLTransferSourceOrTarget(DBTransfer):
 class IcebergTransferSourceOrTarget(DBTransfer):
     type: ICEBERG_TYPE
 
+    TABLE_NAME_PATTERN: ClassVar[str] = r"^[\w\d]+(\.[\w\d]+)+$"
+    table_name: str = Field(description="Table name", json_schema_extra={"pattern": TABLE_NAME_PATTERN})
+
+    # make error message more user friendly
+    @field_validator("table_name", mode="before")
+    @classmethod
+    def _table_name_is_qualified(cls, value):
+        if not re.match(cls.TABLE_NAME_PATTERN, value):
+            raise ValueError("Table name should be in format myschema.mytable")
+        return value
+
 
 DBTransferSourceOrTarget = (
     ClickhouseTransferSourceOrTarget

syncmaster/schemas/v1/transfers/file/base.py

@@ -26,7 +26,10 @@
 # At the moment the CreateTransferSourceParams and CreateTransferTargetParams
 # classes are identical but may change in the future
 class FileTransferSource(BaseModel):
-    directory_path: str
+    directory_path: str = Field(
+        description="Absolute path to directory",
+        json_schema_extra={"pattern": r"^/[\w\d ]+(/[\w\d ]+)*$"},
+    )
     file_format: SOURCE_FILE_FORMAT = Field(discriminator="type")
     options: dict[str, Any] = Field(default_factory=dict)
 
@@ -41,18 +44,22 @@ def _directory_path_is_valid_path(cls, value):
 
 
 class FileTransferTarget(BaseModel):
-    directory_path: str
+    FILE_NAME_PATTERN: ClassVar[str] = r"^[\w.{}-]+$"
+
+    directory_path: str = Field(
+        description="Absolute path to directory",
+        json_schema_extra={"pattern": r"^/[\w\d ]+(/[\w\d ]+)*$"},
+    )
     file_format: TARGET_FILE_FORMAT = Field(discriminator="type")
     file_name_template: str = Field(
         default="{run_created_at}-{index}.{extension}",
         description="Template for file naming with required placeholders 'index' and 'extension'",
+        json_schema_extra={"pattern": FILE_NAME_PATTERN},
     )
     options: dict[str, Any] = Field(default_factory=dict)
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    FILE_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^[\w.{}-]+$")
-
     @field_validator("directory_path", mode="before")
     @classmethod
     def _directory_path_is_valid_path(cls, value):
@@ -63,7 +70,8 @@ def _directory_path_is_valid_path(cls, value):
     @field_validator("file_name_template")
     @classmethod
     def _validate_file_name_template(cls, value: str) -> str:  # noqa: WPS238
-        if not cls.FILE_NAME_PATTERN.match(value):
+        # make error message more user friendly
+        if not re.match(cls.FILE_NAME_PATTERN, value):
             raise ValueError("Template contains invalid characters. Allowed: letters, numbers, '.', '_', '-', '{', '}'")
 
         required_keys = {"index", "extension"}

tests/test_unit/test_transfers/test_db_transfers/test_create_transfer.py

@@ -0,0 +1,77 @@
+import pytest
+from httpx import AsyncClient
+
+from syncmaster.db.models import Queue
+from tests.mocks import MockConnection, MockGroup, UserTestRoles
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.server]
+
+
+@pytest.mark.parametrize(
+    "create_connection_data",
+    [
+        {
+            "type": "postgres",
+            "host": "localhost",
+            "port": 5432,
+        },
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize(
+    "target_source_params",
+    [
+        {
+            "type": "postgres",
+            "table_name": "table",
+        },
+    ],
+)
+async def test_cannot_create_db_transfer_with_short_table_name(
+    client: AsyncClient,
+    two_group_connections: tuple[MockConnection, MockConnection],
+    group_queue: Queue,
+    mock_group: MockGroup,
+    target_source_params: dict,
+    create_connection_data: dict,
+):
+    first_connection, second_connection = two_group_connections
+    user = mock_group.get_member_of_role(UserTestRoles.Developer)
+
+    response = await client.post(
+        "v1/transfers",
+        headers={"Authorization": f"Bearer {user.token}"},
+        json={
+            "group_id": mock_group.group.id,
+            "name": "new test transfer",
+            "source_connection_id": first_connection.id,
+            "target_connection_id": second_connection.id,
+            "source_params": target_source_params,
+            "target_params": target_source_params,
+            "queue_id": group_queue.id,
+        },
+    )
+
+    assert response.status_code == 422, response.text
+    assert response.json() == {
+        "error": {
+            "code": "invalid_request",
+            "message": "Invalid request",
+            "details": [
+                {
+                    "context": {},
+                    "input": "table",
+                    "location": ["body", "source_params", "postgres", "table_name"],
+                    "message": "Value error, Table name should be in format myschema.mytable",
+                    "code": "value_error",
+                },
+                {
+                    "context": {},
+                    "input": "table",
+                    "location": ["body", "target_params", "postgres", "table_name"],
+                    "message": "Value error, Table name should be in format myschema.mytable",
+                    "code": "value_error",
+                },
+            ],
+        },
+    }

tests/test_unit/test_transfers/test_file_transfers/test_create_transfer.py

@@ -400,37 +400,6 @@ async def test_developer_plus_can_create_hdfs_transfer(
                 "type": "csv",
             },
         },
-        {
-            "type": "s3",
-            "directory_path": "some/path",
-            "file_format": {
-                "type": "excel",
-                "include_header": True,
-            },
-        },
-        {
-            "type": "s3",
-            "directory_path": "some/path",
-            "file_format": {
-                "type": "xml",
-                "root_tag": "data",
-                "row_tag": "record",
-            },
-        },
-        {
-            "type": "s3",
-            "directory_path": "some/path",
-            "file_format": {
-                "type": "orc",
-            },
-        },
-        {
-            "type": "s3",
-            "directory_path": "some/path",
-            "file_format": {
-                "type": "parquet",
-            },
-        },
     ],
 )
 async def test_cannot_create_file_transfer_with_relative_path(

tests/test_unit/test_transfers/test_file_transfers/test_update_transfer.py

@@ -143,7 +143,7 @@
     ],
     indirect=["create_connection_data"],
 )
-async def test_developer_plus_can_update_s3_transfer(
+async def test_developer_plus_can_update_ftp_transfer(
     client: AsyncClient,
     group_transfer: MockTransfer,
     role_developer_plus: UserTestRoles,