test(translate): add tests for ConnectionService, DbExecutor, orchestrator direct DB dispatch
- 9 new enhancement test files: test_connection_service.py, test_db_executor.py, test_orchestrator_direct_db.py, test_batch_insert.py, test_lang_stats.py, test_response_field_coverage.py, test_retry.py, test_run_service.py, test_sql_insert_service.py - 5 new integration tests: test_superset_sqllab_e2e.py, test_translate_clickhouse.py, test_translate_corrections.py, test_translate_schedules.py, test_translate_status_fk.py - Updated existing tests for insert_method/connection_id fields
This commit is contained in:
@@ -424,4 +424,64 @@ def superset_admin_headers(superset_url, superset_admin_password):
|
||||
# #endregion superset_admin_headers
|
||||
|
||||
|
||||
# ── JWT-based Superset Fixtures ────────────────────────────────────
|
||||
# ADR-0012: после установки psycopg2 + superset_config.py + Docker bridge IP,
|
||||
# JWT-авторизация (/api/v1/security/login) работает в тестовом контейнере.
|
||||
|
||||
|
||||
# #region superset_jwt_headers [C:2] [TYPE Fixture]
|
||||
# @BRIEF Function-scoped — JWT Bearer token headers for Superset REST API.
|
||||
# @POST Returns dict with Authorization: Bearer <access_token>.
|
||||
@pytest.fixture
|
||||
def superset_jwt_headers(superset_url, superset_admin_password):
|
||||
"""Obtain JWT access token from the running Superset container."""
|
||||
resp = requests.post(
|
||||
f"{superset_url}/api/v1/security/login",
|
||||
json={
|
||||
"username": "admin",
|
||||
"password": superset_admin_password,
|
||||
"provider": "db",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
access_token = data["access_token"]
|
||||
assert access_token, "No access_token in JWT response"
|
||||
return {"Authorization": f"Bearer {access_token}"}
|
||||
# #endregion superset_jwt_headers
|
||||
|
||||
|
||||
# #region superset_env [C:2] [TYPE Fixture]
|
||||
# @BRIEF Function-scoped — real Environment pointing to the running Superset container.
|
||||
@pytest.fixture
|
||||
def superset_env(superset_url, superset_admin_password):
|
||||
"""Create a real Environment config pointing to the Superset container."""
|
||||
from src.core.config_models import Environment
|
||||
|
||||
return Environment(
|
||||
id="test_env",
|
||||
name="Test Superset Container",
|
||||
url=superset_url,
|
||||
username="admin",
|
||||
password=superset_admin_password,
|
||||
verify_ssl=False,
|
||||
timeout=30,
|
||||
)
|
||||
# #endregion superset_env
|
||||
|
||||
|
||||
# #region superset_client [C:2] [TYPE Fixture]
|
||||
# @BRIEF Function-scoped — real SupersetClient connected to the container.
|
||||
@pytest.fixture
|
||||
async def superset_client(superset_env):
|
||||
"""Create and authenticate a SupersetClient against the real container."""
|
||||
from src.core.superset_client import SupersetClient
|
||||
|
||||
client = SupersetClient(superset_env)
|
||||
await client.authenticate()
|
||||
return client
|
||||
# #endregion superset_client
|
||||
|
||||
|
||||
# #endregion IntegrationTestConftest
|
||||
|
||||
@@ -1,257 +1,255 @@
|
||||
# #region TestSupersetSqllabApi [C:4] [TYPE Module] [SEMANTICS test,superset,sqllab,integration]
|
||||
# @BRIEF Integration tests for Superset SQL Lab REST API using real Testcontainers Superset.
|
||||
# #region TestSupersetSqllabApiJwt [C:4] [TYPE Module] [SEMANTICS test,superset,sqllab,integration,jwt]
|
||||
# @BRIEF Integration tests for Superset REST API using JWT Bearer auth against real Testcontainers.
|
||||
# @RELATION BINDS_TO -> [SupersetSqlLabExecutor]
|
||||
# @RELATION BINDS_TO -> [SupersetClient]
|
||||
# @RELATION BINDS_TO -> [ConfigManager]
|
||||
#
|
||||
# @TEST_CONTRACT SupersetSqlLabApi ->
|
||||
# @TEST_CONTRACT SupersetSqllabApi ->
|
||||
# {
|
||||
# invariants: [
|
||||
# "GET /api/v1/database/ returns paginated database list after login",
|
||||
# "JWT login returns access_token + refresh_token",
|
||||
# "Bearer token authorizes all /api/v1/ endpoints",
|
||||
# "GET /api/v1/database/ returns paginated database list",
|
||||
# "POST /api/v1/sqllab/execute/ accepts SQL and returns query_id",
|
||||
# "Form-based session auth works for all API v1 endpoints",
|
||||
# "CSRF token acquisition from login page enables API mutation endpoints"
|
||||
# "GET /api/v1/dashboard/ returns paginated list with Bearer token"
|
||||
# ]
|
||||
# }
|
||||
# @TEST_EDGE: missing_csrf_token -> API calls without CSRF return 403
|
||||
# @TEST_EDGE: wrong_database_id -> SQL execution returns 404
|
||||
# @TEST_EDGE: unauthenticated_sqllab -> 401 without session
|
||||
# @TEST_EDGE: wrong_password -> 401 on JWT login
|
||||
# @TEST_EDGE: expired_token -> 401 on protected endpoints
|
||||
# @TEST_EDGE: missing_database_id -> 400 on sqllab execute
|
||||
#
|
||||
# @RATIONALE Smoke tests for the real Superset 4.1.2 SQL Lab API to verify
|
||||
# our integration assumptions match actual behavior. JWT auth is unavailable
|
||||
# in the test container (ADR-0012), so we use form-based session auth instead.
|
||||
# The raw REST API tests complement the smoke tests already in
|
||||
# test_superset_integration.py.
|
||||
# @RATIONALE ADR-0012 resolved the JWT auth issue via psycopg2 + superset_config.py +
|
||||
# Docker bridge IP. Now JWT works end-to-end with the test container.
|
||||
# All API tests use real Bearer tokens against the live Superset instance.
|
||||
#
|
||||
# @REQUIRES Docker daemon running (testcontainers starts superset + postgres)
|
||||
#
|
||||
# @TEST_INVARIANT sqllab_execute_accepts_sql -> VERIFIED_BY: [test_sqllab_execute_endpoint_reachable]
|
||||
# @TEST_INVARIANT form_auth_works_for_api -> VERIFIED_BY: [test_list_databases_after_form_login]
|
||||
import re
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
# #region TestSupersetSqllabRawApi [C:3] [TYPE Class]
|
||||
# @BRIEF Verify Superset SQL Lab REST API endpoints with form-based session auth.
|
||||
class TestSupersetSqllabRawApi:
|
||||
"""Integration tests for Superset SQL Lab REST API with session auth."""
|
||||
# #region TestSupersetJwtHealthCheck [C:3] [TYPE Class]
|
||||
# @BRIEF Verify health and JWT login work correctly.
|
||||
class TestSupersetJwtHealthCheck:
|
||||
"""JWT auth smoke tests against the real Superset container."""
|
||||
|
||||
# #region _form_login [C:1] [TYPE Function]
|
||||
# @BRIEF Helper: log into Superset via form and return authenticated session.
|
||||
def _form_login(self, superset_url, superset_admin_password) -> requests.Session:
|
||||
session = requests.Session()
|
||||
|
||||
# Step 1: Get CSRF token from login page
|
||||
resp = session.get(f"{superset_url}/login/", timeout=10)
|
||||
resp.raise_for_status()
|
||||
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
|
||||
# Step 2: Submit login form
|
||||
resp = session.post(
|
||||
f"{superset_url}/login/",
|
||||
data={
|
||||
"username": "admin",
|
||||
"password": superset_admin_password,
|
||||
"csrf_token": csrf_token,
|
||||
},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
allow_redirects=True,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
|
||||
# Verify session cookie is set
|
||||
assert session.cookies.get("session") is not None
|
||||
return session
|
||||
# #endregion _form_login
|
||||
|
||||
|
||||
# #region test_list_databases_after_form_login [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/ returns paginated list after form login.
|
||||
# @TEST_EDGE: form_auth_works_for_api — VERIFIED_BY: test_list_databases_after_form_login
|
||||
def test_list_databases_after_form_login(self, superset_url, superset_admin_password):
|
||||
session = self._form_login(superset_url, superset_admin_password)
|
||||
|
||||
resp = session.get(
|
||||
f"{superset_url}/api/v1/database/",
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
# After form login, Superset API should accept session cookie
|
||||
assert resp.status_code in (200, 401), \
|
||||
f"Expected 200/401, got {resp.status_code}"
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# Should contain paginated result
|
||||
assert "result" in data or "count" in data, \
|
||||
f"Unexpected response shape: {list(data.keys())}"
|
||||
# #endregion test_list_databases_after_form_login
|
||||
|
||||
|
||||
# #region test_sqllab_execute_endpoint_reachable [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ is reachable (may fail on DB not found — that's OK).
|
||||
# @TEST_INVARIANT sqllab_execute_accepts_sql — VERIFIED_BY: test_sqllab_execute_endpoint_reachable
|
||||
def test_sqllab_execute_endpoint_reachable(self, superset_url, superset_admin_password):
|
||||
session = self._form_login(superset_url, superset_admin_password)
|
||||
|
||||
# We don't have a configured database in the container, but the endpoint
|
||||
# should be reachable and return a structured error (not 404)
|
||||
payload = {
|
||||
"database_id": 1,
|
||||
"sql": "SELECT 1",
|
||||
"runAsync": True,
|
||||
}
|
||||
|
||||
resp = session.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json=payload,
|
||||
headers={
|
||||
"Referer": f"{superset_url}/",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
# The endpoint exists: returns 200/400/500 depending on DB state
|
||||
# 200 = execution started, 400 = bad request, 500 = DB not found
|
||||
assert resp.status_code in (200, 400, 500), \
|
||||
(f"sqllab/execute/ returned unexpected {resp.status_code}: "
|
||||
f"{resp.text[:200]}")
|
||||
# #endregion test_sqllab_execute_endpoint_reachable
|
||||
|
||||
|
||||
# #region test_sqllab_execute_rejects_naive_sql [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ with missing database_id returns error.
|
||||
# @TEST_EDGE: missing_database_id — VERIFIED_BY: test_sqllab_execute_rejects_naive_sql
|
||||
def test_sqllab_execute_rejects_naive_sql(self, superset_url, superset_admin_password):
|
||||
session = self._form_login(superset_url, superset_admin_password)
|
||||
|
||||
# No database_id in payload
|
||||
resp = session.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json={"sql": "SELECT 1"},
|
||||
headers={
|
||||
"Referer": f"{superset_url}/",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
# Should return 400 (bad request) because database_id is required
|
||||
assert resp.status_code in (400, 500), \
|
||||
f"Expected 400/500, got {resp.status_code}: {resp.text[:200]}"
|
||||
# #endregion test_sqllab_execute_rejects_naive_sql
|
||||
|
||||
|
||||
# #region test_csrf_token_required_for_mutations [C:2] [TYPE Function]
|
||||
# @BRIEF POST without CSRF token returns 403.
|
||||
# @TEST_EDGE: missing_csrf_token — VERIFIED_BY: test_csrf_token_required_for_mutations
|
||||
def test_csrf_token_required_for_mutations(self, superset_url, superset_admin_password):
|
||||
session = self._form_login(superset_url, superset_admin_password)
|
||||
|
||||
# POST without the CSRF header should fail
|
||||
resp = session.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json={"sql": "SELECT 1", "database_id": 1},
|
||||
# NOTE: requests.Session automatically handles cookies,
|
||||
# but Superset also requires a CSRF token in the header
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
# Without Referer or X-CSRFToken, Superset should reject with 403
|
||||
assert resp.status_code != 200, \
|
||||
"Expected CSRF protection to block missing header"
|
||||
# #endregion test_csrf_token_required_for_mutations
|
||||
|
||||
|
||||
# #region test_database_columns_endpoint [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/<id>/columns/ for non-existent DB returns 404.
|
||||
def test_database_columns_endpoint(self, superset_url, superset_admin_password):
|
||||
session = self._form_login(superset_url, superset_admin_password)
|
||||
|
||||
resp = session.get(
|
||||
f"{superset_url}/api/v1/database/99999/columns/",
|
||||
timeout=10,
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert resp.status_code in (404, 401, 500), \
|
||||
f"Expected 404/401/500, got {resp.status_code}"
|
||||
# #endregion test_database_columns_endpoint
|
||||
|
||||
|
||||
# #endregion TestSupersetSqllabRawApi
|
||||
|
||||
|
||||
# #region TestConfigManagerSupersetIntegration [C:3] [TYPE Class]
|
||||
# @BRIEF Verify ConfigManager integration with real Superset URL/credentials.
|
||||
class TestConfigManagerSupersetIntegration:
|
||||
"""Integration tests for ConfigManager with real Superset container."""
|
||||
|
||||
# #region test_config_manager_superset_url [C:2] [TYPE Function]
|
||||
# @BRIEF ConfigManager with test environment connected to real Superset container.
|
||||
def test_config_manager_superset_url(
|
||||
self, superset_url, superset_admin_password, mock_config_manager
|
||||
):
|
||||
"""Verify mock_config_manager's environment URL matches real Superset."""
|
||||
env = mock_config_manager.get_environment("test_env")
|
||||
assert env is not None
|
||||
assert env.url is not None
|
||||
# The mock_config_manager's URL can be overridden to match the container
|
||||
# This test confirms the fixture wiring is correct
|
||||
assert len(env.url) > 0
|
||||
# #endregion test_config_manager_superset_url
|
||||
|
||||
|
||||
# #endregion TestConfigManagerSupersetIntegration
|
||||
|
||||
|
||||
# #region TestSupersetClientRawApi [C:3] [TYPE Class]
|
||||
# @BRIEF Verify SupersetClient can communicate with real container using JWT (known to fail).
|
||||
# @RATIONALE Document the JWT auth limitation: Superset 4.1.2 `fab create-admin` user
|
||||
# cannot obtain JWT token via /api/v1/security/login. This test proves the limitation
|
||||
# and prevents regression if a future Superset version fixes it.
|
||||
# @REJECTED Attempting to work around JWT limitation — it's a Superset/FAB permission
|
||||
# issue that cannot be solved at the client level.
|
||||
class TestSupersetClientRawApi:
|
||||
"""Document JWT auth limitation with the test container."""
|
||||
|
||||
# #region test_jwt_login_fails_as_expected [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/security/login returns 401 — documented limitation.
|
||||
# @TEST_EDGE: jwt_auth_limited — VERIFIED_BY: test_jwt_login_fails_as_expected
|
||||
def test_jwt_login_fails_as_expected(self, superset_url, superset_admin_password):
|
||||
"""Verify JWT auth fails as documented in ADR-0012."""
|
||||
# #region test_jwt_login_returns_token [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/security/login returns access_token with refresh flag.
|
||||
def test_jwt_login_returns_token(self, superset_url, superset_admin_password):
|
||||
resp = requests.post(
|
||||
f"{superset_url}/api/v1/security/login",
|
||||
json={
|
||||
"username": "admin",
|
||||
"password": superset_admin_password,
|
||||
"provider": "db",
|
||||
"refresh": True,
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 401, \
|
||||
(f"JWT login returned {resp.status_code} — if this passes, "
|
||||
f"update ADR-0012 because JWT auth is now working!")
|
||||
assert "Not authorized" in resp.text or "401" in resp.text, \
|
||||
f"Unexpected error: {resp.text[:200]}"
|
||||
# #endregion test_jwt_login_fails_as_expected
|
||||
assert resp.status_code == 200, f"JWT login failed: {resp.status_code} {resp.text}"
|
||||
data = resp.json()
|
||||
assert "access_token" in data, f"No access_token in response: {data.keys()}"
|
||||
assert "refresh_token" in data, f"No refresh_token in response: {data.keys()}"
|
||||
assert len(data["access_token"]) > 20, "access_token too short"
|
||||
# #endregion test_jwt_login_returns_token
|
||||
|
||||
|
||||
# #region test_health_still_works_without_auth [C:2] [TYPE Function]
|
||||
# @BRIEF /health does not require authentication — regression guard.
|
||||
def test_health_still_works_without_auth(self, superset_url):
|
||||
# #region test_jwt_login_wrong_password [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/security/login with wrong password returns 401.
|
||||
# @TEST_EDGE: wrong_password — VERIFIED_BY: test_jwt_login_wrong_password
|
||||
def test_jwt_login_wrong_password(self, superset_url):
|
||||
resp = requests.post(
|
||||
f"{superset_url}/api/v1/security/login",
|
||||
json={
|
||||
"username": "admin",
|
||||
"password": "wrong_password_123",
|
||||
"provider": "db",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 401, f"Expected 401, got {resp.status_code}"
|
||||
# #endregion test_jwt_login_wrong_password
|
||||
|
||||
|
||||
# #region test_health_still_works [C:2] [TYPE Function]
|
||||
# @BRIEF /health does not require auth — regression guard.
|
||||
def test_health_still_works(self, superset_url):
|
||||
resp = requests.get(f"{superset_url}/health", timeout=10)
|
||||
assert resp.status_code == 200
|
||||
assert resp.text.strip() == "OK"
|
||||
# #endregion test_health_still_works_without_auth
|
||||
# #endregion test_health_still_works
|
||||
|
||||
|
||||
# #endregion TestSupersetClientRawApi
|
||||
# #endregion TestSupersetJwtHealthCheck
|
||||
|
||||
# #endregion TestSupersetSqllabApi
|
||||
|
||||
# #region TestSupersetApiWithJwt [C:3] [TYPE Class]
|
||||
# @BRIEF Verify REST API endpoints with JWT Bearer token auth.
|
||||
class TestSupersetApiWithJwt:
|
||||
"""Authenticated API tests against real Superset container."""
|
||||
|
||||
# #region test_list_databases_with_jwt [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/ returns paginated list with JWT Bearer token.
|
||||
def test_list_databases_with_jwt(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/database/",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 200, \
|
||||
f"GET /api/v1/database/ failed: {resp.status_code} {resp.text[:200]}"
|
||||
data = resp.json()
|
||||
assert "result" in data, f"No 'result' key: {list(data.keys())}"
|
||||
assert isinstance(data["result"], list), "result should be a list"
|
||||
# #endregion test_list_databases_with_jwt
|
||||
|
||||
|
||||
# #region test_list_dashboards_with_jwt [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/dashboard/ returns paginated list with JWT Bearer token.
|
||||
def test_list_dashboards_with_jwt(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/dashboard/",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 200, \
|
||||
f"GET /api/v1/dashboard/ failed: {resp.status_code} {resp.text[:200]}"
|
||||
data = resp.json()
|
||||
assert "result" in data, f"No 'result' key: {list(data.keys())}"
|
||||
assert "count" in data, f"No 'count' key: {list(data.keys())}"
|
||||
# #endregion test_list_dashboards_with_jwt
|
||||
|
||||
|
||||
# #region test_sqllab_execute_with_jwt [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ is reachable with JWT.
|
||||
def test_sqllab_execute_with_jwt(self, superset_url, superset_jwt_headers):
|
||||
payload = {
|
||||
"database_id": 1,
|
||||
"sql": "SELECT 1",
|
||||
"runAsync": True,
|
||||
}
|
||||
resp = requests.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json=payload,
|
||||
headers={
|
||||
**superset_jwt_headers,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
# Without a real DB configured, we expect 400 or 500
|
||||
# But the endpoint exists and is reachable (not 404/401)
|
||||
assert resp.status_code in (200, 400, 500), \
|
||||
f"sqllab/execute/ returned {resp.status_code}: {resp.text[:300]}"
|
||||
# #endregion test_sqllab_execute_with_jwt
|
||||
|
||||
|
||||
# #region test_sqllab_without_database_id [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ without database_id returns 400.
|
||||
# @TEST_EDGE: missing_database_id — VERIFIED_BY: test_sqllab_without_database_id
|
||||
def test_sqllab_without_database_id(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json={"sql": "SELECT 1"},
|
||||
headers={
|
||||
**superset_jwt_headers,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code in (400, 500), \
|
||||
f"Expected 400/500, got {resp.status_code}: {resp.text[:200]}"
|
||||
# #endregion test_sqllab_without_database_id
|
||||
|
||||
|
||||
# #region test_get_nonexistent_database [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/<non_existent_id> returns 404.
|
||||
def test_get_nonexistent_database(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/database/99999",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 404, \
|
||||
f"Expected 404, got {resp.status_code}: {resp.text[:200]}"
|
||||
# #endregion test_get_nonexistent_database
|
||||
|
||||
|
||||
# #region test_me_endpoint [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/me/ — в Superset 4.1.2 это публичный эндпоинт, но
|
||||
# доступен только авторизованным через JWT. Проверяем что 401/200, а не 404.
|
||||
def test_me_endpoint(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/me/",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
# Superset 4.1.2 может вернуть 200 (есть endpoint) или 401 (нет прав у admin role).
|
||||
# Главное — не 404, значит endpoint существует.
|
||||
assert resp.status_code in (200, 401, 403), \
|
||||
f"GET /api/v1/me/ failed: {resp.status_code} {resp.text[:200]}"
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
assert "result" in data, f"No 'result' key: {list(data.keys())}"
|
||||
# #endregion test_me_endpoint
|
||||
|
||||
|
||||
# #region test_database_columns_with_jwt [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/<id>/columns/ with valid id returns 404 (no DB configured).
|
||||
def test_database_columns_with_jwt(self, superset_url, superset_jwt_headers):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/database/99999/columns/",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 404, \
|
||||
f"Expected 404, got {resp.status_code}"
|
||||
# #endregion test_database_columns_with_jwt
|
||||
|
||||
|
||||
# #region test_unauthenticated_api_rejected [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/dashboard/ without auth returns 401.
|
||||
def test_unauthenticated_api_rejected(self, superset_url):
|
||||
resp = requests.get(
|
||||
f"{superset_url}/api/v1/dashboard/",
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 401, \
|
||||
f"Expected 401, got {resp.status_code}"
|
||||
# #endregion test_unauthenticated_api_rejected
|
||||
|
||||
|
||||
# #endregion TestSupersetApiWithJwt
|
||||
|
||||
|
||||
# #region TestSupersetJwtIntegration [C:3] [TYPE Class]
|
||||
# @BRIEF Verify SupersetClient and SupersetSqlLabExecutor with real JWT auth.
|
||||
class TestSupersetJwtIntegration:
|
||||
"""Full SupersetClient integration tests with JWT auth."""
|
||||
|
||||
# #region test_superset_client_authenticated [C:2] [TYPE Function]
|
||||
# @BRIEF Real SupersetClient authenticates and can call API.
|
||||
@pytest.mark.asyncio
|
||||
async def test_superset_client_authenticated(self, superset_client):
|
||||
# Authenticated client should be able to list databases
|
||||
count, databases = await superset_client.get_databases()
|
||||
assert isinstance(count, int)
|
||||
assert isinstance(databases, list)
|
||||
# #endregion test_superset_client_authenticated
|
||||
|
||||
|
||||
# #region test_superset_client_user_via_me_endpoint [C:2] [TYPE Function]
|
||||
# @BRIEF SupersetClient can call /me/ raw endpoint — проверяем доступность.
|
||||
@pytest.mark.asyncio
|
||||
async def test_superset_client_user_via_me_endpoint(self, superset_client):
|
||||
me = await superset_client.client.request("GET", "/me/")
|
||||
assert me is not None
|
||||
# Superset 4.1.2 может вернуть 200 (есть endpoint) или {"message":"Not authorized"}
|
||||
# в зависимости от прав роли Admin на конкретный endpoint
|
||||
# #endregion test_superset_client_user_via_me_endpoint
|
||||
|
||||
|
||||
# #endregion TestSupersetJwtIntegration
|
||||
|
||||
|
||||
# #endregion TestSupersetSqllabApiJwt
|
||||
|
||||
227
backend/tests/integration/test_superset_sqllab_e2e.py
Normal file
227
backend/tests/integration/test_superset_sqllab_e2e.py
Normal file
@@ -0,0 +1,227 @@
|
||||
# #region SupersetSqllabE2E [C:5] [TYPE Module] [SEMANTICS test,superset,sqllab,e2e,execution]
|
||||
# @BRIEF End-to-end: register real PostgreSQL in Superset → execute SQL via SupersetClient.
|
||||
# @RELATION BINDS_TO -> [SupersetSqlLabExecutor]
|
||||
# @RELATION BINDS_TO -> [SupersetClient]
|
||||
#
|
||||
# @TEST_CONTRACT SupersetSqlLabE2E ->
|
||||
# {
|
||||
# invariants: [
|
||||
# "SupersetClient.create_database() registers a real Postgres DB via REST API",
|
||||
# "SupersetClient handles JWT + CSRF internally for mutation endpoints",
|
||||
# "SQL Lab execute returns query_id for valid SQL on registered DB",
|
||||
# "SupersetSqlLabExecutor can resolve DB and execute SQL end-to-end"
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# @REQUIRES Docker daemon running (testcontainers starts superset + postgres)
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
# #region TestSupersetSqllabEndToEnd [C:4] [TYPE Class]
|
||||
class TestSupersetSqllabEndToEnd:
|
||||
"""Full E2E: PostgreSQL → Superset registration → SQL Lab via SupersetClient."""
|
||||
|
||||
@staticmethod
|
||||
def _setup_test_table(pg_engine):
|
||||
with pg_engine.begin() as conn:
|
||||
conn.execute(text("DROP TABLE IF EXISTS ss_e2e_test"))
|
||||
conn.execute(text(
|
||||
"CREATE TABLE ss_e2e_test ("
|
||||
" id SERIAL PRIMARY KEY,"
|
||||
" name VARCHAR(100) NOT NULL,"
|
||||
" value INTEGER NOT NULL"
|
||||
")"
|
||||
))
|
||||
conn.execute(text(
|
||||
"INSERT INTO ss_e2e_test (name, value) VALUES "
|
||||
"('alpha', 10), ('beta', 20), ('gamma', 30)"
|
||||
))
|
||||
|
||||
@staticmethod
|
||||
def _cleanup_test_table(pg_engine):
|
||||
with pg_engine.begin() as conn:
|
||||
conn.execute(text("DROP TABLE IF EXISTS ss_e2e_test"))
|
||||
|
||||
# #region test_register_db_and_execute_sql [C:2] [TYPE Function]
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_db_and_execute_sql(
|
||||
self, superset_client, superset_db_url, pg_engine,
|
||||
):
|
||||
"""Register Postgres DB → execute SQL via SupersetClient (handles CSRF)."""
|
||||
self._setup_test_table(pg_engine)
|
||||
db_id = None
|
||||
try:
|
||||
# Build Postgres URI with Docker bridge IP
|
||||
parsed = urlparse(superset_db_url)
|
||||
register_uri = f"postgresql://test:test@{parsed.hostname}:{parsed.port}/test_translate"
|
||||
|
||||
# Register DB via SupersetClient.create_database()
|
||||
reg_resp = await superset_client.create_database(
|
||||
database_name="E2E Test DB",
|
||||
sqlalchemy_uri=register_uri,
|
||||
expose_in_sqllab=True,
|
||||
allow_dml=False,
|
||||
)
|
||||
db_id = reg_resp.get("id")
|
||||
if not db_id:
|
||||
# Fallback: list and find
|
||||
_, dbs = await superset_client.get_databases()
|
||||
target = next(
|
||||
(d for d in dbs if "E2E" in d.get("database_name", "")),
|
||||
None,
|
||||
)
|
||||
if target:
|
||||
db_id = target["id"]
|
||||
else:
|
||||
pytest.skip(f"DB reg failed: {str(reg_resp)[:300]}")
|
||||
|
||||
# Execute SQL via client.request()
|
||||
exec_resp = await superset_client.client.request(
|
||||
"POST", "/sqllab/execute/",
|
||||
data={
|
||||
"database_id": db_id,
|
||||
"sql": "SELECT * FROM ss_e2e_test ORDER BY id",
|
||||
"runAsync": False,
|
||||
},
|
||||
)
|
||||
|
||||
# Verify inline results
|
||||
result_part = exec_resp.get("result", exec_resp)
|
||||
if isinstance(result_part, dict) and "columns" in result_part:
|
||||
data_rows = result_part.get("data", [])
|
||||
assert len(data_rows) == 3
|
||||
col_names = [
|
||||
c.get("column_name", c.get("name", ""))
|
||||
for c in result_part["columns"]
|
||||
]
|
||||
assert "name" in col_names
|
||||
assert "value" in col_names
|
||||
return # success
|
||||
|
||||
# Poll for async results
|
||||
query_id = exec_resp.get("query", {}).get("id") or exec_resp.get("id")
|
||||
if query_id:
|
||||
import asyncio
|
||||
for attempt in range(15):
|
||||
await asyncio.sleep(1)
|
||||
result_data = await superset_client.client.request(
|
||||
"GET", f"/sqllab/results/{query_id}/",
|
||||
)
|
||||
for key in ("result", "data", "query"):
|
||||
part = result_data.get(key, {})
|
||||
if isinstance(part, dict) and "columns" in part:
|
||||
data_rows = part.get("data", [])
|
||||
if len(data_rows) == 3:
|
||||
return # success
|
||||
if result_data.get("status") in ("pending", "running"):
|
||||
continue
|
||||
pytest.fail(f"No valid results after 15s for query {query_id}")
|
||||
else:
|
||||
pytest.skip(f"No query_id: {list(exec_resp.keys())}")
|
||||
finally:
|
||||
self._cleanup_test_table(pg_engine)
|
||||
if db_id:
|
||||
try:
|
||||
await superset_client.delete_database(db_id)
|
||||
except Exception:
|
||||
pass
|
||||
# #endregion test_register_db_and_execute_sql
|
||||
# #endregion TestSupersetSqllabEndToEnd
|
||||
|
||||
|
||||
# #region TestSupersetSqlLabExecutorE2E [C:4] [TYPE Class]
|
||||
class TestSupersetSqlLabExecutorE2E:
|
||||
"""SupersetSqlLabExecutor against a real registered database."""
|
||||
|
||||
# #region test_executor_against_real_db [C:2] [TYPE Function]
|
||||
@pytest.mark.asyncio
|
||||
async def test_executor_against_real_db(
|
||||
self, superset_client,
|
||||
superset_admin_password, superset_db_url, pg_engine,
|
||||
):
|
||||
"""SupersetSqlLabExecutor resolves DB and executes SQL end-to-end."""
|
||||
from unittest.mock import MagicMock
|
||||
from src.plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
|
||||
TestSupersetSqllabEndToEnd._setup_test_table(pg_engine)
|
||||
|
||||
config_manager = MagicMock()
|
||||
config_manager.get_environments.return_value = [superset_client.env]
|
||||
config_manager.get_environment.return_value = superset_client.env
|
||||
config_manager.get_config.return_value = MagicMock()
|
||||
|
||||
executor = SupersetSqlLabExecutor(config_manager, "test_env")
|
||||
db_id = None
|
||||
|
||||
try:
|
||||
# Register DB via SupersetClient.create_database()
|
||||
parsed = urlparse(superset_db_url)
|
||||
register_uri = f"postgresql://test:test@{parsed.hostname}:{parsed.port}/test_translate"
|
||||
reg_resp = await superset_client.create_database(
|
||||
database_name="Executor E2E DB",
|
||||
sqlalchemy_uri=register_uri,
|
||||
expose_in_sqllab=True,
|
||||
allow_dml=False,
|
||||
)
|
||||
db_id = reg_resp.get("id")
|
||||
if not db_id:
|
||||
pytest.skip(f"DB reg: {str(reg_resp)[:200]}")
|
||||
|
||||
# Resolve database_id via executor
|
||||
resolved_id = await executor.resolve_database_id(
|
||||
database_name="Executor E2E DB",
|
||||
)
|
||||
assert isinstance(resolved_id, int) and resolved_id > 0
|
||||
assert resolved_id == db_id
|
||||
|
||||
# Execute SQL with runAsync=False (inline results in raw_response)
|
||||
result = await executor.execute_sql(
|
||||
"SELECT * FROM ss_e2e_test ORDER BY id",
|
||||
database_id=db_id, run_async=False,
|
||||
)
|
||||
|
||||
# Inline results from execute_sql's raw_response
|
||||
raw = result.get("raw_response", {})
|
||||
result_part = raw.get("result", raw)
|
||||
if isinstance(result_part, dict) and "columns" in result_part:
|
||||
data_rows = result_part.get("data", [])
|
||||
assert len(data_rows) == 3, \
|
||||
f"Expected 3 rows, got {len(data_rows)}"
|
||||
col_names = [
|
||||
c.get("column_name", c.get("name", ""))
|
||||
for c in result_part["columns"]
|
||||
]
|
||||
assert "name" in col_names
|
||||
assert "value" in col_names
|
||||
return # success
|
||||
|
||||
# Fallback: poll with runAsync=True
|
||||
query_id = result.get("query_id")
|
||||
if query_id:
|
||||
import asyncio
|
||||
for attempt in range(15):
|
||||
await asyncio.sleep(1)
|
||||
qr = await executor.get_query_results(query_id)
|
||||
if qr and qr.get("status") == "success" and qr.get("results"):
|
||||
assert len(qr["results"]) == 3
|
||||
return
|
||||
pytest.fail(f"No results for query {query_id}")
|
||||
else:
|
||||
pytest.skip(f"No query_id or inline results: {list(result.keys())}")
|
||||
finally:
|
||||
TestSupersetSqllabEndToEnd._cleanup_test_table(pg_engine)
|
||||
if db_id:
|
||||
try:
|
||||
await superset_client.delete_database(db_id)
|
||||
except Exception:
|
||||
pass
|
||||
# #endregion test_executor_against_real_db
|
||||
# #endregion TestSupersetSqlLabExecutorE2E
|
||||
# #endregion SupersetSqllabE2E
|
||||
@@ -1,22 +1,23 @@
|
||||
# #region SupersetSqlLabExecutorIntegration [C:4] [TYPE Module] [SEMANTICS test,superset,sqllab,executor,integration]
|
||||
# @BRIEF Integration tests for SupersetSqlLabExecutor using real Testcontainers Superset.
|
||||
# #region SupersetSqlLabExecutorIntegrationJwt [C:4] [TYPE Module] [SEMANTICS test,superset,sqllab,executor,integration,jwt]
|
||||
# @BRIEF Integration tests for SupersetSqlLabExecutor using real JWT auth and testcontainers.
|
||||
# @RELATION BINDS_TO -> [SupersetSqlLabExecutor]
|
||||
# @RELATION BINDS_TO -> [SupersetClient]
|
||||
# @RELATION BINDS_TO -> [ConfigManager]
|
||||
#
|
||||
# @TEST_CONTRACT SupersetSqlLabExecutor ->
|
||||
# {
|
||||
# invariants: [
|
||||
# "Form-based session auth works with Superset REST API",
|
||||
# "SQL Lab execute endpoint accepts valid SQL and returns execution reference",
|
||||
# "Database ID resolution works via real API",
|
||||
# "SupersetClient construction works with test container URL"
|
||||
# "JWT login works against the test container (ADR-0012 fixed)",
|
||||
# "SupersetSqlLabExecutor resolves database IDs via real API",
|
||||
# "SupersetSqlLabExecutor accepts SQL and returns execution reference",
|
||||
# "SupersetClient.get_databases() returns paginated list",
|
||||
# "SupersetClient.get_database() returns single DB info"
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# @RATIONALE JWT auth is unavailable on the test container (ADR-0012). These
|
||||
# tests work around the limitation by using form-based session auth with raw
|
||||
# httpx calls, or by patching only the jwt auth step in SupersetSqlLabExecutor
|
||||
# while letting all other HTTP calls go through the real container.
|
||||
# @RATIONALE ADR-0012 resolved JWT auth: psycopg2 installs driver for Postgres,
|
||||
# superset_config.py overrides SQLALCHEMY_DATABASE_URI, Docker bridge IP
|
||||
# allows cross-container networking. All tests use real Superset HTTP calls.
|
||||
#
|
||||
# @REQUIRES Docker daemon running (testcontainers starts superset + postgres)
|
||||
import sys
|
||||
@@ -26,211 +27,57 @@ sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import re
|
||||
from unittest.mock import AsyncMock, patch, MagicMock
|
||||
|
||||
|
||||
# #region TestSupersetSqlLabExecutorFormAuth [C:3] [TYPE Class]
|
||||
# @BRIEF Test Superset REST API endpoints with form-based session auth.
|
||||
class TestSupersetSqlLabExecutorFormAuth:
|
||||
"""Integration tests using form-based auth (works) instead of JWT (broken on 4.1.2)."""
|
||||
# #region TestSupersetSqlLabExecutorWithJwt [C:3] [TYPE Class]
|
||||
# @BRIEF Integration tests for SupersetSqlLabExecutor with real Superset container.
|
||||
class TestSupersetSqlLabExecutorWithJwt:
|
||||
"""Full SupersetSqlLabExecutor integration tests with real container."""
|
||||
|
||||
# #region _form_login_client [C:1] [TYPE Function]
|
||||
# @BRIEF Helper: create authenticated httpx.AsyncClient with form-based session cookies.
|
||||
@staticmethod
|
||||
async def _form_login_client(superset_url: str, superset_admin_password: str) -> httpx.AsyncClient:
|
||||
"""Build an AsyncClient pre-authenticated via form-based login."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Step 1: Get CSRF token
|
||||
resp = await client.get(f"{superset_url}/login/", timeout=10)
|
||||
resp.raise_for_status()
|
||||
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
|
||||
# Step 2: Submit login form
|
||||
resp = await client.post(
|
||||
f"{superset_url}/login/",
|
||||
data={
|
||||
"username": "admin",
|
||||
"password": superset_admin_password,
|
||||
"csrf_token": csrf_token,
|
||||
},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
|
||||
# Return a NEW AsyncClient that inherited the cookies from the first one
|
||||
# Actually, we need to share cookies — use a single client
|
||||
return client
|
||||
# #endregion _form_login_client
|
||||
|
||||
|
||||
# #region test_list_databases_via_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/ with form-based session auth via httpx.
|
||||
# #region test_constructs_and_resolves_env [C:2] [TYPE Function]
|
||||
# @BRIEF Executor constructs with real env and can resolve database IDs.
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_databases_via_httpx(self, superset_url, superset_admin_password):
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Login
|
||||
resp = await client.get(f"{superset_url}/login/", timeout=10)
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
|
||||
await client.post(
|
||||
f"{superset_url}/login/",
|
||||
data={"username": "admin", "password": superset_admin_password, "csrf_token": csrf_token},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
# List databases
|
||||
resp = await client.get(
|
||||
f"{superset_url}/api/v1/database/",
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code in (200, 401), \
|
||||
f"Expected 200/401, got {resp.status_code}"
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
assert "result" in data, f"No 'result' key in response: {data.keys()}"
|
||||
# #endregion test_list_databases_via_httpx
|
||||
|
||||
|
||||
# #region test_sqllab_execute_via_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ with form auth, check endpoint is alive.
|
||||
@pytest.mark.asyncio
|
||||
async def test_sqllab_execute_via_httpx(self, superset_url, superset_admin_password):
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Login
|
||||
resp = await client.get(f"{superset_url}/login/", timeout=10)
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
|
||||
await client.post(
|
||||
f"{superset_url}/login/",
|
||||
data={"username": "admin", "password": superset_admin_password, "csrf_token": csrf_token},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
# Try SQL Lab execute
|
||||
resp = await client.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json={"database_id": 1, "sql": "SELECT 1", "runAsync": True},
|
||||
headers={
|
||||
"Referer": f"{superset_url}/",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
# 200 = OK, 400/500 = endpoint exists but no real DB configured
|
||||
assert resp.status_code in (200, 400, 500), \
|
||||
f"sqllab/execute/ returned {resp.status_code}: {resp.text[:200]}"
|
||||
# #endregion test_sqllab_execute_via_httpx
|
||||
|
||||
|
||||
# #region test_get_database_by_id_via_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/<id> with form auth — check structured error for non-existent DB.
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_database_by_id_via_httpx(self, superset_url, superset_admin_password):
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Login
|
||||
resp = await client.get(f"{superset_url}/login/", timeout=10)
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
|
||||
await client.post(
|
||||
f"{superset_url}/login/",
|
||||
data={"username": "admin", "password": superset_admin_password, "csrf_token": csrf_token},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
# Get a non-existent database
|
||||
resp = await client.get(
|
||||
f"{superset_url}/api/v1/database/99999",
|
||||
timeout=10,
|
||||
)
|
||||
# Should return 404 for non-existent DB
|
||||
assert resp.status_code in (404, 401, 200), \
|
||||
f"Expected 404/401, got {resp.status_code}"
|
||||
# #endregion test_get_database_by_id_via_httpx
|
||||
|
||||
|
||||
# #endregion TestSupersetSqlLabExecutorFormAuth
|
||||
|
||||
|
||||
# #region TestSupersetSqlLabExecutorIntegration [C:3] [TYPE Class]
|
||||
# @BRIEF Test SupersetSqlLabExecutor with patched JWT auth but real HTTP calls.
|
||||
# @RATIONALE We patch only the JWT authentication step (known limitation per ADR-0012)
|
||||
# while using the real container URL and verifying all HTTP call shapes.
|
||||
class TestSupersetSqlLabExecutorIntegration:
|
||||
"""Integration tests for SupersetSqlLabExecutor patching only JWT auth."""
|
||||
|
||||
# #region test_executor_constructs_with_env [C:2] [TYPE Function]
|
||||
# @BRIEF SupersetSqlLabExecutor constructs with a real config manager.
|
||||
@pytest.mark.asyncio
|
||||
async def test_executor_constructs_with_env(
|
||||
self, superset_url, mock_config_manager
|
||||
async def test_constructs_and_resolves_env(
|
||||
self, superset_env, superset_url, superset_admin_password
|
||||
):
|
||||
"""Verify executor initializes with environment config."""
|
||||
"""Verify executor can resolve database_id from a real Superset instance."""
|
||||
from unittest.mock import MagicMock
|
||||
from src.plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
|
||||
executor = SupersetSqlLabExecutor(mock_config_manager, "test_env")
|
||||
assert executor.env_id == "test_env"
|
||||
assert executor._client is None # Lazy init
|
||||
# #endregion test_executor_constructs_with_env
|
||||
# Build a real ConfigManager that returns our container environment
|
||||
from src.core.config_models import Environment
|
||||
|
||||
config_manager = MagicMock()
|
||||
config_manager.get_environments.return_value = [superset_env]
|
||||
config_manager.get_environment.return_value = superset_env
|
||||
config_manager.get_config.return_value = MagicMock()
|
||||
|
||||
executor = SupersetSqlLabExecutor(config_manager, "test_env")
|
||||
|
||||
# Try to resolve database_id — should fail gracefully since no DB is configured
|
||||
# in the fresh Superset container, but the HTTP call should succeed
|
||||
try:
|
||||
db_id = await executor.resolve_database_id()
|
||||
# If a database exists (unlikely but possible), it should be an int
|
||||
assert isinstance(db_id, int)
|
||||
except (ValueError, httpx.HTTPError) as e:
|
||||
# Expected: no databases configured in fresh container
|
||||
assert any(msg in str(e).lower() for msg in [
|
||||
"no databases found", "not found", "database"
|
||||
]) or "no databases" in str(e).lower()
|
||||
# #endregion test_constructs_and_resolves_env
|
||||
|
||||
|
||||
# #region test_resolve_database_id_fails_gracefully [C:2] [TYPE Function]
|
||||
# @BRIEF resolve_database_id raises error — JWT auth is not available.
|
||||
# #region test_execute_sql_via_real_executor [C:2] [TYPE Function]
|
||||
# @BRIEF Executor.execute_sql() is callable and returns expected error for no-DB scenario.
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_database_id_fails_gracefully(
|
||||
self, superset_url, mock_config_manager
|
||||
):
|
||||
"""Verify executor fails gracefully when JWT auth is unavailable (ADR-0012)."""
|
||||
from src.plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
|
||||
executor = SupersetSqlLabExecutor(mock_config_manager, "test_env")
|
||||
|
||||
# JWT auth fails on the test container — expect any error
|
||||
with pytest.raises(Exception):
|
||||
await executor.resolve_database_id(
|
||||
database_name="examples",
|
||||
)
|
||||
# #endregion test_resolve_database_id_fails_gracefully
|
||||
|
||||
|
||||
# #region test_execute_sql_structure [C:2] [TYPE Function]
|
||||
# @BRIEF SupersetSqlLabExecutor.execute_sql returns dict with expected keys.
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_sql_structure(self):
|
||||
"""Verify the SupersetSqlLabExecutor is structured correctly via module import."""
|
||||
from src.plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
|
||||
# Just verify the class exists and has expected methods
|
||||
assert hasattr(SupersetSqlLabExecutor, "execute_sql")
|
||||
assert hasattr(SupersetSqlLabExecutor, "resolve_database_id")
|
||||
assert hasattr(SupersetSqlLabExecutor, "_get_client")
|
||||
assert hasattr(SupersetSqlLabExecutor, "execute_and_poll")
|
||||
assert hasattr(SupersetSqlLabExecutor, "get_query_results")
|
||||
assert hasattr(SupersetSqlLabExecutor, "poll_execution_status")
|
||||
# #endregion test_execute_sql_structure
|
||||
|
||||
|
||||
# #region test_get_databases_raw [C:2] [TYPE Function]
|
||||
# @BRIEF Test raw /api/v1/database/ endpoint structure via SupersetClient.
|
||||
# @RATIONAVE Even though JWT fails, we verify the SupersetClient constructs
|
||||
# correctly and the underlying AsyncAPIClient is set up with proper URL.
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_databases_from_container(
|
||||
async def test_execute_sql_via_real_executor(
|
||||
self, superset_url, superset_admin_password
|
||||
):
|
||||
"""Use httpx directly to verify the database API response shape."""
|
||||
"""Verify execute_sql structure — errors are expected but must be Superset API errors."""
|
||||
from unittest.mock import MagicMock
|
||||
from src.plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
from src.core.config_models import Environment
|
||||
from src.core.superset_client import SupersetClient
|
||||
|
||||
env = Environment(
|
||||
id="test_env",
|
||||
@@ -239,25 +86,168 @@ class TestSupersetSqlLabExecutorIntegration:
|
||||
username="admin",
|
||||
password=superset_admin_password,
|
||||
verify_ssl=False,
|
||||
timeout=10,
|
||||
timeout=30,
|
||||
)
|
||||
config_manager = MagicMock()
|
||||
config_manager.get_environments.return_value = [env]
|
||||
config_manager.get_environment.return_value = env
|
||||
config_manager.get_config.return_value = MagicMock()
|
||||
|
||||
client = SupersetClient(env)
|
||||
executor = SupersetSqlLabExecutor(config_manager, "test_env")
|
||||
|
||||
# Verify client constructed with the right URL
|
||||
assert client.env.url == superset_url
|
||||
assert client.client is not None
|
||||
assert client.client.base_url == superset_url
|
||||
# #endregion test_get_databases_from_container
|
||||
# Try executing SQL — should raise an error (no database configured)
|
||||
# but this proves the HTTP pipeline works end-to-end
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
await executor.execute_sql("SELECT 1")
|
||||
|
||||
error_msg = str(exc_info.value).lower()
|
||||
assert any(term in error_msg for term in [
|
||||
"database", "not found", "no databases", "superset", "401", "403", "500"
|
||||
]), f"Unexpected error: {error_msg}"
|
||||
# #endregion test_execute_sql_via_real_executor
|
||||
|
||||
|
||||
# #endregion TestSupersetSqlLabExecutorIntegration
|
||||
# #endregion TestSupersetSqlLabExecutorWithJwt
|
||||
|
||||
|
||||
# #region TestSupersetClientMethods [C:3] [TYPE Class]
|
||||
# @BRIEF Full integration tests for SupersetClient methods with real container.
|
||||
class TestSupersetClientMethods:
|
||||
"""Test actual SupersetClient API calls against the real container."""
|
||||
|
||||
# #region test_get_databases_returns_list [C:2] [TYPE Function]
|
||||
# @BRIEF get_databases() returns (count, list) of databases.
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_databases_returns_list(self, superset_client):
|
||||
count, databases = await superset_client.get_databases()
|
||||
assert isinstance(count, int)
|
||||
assert isinstance(databases, list)
|
||||
# Fresh container has no databases, but API should still return empty list
|
||||
assert count >= 0
|
||||
# #endregion test_get_databases_returns_list
|
||||
|
||||
|
||||
# #region test_get_databases_with_columns_filter [C:2] [TYPE Function]
|
||||
# @BRIEF get_databases() with columns filter works correctly.
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_databases_with_columns_filter(self, superset_client):
|
||||
_, databases = await superset_client.get_databases(
|
||||
query={"columns": ["id", "database_name", "backend"]}
|
||||
)
|
||||
assert isinstance(databases, list)
|
||||
if databases:
|
||||
# Verify the columns we requested are present
|
||||
db = databases[0]
|
||||
assert "id" in db
|
||||
assert "database_name" in db
|
||||
assert "backend" in db
|
||||
# #endregion test_get_databases_with_columns_filter
|
||||
|
||||
|
||||
# #region test_get_nonexistent_database_returns_404_body [C:2] [TYPE Function]
|
||||
# @BRIEF get_database(non_existent_id) returns error response (not raises).
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_nonexistent_database_returns_404_body(self, superset_client):
|
||||
result = await superset_client.get_database(99999)
|
||||
# SupersetClient.get_database() does NOT raise on 404 — it returns the response dict
|
||||
assert result is not None
|
||||
# Should contain an error message or be an error-shaped dict
|
||||
assert isinstance(result, dict)
|
||||
# #endregion test_get_nonexistent_database_returns_404_body
|
||||
|
||||
|
||||
# #region test_me_via_client_request [C:2] [TYPE Function]
|
||||
# @BRIEF /api/v1/me/ endpoint via client.request() — проверяем что endpoint жив.
|
||||
@pytest.mark.asyncio
|
||||
async def test_me_via_client_request(self, superset_client):
|
||||
me = await superset_client.client.request("GET", "/me/")
|
||||
assert me is not None
|
||||
# может быть 200 или {"message":"Not authorized"} — endpoint существует
|
||||
# #endregion test_me_via_client_request
|
||||
|
||||
|
||||
# #region test_authenticate_is_idempotent [C:2] [TYPE Function]
|
||||
# @BRIEF Calling authenticate() twice does not raise.
|
||||
@pytest.mark.asyncio
|
||||
async def test_authenticate_is_idempotent(self, superset_client):
|
||||
# Already authenticated by the fixture — calling again should be safe
|
||||
tokens = await superset_client.authenticate()
|
||||
assert "access_token" in tokens or isinstance(tokens, dict)
|
||||
# #endregion test_authenticate_is_idempotent
|
||||
|
||||
|
||||
# #endregion TestSupersetClientMethods
|
||||
|
||||
|
||||
# #region TestSupersetRawApiWithHttpx [C:3] [TYPE Class]
|
||||
# @BRIEF Verify REST API directly with httpx (low-level endpoint testing).
|
||||
class TestSupersetRawApiWithHttpx:
|
||||
"""Raw httpx-based tests against the real Superset container."""
|
||||
|
||||
# #region test_sqllab_execute_raw_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF POST /api/v1/sqllab/execute/ via httpx with JWT token.
|
||||
@pytest.mark.asyncio
|
||||
async def test_sqllab_execute_raw_httpx(self, superset_url, superset_jwt_headers):
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"{superset_url}/api/v1/sqllab/execute/",
|
||||
json={"database_id": 1, "sql": "SELECT 1", "runAsync": True},
|
||||
headers={
|
||||
**superset_jwt_headers,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
# Endpoint is reachable (not 404/401)
|
||||
assert resp.status_code in (200, 400, 500), \
|
||||
f"sqllab/execute/ returned {resp.status_code}: {resp.text[:300]}"
|
||||
# #endregion test_sqllab_execute_raw_httpx
|
||||
|
||||
|
||||
# #region test_database_list_raw_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF GET /api/v1/database/ via httpx with JWT token.
|
||||
@pytest.mark.asyncio
|
||||
async def test_database_list_raw_httpx(self, superset_url, superset_jwt_headers):
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(
|
||||
f"{superset_url}/api/v1/database/",
|
||||
headers=superset_jwt_headers,
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 200, \
|
||||
f"GET /api/v1/database/ failed: {resp.status_code}"
|
||||
data = resp.json()
|
||||
assert "result" in data
|
||||
# #endregion test_database_list_raw_httpx
|
||||
|
||||
|
||||
# #region test_jwt_token_via_httpx [C:2] [TYPE Function]
|
||||
# @BRIEF Obtain JWT token directly via httpx.
|
||||
@pytest.mark.asyncio
|
||||
async def test_jwt_token_via_httpx(self, superset_url, superset_admin_password):
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.post(
|
||||
f"{superset_url}/api/v1/security/login",
|
||||
json={
|
||||
"username": "admin",
|
||||
"password": superset_admin_password,
|
||||
"provider": "db",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
# #endregion test_jwt_token_via_httpx
|
||||
|
||||
|
||||
# #endregion TestSupersetRawApiWithHttpx
|
||||
|
||||
|
||||
# #region TestBatchInsertSupersetIntegration [C:3] [TYPE Class]
|
||||
# @BRIEF Integration tests for batch insert with real Superset container.
|
||||
# @BRIEF Verify batch insert module integration with real Superset.
|
||||
class TestBatchInsertSupersetIntegration:
|
||||
"""Tests that exercise the batch insert pipeline with the real container."""
|
||||
"""Integration tests for the batch insert pipeline."""
|
||||
|
||||
# #region test_batch_insert_module_importable [C:2] [TYPE Function]
|
||||
# @BRIEF Verify the batch insert module is importable and has expected structure.
|
||||
@@ -267,50 +257,17 @@ class TestBatchInsertSupersetIntegration:
|
||||
# #endregion test_batch_insert_module_importable
|
||||
|
||||
|
||||
# #region test_superset_executor_resolve_database_raw_api [C:2] [TYPE Function]
|
||||
# @BRIEF Verify the databases endpoint shape matches what resolve_database_id expects.
|
||||
@pytest.mark.asyncio
|
||||
async def test_superset_executor_resolve_database_raw_api(
|
||||
self, superset_url, superset_admin_password
|
||||
):
|
||||
"""Use httpx to verify database API response fields match executor expectations."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Login via form
|
||||
resp = await client.get(f"{superset_url}/login/", timeout=10)
|
||||
csrf_match = re.search(r'csrf_token["\s:=]+[\'"]([^\'"]+)', resp.text)
|
||||
csrf_token = csrf_match.group(1) if csrf_match else ""
|
||||
await client.post(
|
||||
f"{superset_url}/login/",
|
||||
data={"username": "admin", "password": superset_admin_password, "csrf_token": csrf_token},
|
||||
headers={"Referer": f"{superset_url}/login/"},
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
# GET /api/v1/database/ with columns filter
|
||||
resp = await client.get(
|
||||
f"{superset_url}/api/v1/database/?q=(columns:!(id,database_name,backend))",
|
||||
timeout=10,
|
||||
)
|
||||
# The API should respond — either 200 with data or 401 without JWT
|
||||
assert resp.status_code in (200, 401), \
|
||||
f"Unexpected status: {resp.status_code}"
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# Verify response shape matches executor's expectation
|
||||
assert "result" in data, f"Missing 'result' key: {data.keys()}"
|
||||
if data["result"]:
|
||||
db = data["result"][0]
|
||||
# Verify fields used by resolve_database_id
|
||||
db_id = db.get("id")
|
||||
db_name = db.get("database_name")
|
||||
db_backend = db.get("backend")
|
||||
assert db_id is not None, "Missing 'id' in database entry"
|
||||
# Log for debugging
|
||||
print(f" Database found: id={db_id}, name={db_name}, backend={db_backend}")
|
||||
# #endregion test_superset_executor_resolve_database_raw_api
|
||||
# #region test_superset_config_manager_real_env [C:2] [TYPE Function]
|
||||
# @BRIEF ConfigManager with real environment pointing to Superset container.
|
||||
def test_superset_config_manager_real_env(self, superset_env, mock_config_manager):
|
||||
"""Verify environment points to the real container."""
|
||||
env = mock_config_manager.get_environment("test_env")
|
||||
assert env is not None
|
||||
assert env.username == "admin"
|
||||
assert env.timeout == 30
|
||||
# #endregion test_superset_config_manager_real_env
|
||||
|
||||
|
||||
# #endregion TestBatchInsertSupersetIntegration
|
||||
|
||||
# #endregion SupersetSqlLabExecutorIntegration
|
||||
# #endregion SupersetSqlLabExecutorIntegrationJwt
|
||||
|
||||
181
backend/tests/integration/test_translate_clickhouse.py
Normal file
181
backend/tests/integration/test_translate_clickhouse.py
Normal file
@@ -0,0 +1,181 @@
|
||||
# #region TestTranslateClickHouseIntegration [C:3] [TYPE Module] [SEMANTICS test,translate,clickhouse,integration]
|
||||
# @BRIEF Integration tests for ClickHouse insert path using testcontainers.
|
||||
# @RELATION BINDS_TO -> [_batch_insert_module]
|
||||
# @RELATION BINDS_TO -> [SQLGenerator]
|
||||
#
|
||||
# @TEST_CONTRACT ClickHouseInsert ->
|
||||
# SQL generation: clickhouse dialect -> valid INSERT SQL with backtick quoting
|
||||
# Rows with timestamps: timestamp values encode to YYYY-MM-DD format
|
||||
# Null handling: None values encode to NULL
|
||||
# UPSERT: clickhouse supports INSERT with key_cols (uses INSERT ... VALUES)
|
||||
#
|
||||
# @REQUIRES Docker daemon running (testcontainers starts clickhouse server)
|
||||
import pytest
|
||||
import uuid
|
||||
|
||||
from testcontainers.clickhouse import ClickHouseContainer
|
||||
|
||||
from src.plugins.translate.sql_generator import SQLGenerator, _encode_sql_value
|
||||
|
||||
|
||||
# #region clickhouse_container [C:2] [TYPE Fixture]
|
||||
@pytest.fixture(scope="module")
|
||||
def clickhouse_container():
|
||||
"""Start a ClickHouse server for the test module."""
|
||||
with ClickHouseContainer("clickhouse/clickhouse-server:24.3") as ch:
|
||||
yield ch
|
||||
# #endregion clickhouse_container
|
||||
|
||||
|
||||
# #region clickhouse_client [C:2] [TYPE Fixture]
|
||||
@pytest.fixture(scope="module")
|
||||
def clickhouse_client(clickhouse_container):
|
||||
"""clickhouse-connect client connected to the ClickHouse container."""
|
||||
import clickhouse_connect
|
||||
|
||||
client = clickhouse_connect.get_client(
|
||||
host=clickhouse_container.get_container_host_ip(),
|
||||
port=int(clickhouse_container.get_exposed_port(8123)),
|
||||
username="test",
|
||||
password="test",
|
||||
)
|
||||
yield client
|
||||
client.close()
|
||||
# #endregion clickhouse_client
|
||||
|
||||
|
||||
# #region TestClickHouseSQLGeneration [C:3] [TYPE Class]
|
||||
class TestClickHouseSQLGeneration:
|
||||
"""Verify SQL generation produces valid ClickHouse SQL."""
|
||||
|
||||
def test_generate_insert(self):
|
||||
sql, count = SQLGenerator.generate(
|
||||
dialect="clickhouse",
|
||||
target_schema="",
|
||||
target_table="test_table",
|
||||
columns=["id", "name", "value"],
|
||||
rows=[
|
||||
{"id": 1, "name": "Alice", "value": "Hello"},
|
||||
{"id": 2, "name": "Bob", "value": "World"},
|
||||
],
|
||||
key_columns=["id"],
|
||||
upsert_strategy="INSERT",
|
||||
)
|
||||
assert sql is not None
|
||||
assert "INSERT INTO" in sql
|
||||
assert "test_table" in sql
|
||||
assert "Alice" in sql
|
||||
assert "Bob" in sql
|
||||
assert count == 2
|
||||
|
||||
def test_generate_with_backtick_quoting(self):
|
||||
sql, _ = SQLGenerator.generate(
|
||||
dialect="clickhouse",
|
||||
target_schema="",
|
||||
target_table="test_table",
|
||||
columns=["id", "translated-text"],
|
||||
rows=[{"id": 1, "translated-text": "hello"}],
|
||||
key_columns=["id"],
|
||||
upsert_strategy="INSERT",
|
||||
)
|
||||
assert "`translated-text`" in sql
|
||||
|
||||
def test_encode_timestamp(self):
|
||||
result = _encode_sql_value("1726358400000.0", dialect="clickhouse")
|
||||
assert "2024" in result # should produce a date string
|
||||
|
||||
def test_encode_string(self):
|
||||
result = _encode_sql_value("Hello World", dialect="clickhouse")
|
||||
assert result == "'Hello World'"
|
||||
|
||||
def test_encode_none(self):
|
||||
result = _encode_sql_value(None, dialect="clickhouse")
|
||||
assert result == "NULL"
|
||||
|
||||
def test_encode_int(self):
|
||||
result = _encode_sql_value(42, dialect="clickhouse")
|
||||
assert result == "42"
|
||||
|
||||
def test_encode_bool(self):
|
||||
result = _encode_sql_value(True, dialect="clickhouse")
|
||||
assert result == "TRUE"
|
||||
# #endregion TestClickHouseSQLGeneration
|
||||
|
||||
|
||||
# #region TestClickHouseRealExecution [C:3] [TYPE Class]
|
||||
class TestClickHouseRealExecution:
|
||||
"""Execute generated SQL against a real ClickHouse container."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_table(self, clickhouse_client):
|
||||
"""Create test table before each test."""
|
||||
clickhouse_client.command("""
|
||||
CREATE TABLE IF NOT EXISTS test_translations (
|
||||
id UInt32,
|
||||
source_text String,
|
||||
translated_text String,
|
||||
lang_code String,
|
||||
is_original UInt8
|
||||
) ENGINE = MergeTree()
|
||||
ORDER BY id
|
||||
""")
|
||||
yield
|
||||
clickhouse_client.command("DROP TABLE IF EXISTS test_translations")
|
||||
|
||||
def test_insert_and_select(self, clickhouse_client):
|
||||
"""Insert rows and verify they can be selected."""
|
||||
sql, count = SQLGenerator.generate(
|
||||
dialect="clickhouse",
|
||||
target_schema="",
|
||||
target_table="test_translations",
|
||||
columns=["id", "source_text", "translated_text", "lang_code", "is_original"],
|
||||
rows=[
|
||||
{"id": 1, "source_text": "Hello", "translated_text": "Привет", "lang_code": "ru", "is_original": 1},
|
||||
{"id": 1, "source_text": "Hello", "translated_text": "Привет", "lang_code": "ru", "is_original": 0},
|
||||
],
|
||||
key_columns=["id"],
|
||||
upsert_strategy="INSERT",
|
||||
)
|
||||
assert sql is not None
|
||||
assert count == 2
|
||||
|
||||
clickhouse_client.command(sql)
|
||||
|
||||
result = clickhouse_client.query("SELECT count(*) FROM test_translations")
|
||||
assert result.result_rows[0][0] == 2
|
||||
|
||||
def test_multiple_batches(self, clickhouse_client):
|
||||
"""Insert in multiple chunks."""
|
||||
rows = [{"id": i, "source_text": f"text_{i}", "translated_text": f"trans_{i}", "lang_code": "ru", "is_original": 1} for i in range(10)]
|
||||
|
||||
statements = SQLGenerator.generate_batch(
|
||||
dialect="clickhouse",
|
||||
target_schema="",
|
||||
target_table="test_translations",
|
||||
columns=["id", "source_text", "translated_text", "lang_code", "is_original"],
|
||||
rows=rows,
|
||||
key_columns=["id"],
|
||||
upsert_strategy="INSERT",
|
||||
max_rows_per_statement=3,
|
||||
)
|
||||
|
||||
assert len(statements) == 4
|
||||
for sql, chunk_count in statements:
|
||||
clickhouse_client.command(sql)
|
||||
|
||||
result = clickhouse_client.query("SELECT count(*) FROM test_translations")
|
||||
assert result.result_rows[0][0] == 10
|
||||
|
||||
def test_drop_and_reinsert(self, clickhouse_client):
|
||||
"""Drop rows to verify idempotency."""
|
||||
clickhouse_client.command("""
|
||||
INSERT INTO test_translations (id, source_text, translated_text, lang_code, is_original)
|
||||
VALUES (1, 'Hello', 'Привет', 'ru', 1)
|
||||
""")
|
||||
|
||||
clickhouse_client.command("TRUNCATE TABLE test_translations")
|
||||
|
||||
result = clickhouse_client.query("SELECT count(*) FROM test_translations")
|
||||
assert result.result_rows[0][0] == 0
|
||||
# #endregion TestClickHouseRealExecution
|
||||
# #endregion TestTranslateClickHouseIntegration
|
||||
127
backend/tests/integration/test_translate_corrections.py
Normal file
127
backend/tests/integration/test_translate_corrections.py
Normal file
@@ -0,0 +1,127 @@
|
||||
# #region TestTranslateCorrectionsIntegration [C:3] [TYPE Module] [SEMANTICS test,translate,corrections,integration]
|
||||
# @BRIEF Integration tests for InlineCorrectionService and BulkFindReplaceService with real PostgreSQL.
|
||||
# @RELATION BINDS_TO -> [InlineCorrectionService]
|
||||
# @RELATION BINDS_TO -> [BulkFindReplaceService]
|
||||
#
|
||||
# @TEST_CONTRACT InlineCorrectionService ->
|
||||
# apply_inline_edit: db + record + language + text -> language_updated
|
||||
#
|
||||
# @TEST_CONTRACT BulkFindReplaceService ->
|
||||
# apply: db + run_id + pattern + replacement -> rows_affected in result dict
|
||||
# preview: db + run_id + pattern -> list of matching items
|
||||
#
|
||||
# @TEST_EDGE: non_existent_record -> ValueError on inline edit
|
||||
# @TEST_EDGE: non_existent_language -> ValueError
|
||||
# @TEST_EDGE: empty_replace_run -> rows_affected=0, no error
|
||||
import pytest
|
||||
import uuid
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.models.translate import (
|
||||
TranslationJob,
|
||||
TranslationRun,
|
||||
TranslationBatch,
|
||||
TranslationRecord,
|
||||
TranslationLanguage,
|
||||
)
|
||||
from src.plugins.translate.service_inline_correction import InlineCorrectionService
|
||||
|
||||
|
||||
# #region TestInlineCorrectionIntegration [C:3] [TYPE Class]
|
||||
class TestInlineCorrectionIntegration:
|
||||
"""Verify InlineCorrectionService operations with real PostgreSQL."""
|
||||
|
||||
def _create_full_run_tree(self, db: Session, lang_status: str = "pending") -> tuple:
|
||||
job = TranslationJob(
|
||||
name="Inline Test Job", status="ACTIVE",
|
||||
source_dialect="postgresql", target_dialect="clickhouse",
|
||||
translation_column="name", target_column="name",
|
||||
target_languages=["ru", "de"],
|
||||
created_by="test_user",
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
|
||||
run = TranslationRun(
|
||||
job_id=job.id, status="COMPLETED", trigger_type="manual",
|
||||
)
|
||||
db.add(run)
|
||||
db.commit()
|
||||
db.refresh(run)
|
||||
|
||||
batch = TranslationBatch(
|
||||
run_id=run.id, batch_index=0, status="COMPLETED",
|
||||
total_records=1, successful_records=1,
|
||||
)
|
||||
db.add(batch)
|
||||
db.commit()
|
||||
db.refresh(batch)
|
||||
|
||||
record = TranslationRecord(
|
||||
batch_id=batch.id, run_id=run.id,
|
||||
source_sql="Hello World",
|
||||
target_sql="Hola Mundo",
|
||||
status="SUCCESS",
|
||||
source_hash=uuid.uuid4().hex,
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
db.refresh(record)
|
||||
|
||||
lang = TranslationLanguage(
|
||||
record_id=record.id,
|
||||
language_code="ru",
|
||||
source_language_detected="en",
|
||||
translated_value="Привет мир",
|
||||
final_value="Привет мир",
|
||||
status=lang_status,
|
||||
)
|
||||
db.add(lang)
|
||||
db.commit()
|
||||
db.refresh(lang)
|
||||
|
||||
return job, run, batch, record, lang
|
||||
|
||||
# #region test_apply_inline_edit [C:2] [TYPE Function]
|
||||
def test_apply_inline_edit(self, db_session: Session):
|
||||
_, _, _, record, lang = self._create_full_run_tree(db_session)
|
||||
|
||||
result = InlineCorrectionService.apply_inline_edit(
|
||||
db=db_session, run_id=record.run_id,
|
||||
record_id=record.id, language_code="ru",
|
||||
final_value="Здравствуй мир",
|
||||
)
|
||||
|
||||
assert result["language_code"] == "ru"
|
||||
assert result["final_value"] == "Здравствуй мир"
|
||||
|
||||
db_session.refresh(lang)
|
||||
assert lang.final_value == "Здравствуй мир"
|
||||
assert lang.user_edit == "Здравствуй мир"
|
||||
# #endregion test_apply_inline_edit
|
||||
|
||||
# #region test_apply_inline_edit_non_existent_record [C:2] [TYPE Function]
|
||||
def test_apply_inline_edit_non_existent_record(self, db_session: Session):
|
||||
with pytest.raises((ValueError, KeyError)):
|
||||
InlineCorrectionService.apply_inline_edit(
|
||||
db=db_session, run_id=str(uuid.uuid4()),
|
||||
record_id=str(uuid.uuid4()), language_code="ru",
|
||||
final_value="test",
|
||||
)
|
||||
# #endregion test_apply_inline_edit_non_existent_record
|
||||
|
||||
# #region test_apply_inline_edit_non_existent_language [C:2] [TYPE Function]
|
||||
def test_apply_inline_edit_non_existent_language(self, db_session: Session):
|
||||
_, _, _, record, _ = self._create_full_run_tree(db_session)
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
InlineCorrectionService.apply_inline_edit(
|
||||
db=db_session, run_id=record.run_id,
|
||||
record_id=record.id, language_code="nonexistent",
|
||||
final_value="test",
|
||||
)
|
||||
# #endregion test_apply_inline_edit_non_existent_language
|
||||
# #endregion TestInlineCorrectionIntegration
|
||||
# #endregion TestTranslateCorrectionsIntegration
|
||||
235
backend/tests/integration/test_translate_schedules.py
Normal file
235
backend/tests/integration/test_translate_schedules.py
Normal file
@@ -0,0 +1,235 @@
|
||||
# #region TestTranslateSchedulesIntegration [C:3] [TYPE Module] [SEMANTICS test,translate,schedule,integration]
|
||||
# @BRIEF Integration tests for TranslationSchedule CRUD with real PostgreSQL via Testcontainers.
|
||||
# @RELATION BINDS_TO -> [TranslationScheduler]
|
||||
#
|
||||
# @TEST_CONTRACT TranslationScheduler ->
|
||||
# create_schedule: job_id + cron + timezone -> TranslationSchedule persisted
|
||||
# get_schedule: job_id -> TranslationSchedule or ValueError
|
||||
# update_schedule: job_id + fields -> updated schedule
|
||||
# delete_schedule: job_id -> None (schedule removed)
|
||||
# set_schedule_active: job_id + is_active -> schedule toggled
|
||||
# get_next_executions: cron + tz + n -> list of ISO datetime strings
|
||||
# list_active_schedules: db -> list[TranslationSchedule]
|
||||
#
|
||||
# @TEST_EDGE: non_existent_job -> ValueError on create
|
||||
# @TEST_EDGE: get_non_existent -> ValueError with "No schedule found"
|
||||
# @TEST_EDGE: delete_non_existent -> ValueError
|
||||
# @TEST_EDGE: update_non_existent -> ValueError
|
||||
import pytest
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.models.translate import TranslationJob, TranslationSchedule
|
||||
from src.plugins.translate.scheduler import TranslationScheduler
|
||||
|
||||
|
||||
# #region TestScheduleCRUD [C:3] [TYPE Class]
|
||||
# @BRIEF Integration tests for schedule CRUD operations with real PostgreSQL.
|
||||
class TestScheduleCRUD:
|
||||
"""Verify TranslationScheduler CRUD with real PostgreSQL."""
|
||||
|
||||
def _create_job(self, db: Session, status: str = "DRAFT") -> TranslationJob:
|
||||
job = TranslationJob(
|
||||
name="Test Job",
|
||||
status=status,
|
||||
source_dialect="postgresql",
|
||||
target_dialect="clickhouse",
|
||||
created_by="test_user",
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
return job
|
||||
|
||||
# #region test_create_schedule [C:2] [TYPE Function]
|
||||
def test_create_schedule(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
|
||||
sched = scheduler.create_schedule(
|
||||
job_id=job.id,
|
||||
cron_expression="0 2 * * *",
|
||||
timezone="Europe/Moscow",
|
||||
execution_mode="new_key_only",
|
||||
)
|
||||
|
||||
assert sched.job_id == job.id
|
||||
assert sched.cron_expression == "0 2 * * *"
|
||||
assert sched.timezone == "Europe/Moscow"
|
||||
assert sched.execution_mode == "new_key_only"
|
||||
assert sched.is_active is True
|
||||
# #endregion test_create_schedule
|
||||
|
||||
# #region test_create_schedule_full_mode_default [C:2] [TYPE Function]
|
||||
def test_create_schedule_full_mode_default(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
sched = scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
assert sched.execution_mode == "full"
|
||||
# #endregion test_create_schedule_full_mode_default
|
||||
|
||||
# #region test_get_schedule [C:2] [TYPE Function]
|
||||
def test_get_schedule(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
db_session.commit()
|
||||
|
||||
result = scheduler.get_schedule(job.id)
|
||||
assert result is not None
|
||||
assert result.cron_expression == "0 2 * * *"
|
||||
# #endregion test_get_schedule
|
||||
|
||||
# #region test_get_schedule_no_schedule [C:2] [TYPE Function]
|
||||
def test_get_schedule_no_schedule(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
with pytest.raises(ValueError, match="No schedule found"):
|
||||
scheduler.get_schedule(job.id)
|
||||
# #endregion test_get_schedule_no_schedule
|
||||
|
||||
# #region test_update_schedule_all_fields [C:2] [TYPE Function]
|
||||
def test_update_schedule_all_fields(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
db_session.commit()
|
||||
|
||||
updated = scheduler.update_schedule(
|
||||
job_id=job.id, cron_expression="30 6 * * 1",
|
||||
timezone_str="Asia/Tokyo", is_active=False,
|
||||
execution_mode="full",
|
||||
)
|
||||
assert updated.cron_expression == "30 6 * * 1"
|
||||
assert updated.timezone == "Asia/Tokyo"
|
||||
assert updated.is_active is False
|
||||
assert updated.execution_mode == "full"
|
||||
# #endregion test_update_schedule_all_fields
|
||||
|
||||
# #region test_update_schedule_partial [C:2] [TYPE Function]
|
||||
def test_update_schedule_partial(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *", timezone="Europe/Moscow")
|
||||
db_session.commit()
|
||||
|
||||
updated = scheduler.update_schedule(job_id=job.id, cron_expression="0 4 * * *")
|
||||
assert updated.cron_expression == "0 4 * * *"
|
||||
assert updated.timezone == "Europe/Moscow"
|
||||
# #endregion test_update_schedule_partial
|
||||
|
||||
# #region test_delete_schedule [C:2] [TYPE Function]
|
||||
def test_delete_schedule(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
db_session.commit()
|
||||
|
||||
scheduler.delete_schedule(job.id)
|
||||
with pytest.raises(ValueError, match="No schedule found"):
|
||||
scheduler.get_schedule(job.id)
|
||||
# #endregion test_delete_schedule
|
||||
|
||||
# #region test_set_schedule_active [C:2] [TYPE Function]
|
||||
def test_set_schedule_active(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *", is_active=False)
|
||||
db_session.commit()
|
||||
|
||||
sched = scheduler.set_schedule_active(job.id, True)
|
||||
assert sched.is_active is True
|
||||
|
||||
sched = scheduler.set_schedule_active(job.id, False)
|
||||
assert sched.is_active is False
|
||||
# #endregion test_set_schedule_active
|
||||
|
||||
# #region test_set_schedule_active_twice [C:2] [TYPE Function]
|
||||
def test_set_schedule_active_twice(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
db_session.commit()
|
||||
|
||||
sched = scheduler.set_schedule_active(job.id, True)
|
||||
assert sched.is_active is True
|
||||
sched = scheduler.set_schedule_active(job.id, True)
|
||||
assert sched.is_active is True
|
||||
# #endregion test_set_schedule_active_twice
|
||||
|
||||
# #region test_create_schedule_non_existent_job [C:2] [TYPE Function]
|
||||
def test_create_schedule_non_existent_job(self, db_session: Session, mock_config_manager):
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
scheduler.create_schedule(job_id="non-existent-id", cron_expression="0 2 * * *")
|
||||
# #endregion test_create_schedule_non_existent_job
|
||||
|
||||
# #region test_delete_schedule_non_existent [C:2] [TYPE Function]
|
||||
def test_delete_schedule_non_existent(self, db_session: Session, mock_config_manager):
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
with pytest.raises(ValueError, match="No schedule found"):
|
||||
scheduler.delete_schedule(job_id="non-existent-id")
|
||||
# #endregion test_delete_schedule_non_existent
|
||||
|
||||
# #region test_update_schedule_no_schedule [C:2] [TYPE Function]
|
||||
def test_update_schedule_no_schedule(self, db_session: Session, mock_config_manager):
|
||||
job = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
with pytest.raises(ValueError, match="No schedule found"):
|
||||
scheduler.update_schedule(job_id=job.id, cron_expression="0 2 * * *")
|
||||
# #endregion test_update_schedule_no_schedule
|
||||
|
||||
# #region test_set_active_non_existent [C:2] [TYPE Function]
|
||||
def test_set_active_non_existent(self, db_session: Session, mock_config_manager):
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
with pytest.raises(ValueError, match="No schedule found"):
|
||||
scheduler.set_schedule_active(job_id="non-existent", is_active=True)
|
||||
# #endregion test_set_active_non_existent
|
||||
|
||||
# #region test_list_active_schedules [C:2] [TYPE Function]
|
||||
def test_list_active_schedules(self, db_session: Session, mock_config_manager):
|
||||
job1 = self._create_job(db_session)
|
||||
job2 = self._create_job(db_session)
|
||||
scheduler = TranslationScheduler(db_session, mock_config_manager)
|
||||
scheduler.create_schedule(job_id=job1.id, cron_expression="0 2 * * *", is_active=True)
|
||||
scheduler.create_schedule(job_id=job2.id, cron_expression="0 4 * * *", is_active=False)
|
||||
db_session.commit()
|
||||
|
||||
active = TranslationScheduler.list_active_schedules(db_session)
|
||||
assert len(active) == 1
|
||||
assert active[0].job_id == job1.id
|
||||
# #endregion test_list_active_schedules
|
||||
|
||||
# #region test_get_next_executions_static [C:2] [TYPE Function]
|
||||
def test_get_next_executions_static(self):
|
||||
executions = TranslationScheduler.get_next_executions(
|
||||
cron_expression="0 2 * * *",
|
||||
timezone_str="UTC",
|
||||
n=3,
|
||||
)
|
||||
assert len(executions) == 3
|
||||
for dt_str in executions:
|
||||
assert "T02:00:00" in dt_str
|
||||
# #endregion test_get_next_executions_static
|
||||
|
||||
# #region test_get_next_executions_invalid_cron [C:2] [TYPE Function]
|
||||
def test_get_next_executions_invalid_cron(self):
|
||||
executions = TranslationScheduler.get_next_executions(
|
||||
cron_expression="invalid-cron",
|
||||
timezone_str="UTC",
|
||||
n=3,
|
||||
)
|
||||
assert executions == []
|
||||
# #endregion test_get_next_executions_invalid_cron
|
||||
|
||||
# #region test_get_next_executions_different_timezone [C:2] [TYPE Function]
|
||||
def test_get_next_executions_different_timezone(self):
|
||||
executions = TranslationScheduler.get_next_executions(
|
||||
cron_expression="0 2 * * *",
|
||||
timezone_str="Asia/Tokyo",
|
||||
n=2,
|
||||
)
|
||||
assert len(executions) == 2
|
||||
# #endregion test_get_next_executions_different_timezone
|
||||
# #endregion TestScheduleCRUD
|
||||
# #endregion TestTranslateSchedulesIntegration
|
||||
245
backend/tests/integration/test_translate_status_fk.py
Normal file
245
backend/tests/integration/test_translate_status_fk.py
Normal file
@@ -0,0 +1,245 @@
|
||||
# #region TestTranslateStatusAndFK [C:3] [TYPE Module] [SEMANTICS test,translate,integration,status,fk,constraints]
|
||||
# @BRIEF Integration tests for job status transitions, run state machine, FK constraint enforcement
|
||||
# with real PostgreSQL via Testcontainers.
|
||||
# @RELATION BINDS_TO -> [TranslateJobService]
|
||||
# @RELATION BINDS_TO -> [TranslationEventLog]
|
||||
#
|
||||
# @TEST_CONTRACT JobStatusTransitions ->
|
||||
# DRAFT job: create -> delete works
|
||||
# non-existent: get/update/delete raises ValueError
|
||||
# long name: persists
|
||||
#
|
||||
# @TEST_CONTRACT RunStateMachine ->
|
||||
# PENDING -> COMPLETED: via cancel_run
|
||||
# COMPLETED -> cancel raises ValueError
|
||||
# CANCELLED -> cancel raises ValueError
|
||||
#
|
||||
# @TEST_EDGE: delete_non_existent_job -> ValueError
|
||||
# @TEST_EDGE: cancel_non_existent_run -> ValueError
|
||||
# @TEST_EDGE: cancel_cancelled_run -> ValueError
|
||||
import pytest
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.models.translate import (
|
||||
TranslationJob,
|
||||
TranslationRun,
|
||||
TranslationBatch,
|
||||
TranslationRecord,
|
||||
TranslationLanguage,
|
||||
TranslationEvent,
|
||||
)
|
||||
from src.plugins.translate.service import TranslateJobService
|
||||
from src.schemas.translate import TranslateJobCreate
|
||||
|
||||
|
||||
# #region TestJobStatusTransitions [C:3] [TYPE Class]
|
||||
@pytest.mark.asyncio
|
||||
class TestJobStatusTransitions:
|
||||
"""Verify job status lifecycle transitions with real PostgreSQL."""
|
||||
|
||||
async def test_create_draft_job(self, db_session: Session, mock_config_manager):
|
||||
service = TranslateJobService(db_session, mock_config_manager, "test_user")
|
||||
payload = TranslateJobCreate(
|
||||
name="Draft Test Job",
|
||||
source_dialect="postgresql",
|
||||
target_dialect="clickhouse",
|
||||
translation_column="name",
|
||||
target_languages=["ru"],
|
||||
)
|
||||
job = await service.create_job(payload)
|
||||
assert job.status == "DRAFT"
|
||||
|
||||
async def test_delete_job(self, db_session: Session, mock_config_manager):
|
||||
service = TranslateJobService(db_session, mock_config_manager, "test_user")
|
||||
payload = TranslateJobCreate(
|
||||
name="Delete Test", source_dialect="postgresql",
|
||||
target_dialect="clickhouse", translation_column="name",
|
||||
target_languages=["ru"],
|
||||
)
|
||||
job = await service.create_job(payload)
|
||||
service.delete_job(job.id)
|
||||
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
service.get_job(job.id)
|
||||
|
||||
async def test_delete_non_existent_job(self, db_session: Session, mock_config_manager):
|
||||
service = TranslateJobService(db_session, mock_config_manager, "test_user")
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
service.delete_job("non-existent-id")
|
||||
|
||||
async def test_get_non_existent_job(self, db_session: Session, mock_config_manager):
|
||||
service = TranslateJobService(db_session, mock_config_manager, "test_user")
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
service.get_job("non-existent-id")
|
||||
|
||||
async def test_create_job_max_name_length(self, db_session: Session, mock_config_manager):
|
||||
service = TranslateJobService(db_session, mock_config_manager, "test_user")
|
||||
long_name = "A" * 500
|
||||
payload = TranslateJobCreate(
|
||||
name=long_name,
|
||||
source_dialect="postgresql",
|
||||
target_dialect="clickhouse",
|
||||
translation_column="name",
|
||||
target_languages=["ru"],
|
||||
)
|
||||
job = await service.create_job(payload)
|
||||
assert len(job.name) == 500
|
||||
# #endregion TestJobStatusTransitions
|
||||
|
||||
|
||||
# #region TestFKConstraintEnforcement [C:3] [TYPE Class]
|
||||
class TestFKConstraintEnforcement:
|
||||
"""Verify PostgreSQL FK constraints catch orphaned data."""
|
||||
|
||||
def _assert_fk_violation(self, db: Session, obj):
|
||||
"""Helper: try adding an orphaned object and expect FK violation."""
|
||||
db.add(obj)
|
||||
try:
|
||||
db.flush()
|
||||
db.rollback()
|
||||
pytest.fail("Expected FK violation was not raised")
|
||||
except Exception:
|
||||
db.rollback()
|
||||
|
||||
def test_orphan_run_rejected(self, db_session: Session):
|
||||
run = TranslationRun(job_id=str(uuid.uuid4()), status="PENDING", trigger_type="manual")
|
||||
self._assert_fk_violation(db_session, run)
|
||||
|
||||
def test_orphan_batch_rejected(self, db_session: Session):
|
||||
batch = TranslationBatch(run_id=str(uuid.uuid4()), batch_index=0, status="PENDING")
|
||||
self._assert_fk_violation(db_session, batch)
|
||||
|
||||
def test_valid_run_with_job_succeeds(self, db_session: Session):
|
||||
job = TranslationJob(
|
||||
name="FK Test Job", status="ACTIVE",
|
||||
source_dialect="postgresql", target_dialect="clickhouse",
|
||||
created_by="test_user",
|
||||
)
|
||||
db_session.add(job)
|
||||
db_session.flush()
|
||||
db_session.refresh(job)
|
||||
|
||||
run = TranslationRun(job_id=job.id, status="PENDING", trigger_type="manual")
|
||||
db_session.add(run)
|
||||
db_session.flush()
|
||||
assert run.job_id == job.id
|
||||
# #endregion TestFKConstraintEnforcement
|
||||
|
||||
|
||||
# #region TestRunRecovery [C:3] [TYPE Class]
|
||||
class TestRunRecovery:
|
||||
"""Verify run cancel behavior with real PostgreSQL."""
|
||||
|
||||
def _create_run(self, db: Session, status: str = "PENDING") -> TranslationRun:
|
||||
job = TranslationJob(
|
||||
name="Cancel Test", status="ACTIVE",
|
||||
source_dialect="postgresql", target_dialect="clickhouse",
|
||||
created_by="test_user",
|
||||
)
|
||||
db.add(job)
|
||||
db.flush()
|
||||
db.refresh(job)
|
||||
|
||||
run = TranslationRun(job_id=job.id, status=status, trigger_type="manual")
|
||||
db.add(run)
|
||||
db.flush()
|
||||
db.refresh(run)
|
||||
return run
|
||||
|
||||
def test_cancel_pending_run(self, db_session: Session):
|
||||
run = self._create_run(db_session, "PENDING")
|
||||
from src.plugins.translate.orchestrator_cancel import cancel_run
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
result = cancel_run(db_session, event_log, "test_user", run.id)
|
||||
assert result.status == "CANCELLED"
|
||||
assert result.completed_at is not None
|
||||
|
||||
def test_cancel_failed_run(self, db_session: Session):
|
||||
run = self._create_run(db_session, "FAILED")
|
||||
from src.plugins.translate.orchestrator_cancel import cancel_run
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
with pytest.raises(ValueError, match="Cannot cancel"):
|
||||
cancel_run(db_session, event_log, "test_user", run.id)
|
||||
|
||||
def test_cancel_completed_run_raises(self, db_session: Session):
|
||||
run = self._create_run(db_session, "COMPLETED")
|
||||
from src.plugins.translate.orchestrator_cancel import cancel_run
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
with pytest.raises(ValueError, match="Cannot cancel"):
|
||||
cancel_run(db_session, event_log, "test_user", run.id)
|
||||
|
||||
def test_cancel_cancelled_run_raises(self, db_session: Session):
|
||||
run = self._create_run(db_session, "CANCELLED")
|
||||
from src.plugins.translate.orchestrator_cancel import cancel_run
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
with pytest.raises(ValueError, match="Cannot cancel"):
|
||||
cancel_run(db_session, event_log, "test_user", run.id)
|
||||
|
||||
def test_cancel_non_existent_run(self, db_session: Session):
|
||||
from src.plugins.translate.orchestrator_cancel import cancel_run
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
with pytest.raises((ValueError, KeyError)):
|
||||
cancel_run(db_session, event_log, "test_user", str(uuid.uuid4()))
|
||||
# #endregion TestRunRecovery
|
||||
|
||||
|
||||
# #region TestEventLogStateMachine [C:3] [TYPE Class]
|
||||
class TestEventLogStateMachine:
|
||||
"""Verify event log state machine rules with real PostgreSQL."""
|
||||
|
||||
def _create_job_and_run(self, db: Session) -> tuple:
|
||||
job = TranslationJob(
|
||||
name="Event Log Job", status="ACTIVE",
|
||||
source_dialect="postgresql", target_dialect="clickhouse",
|
||||
created_by="test_user",
|
||||
)
|
||||
db.add(job)
|
||||
db.flush()
|
||||
db.refresh(job)
|
||||
|
||||
run = TranslationRun(job_id=job.id, status="RUNNING", trigger_type="manual")
|
||||
db.add(run)
|
||||
db.flush()
|
||||
db.refresh(run)
|
||||
return job, run
|
||||
|
||||
def test_event_log_persistence(self, db_session: Session):
|
||||
job, run = self._create_job_and_run(db_session)
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
event_log.log_event(job.id, "RUN_STARTED", {"strategy": "incremental"}, run.id)
|
||||
event_log.log_event(job.id, "BATCH_STARTED", {"batch_index": 0}, run.id)
|
||||
event_log.log_event(job.id, "BATCH_COMPLETED", {"batch_index": 0, "records": 10}, run.id)
|
||||
event_log.log_event(job.id, "RUN_COMPLETED", {"total_records": 10}, run.id)
|
||||
|
||||
events = (
|
||||
db_session.query(TranslationEvent)
|
||||
.filter(TranslationEvent.run_id == run.id)
|
||||
.order_by(TranslationEvent.created_at)
|
||||
.all()
|
||||
)
|
||||
assert len(events) == 4
|
||||
|
||||
def test_invalid_event_type_raises(self, db_session: Session):
|
||||
job, run = self._create_job_and_run(db_session)
|
||||
from src.plugins.translate.events import TranslationEventLog
|
||||
|
||||
event_log = TranslationEventLog(db_session)
|
||||
with pytest.raises(ValueError, match="Invalid event_type"):
|
||||
event_log.log_event(job.id, run.id, "INVALID_EVENT", {})
|
||||
# #endregion TestEventLogStateMachine
|
||||
# #endregion TestTranslateStatusAndFK
|
||||
Reference in New Issue
Block a user