30 lines
947 B
Python
30 lines
947 B
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from app.modules.data_process.schema_cli import _target_label
|
||
|
|
|
||
|
|
|
||
|
|
def test_runtime_migration_fails_fast_on_incompatible_schema() -> None:
|
||
|
|
sql_path = (
|
||
|
|
Path(__file__).resolve().parents[1]
|
||
|
|
/ "app"
|
||
|
|
/ "db"
|
||
|
|
/ "sql"
|
||
|
|
/ "002_data_process.sql"
|
||
|
|
)
|
||
|
|
sql = sql_path.read_text(encoding="utf-8")
|
||
|
|
|
||
|
|
assert "requires 001_platform_runtime.sql first" in sql
|
||
|
|
assert "supports only the current TEXT runtime schema" in sql
|
||
|
|
assert "generation_run_id" in sql
|
||
|
|
assert "CREATE TABLE IF NOT EXISTS data_process_results" in sql
|
||
|
|
assert sql.count("BEGIN;") == 1
|
||
|
|
assert sql.rstrip().endswith("COMMIT;")
|
||
|
|
|
||
|
|
|
||
|
|
def test_schema_cli_target_label_never_contains_credentials() -> None:
|
||
|
|
label = _target_label("postgresql://secret-user:secret-password@db.example:5433/yg_ft")
|
||
|
|
assert label == "db.example:5433/yg_ft"
|
||
|
|
assert "secret" not in label
|