32 lines
994 B
Python
32 lines
994 B
Python
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from alembic import command
|
||
|
|
from alembic.config import Config
|
||
|
|
from sqlalchemy import Engine, create_engine, inspect
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.integration
|
||
|
|
def test_postgres_migration_creates_market_data_contract() -> None:
|
||
|
|
database_url = os.getenv("ZHIXING_TEST_DATABASE_URL")
|
||
|
|
if not database_url:
|
||
|
|
pytest.skip("set ZHIXING_TEST_DATABASE_URL to run PostgreSQL integration tests")
|
||
|
|
|
||
|
|
server_root = Path(__file__).parents[2]
|
||
|
|
config = Config(str(server_root / "alembic.ini"))
|
||
|
|
config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%"))
|
||
|
|
engine: Engine = create_engine(database_url)
|
||
|
|
command.upgrade(config, "head")
|
||
|
|
try:
|
||
|
|
tables = set(inspect(engine).get_table_names())
|
||
|
|
assert {
|
||
|
|
"market_stock",
|
||
|
|
"market_daily_bar",
|
||
|
|
"market_daily_basic",
|
||
|
|
"market_sync_batch",
|
||
|
|
"market_sync_item",
|
||
|
|
} <= tables
|
||
|
|
finally:
|
||
|
|
engine.dispose()
|