JiuContinent
eb8267ed18
feat(feishu): 添加飞书入站事件inbox和混合数据库协调功能 - 实现飞书入站事件持久化inbox机制，支持状态管理、租约锁定和重试退避 - 添加混合数据库基线协调工具，确保平台PostgreSQL结构安全对齐 - 增加运行组件心跳检测和readiness就绪检查机制 - 实现app_ticket事件的安全轮换和验证处理 - 添加生产环境运行编排和fail-closed安全机制 - 支持webhook快速确认和长连接独立进程处理 - 完善个人数据擦除时的待处理事件清理功能 ```
282 lines
9.8 KiB
Python
282 lines
9.8 KiB
Python
import os
|
|
import re
|
|
from collections.abc import Iterator
|
|
from uuid import uuid4
|
|
|
|
from alembic import command
|
|
import pytest
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.engine import Engine, URL, make_url
|
|
from sqlalchemy.exc import ArgumentError, DBAPIError
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.core.config import get_settings
|
|
from app.tools import reconcile_platform_schema as schema_tool
|
|
|
|
|
|
TEST_DATABASE_ENV = "TEST_POSTGRES_DATABASE_URL"
|
|
TEST_SCHEMA_PATTERN = re.compile(r"caip_test_[0-9a-f]{32}\Z")
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not os.getenv(TEST_DATABASE_ENV, "").strip(),
|
|
reason=f"{TEST_DATABASE_ENV} is not configured",
|
|
)
|
|
|
|
|
|
def _configured_postgres_url() -> URL:
|
|
raw_url = os.getenv(TEST_DATABASE_ENV, "").strip()
|
|
if not raw_url:
|
|
pytest.skip(f"{TEST_DATABASE_ENV} is not configured")
|
|
try:
|
|
url = make_url(raw_url)
|
|
except (ArgumentError, TypeError, ValueError):
|
|
pytest.fail(f"{TEST_DATABASE_ENV} is not a valid SQLAlchemy URL")
|
|
if url.get_backend_name() != "postgresql":
|
|
pytest.fail(f"{TEST_DATABASE_ENV} must use PostgreSQL")
|
|
return url
|
|
|
|
|
|
def _quoted_schema(engine: Engine, schema_name: str) -> str:
|
|
if not TEST_SCHEMA_PATTERN.fullmatch(schema_name):
|
|
raise RuntimeError("Refusing unsafe PostgreSQL test schema name")
|
|
return engine.dialect.identifier_preparer.quote_identifier(schema_name)
|
|
|
|
|
|
@pytest.fixture
|
|
def postgres_schema_engine(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> Iterator[Engine]:
|
|
base_url = _configured_postgres_url()
|
|
schema_name = f"caip_test_{uuid4().hex}"
|
|
admin_engine = create_engine(base_url, poolclass=NullPool)
|
|
isolated_engine: Engine | None = None
|
|
schema_created = False
|
|
try:
|
|
quoted_schema = _quoted_schema(admin_engine, schema_name)
|
|
with admin_engine.begin() as connection:
|
|
connection.exec_driver_sql(f"CREATE SCHEMA {quoted_schema}")
|
|
schema_created = True
|
|
|
|
isolated_url = base_url.update_query_dict(
|
|
{"options": f"-csearch_path={schema_name}"}
|
|
)
|
|
isolated_engine = create_engine(isolated_url, poolclass=NullPool)
|
|
monkeypatch.setenv(
|
|
"DATABASE_URL",
|
|
isolated_url.render_as_string(hide_password=False),
|
|
)
|
|
get_settings.cache_clear()
|
|
with isolated_engine.connect() as connection:
|
|
assert connection.scalar(text("SELECT current_schema()")) == schema_name
|
|
yield isolated_engine
|
|
finally:
|
|
if isolated_engine is not None:
|
|
isolated_engine.dispose()
|
|
get_settings.cache_clear()
|
|
if schema_created:
|
|
quoted_schema = _quoted_schema(admin_engine, schema_name)
|
|
with admin_engine.begin() as connection:
|
|
connection.exec_driver_sql(
|
|
f"DROP SCHEMA IF EXISTS {quoted_schema} CASCADE"
|
|
)
|
|
admin_engine.dispose()
|
|
|
|
|
|
def _upgrade_to_previous(engine: Engine) -> None:
|
|
with engine.begin() as connection:
|
|
command.upgrade(
|
|
schema_tool._alembic_config(connection),
|
|
schema_tool.PREVIOUS_REVISION,
|
|
)
|
|
|
|
|
|
def _prepare_mixed_unversioned_baseline(engine: Engine) -> None:
|
|
_upgrade_to_previous(engine)
|
|
with engine.begin() as connection:
|
|
connection.execute(
|
|
text("ALTER TABLE projects DROP COLUMN department_code")
|
|
)
|
|
connection.execute(
|
|
text("CREATE TABLE approval_requests (id INTEGER PRIMARY KEY)")
|
|
)
|
|
connection.execute(text("DELETE FROM alembic_version"))
|
|
|
|
|
|
def test_postgres_mixed_baseline_applies_to_head_without_metadata_drift(
|
|
postgres_schema_engine: Engine,
|
|
) -> None:
|
|
_prepare_mixed_unversioned_baseline(postgres_schema_engine)
|
|
before = schema_tool.audit_engine(postgres_schema_engine)
|
|
|
|
assert before.eligible is True
|
|
assert (
|
|
before.postgresql_server_version_num
|
|
and before.postgresql_server_version_num
|
|
>= schema_tool.MINIMUM_POSTGRESQL_VERSION_NUM
|
|
)
|
|
assert before.postgresql_version_supported is True
|
|
assert before.revision_rows == ()
|
|
assert schema_tool.DriftKey(
|
|
"add_column",
|
|
"projects",
|
|
"department_code",
|
|
) in before.observed_drift
|
|
assert schema_tool.DriftKey(
|
|
"remove_table",
|
|
schema_tool.APPROVAL_TABLE,
|
|
"",
|
|
) in before.observed_drift
|
|
|
|
after = schema_tool.apply_baseline(
|
|
postgres_schema_engine,
|
|
before.fingerprint,
|
|
)
|
|
|
|
assert after.revision_rows == (schema_tool.TARGET_REVISION,)
|
|
assert after.observed_drift == frozenset()
|
|
assert after.unexpected_drift == frozenset()
|
|
assert schema_tool.APPROVAL_TABLE not in inspect(
|
|
postgres_schema_engine
|
|
).get_table_names()
|
|
|
|
|
|
def test_postgres_upgrade_failure_rolls_back_ddl_and_stamp(
|
|
postgres_schema_engine: Engine,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_upgrade_to_previous(postgres_schema_engine)
|
|
with postgres_schema_engine.begin() as connection:
|
|
connection.execute(text("DELETE FROM alembic_version"))
|
|
before = schema_tool.audit_engine(postgres_schema_engine)
|
|
|
|
def fail_upgrade(config, _revision: str) -> None:
|
|
connection = config.attributes["connection"]
|
|
connection.execute(
|
|
text("CREATE TABLE injected_upgrade_artifact (id INTEGER)")
|
|
)
|
|
connection.execute(
|
|
text("ALTER TABLE projects ADD COLUMN injected_upgrade_marker TEXT")
|
|
)
|
|
raise RuntimeError("injected PostgreSQL migration failure")
|
|
|
|
monkeypatch.setattr(schema_tool.command, "upgrade", fail_upgrade)
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="injected PostgreSQL migration failure",
|
|
):
|
|
schema_tool.apply_baseline(
|
|
postgres_schema_engine,
|
|
before.fingerprint,
|
|
)
|
|
|
|
with postgres_schema_engine.connect() as connection:
|
|
after = schema_tool.audit_connection(connection)
|
|
table_names = inspect(connection).get_table_names()
|
|
project_columns = {
|
|
column["name"]
|
|
for column in inspect(connection).get_columns("projects")
|
|
}
|
|
assert after.fingerprint == before.fingerprint
|
|
assert after.revision_rows == ()
|
|
assert "injected_upgrade_artifact" not in table_names
|
|
assert "injected_upgrade_marker" not in project_columns
|
|
|
|
|
|
def test_postgres_apply_rejects_same_count_content_change(
|
|
postgres_schema_engine: Engine,
|
|
) -> None:
|
|
_upgrade_to_previous(postgres_schema_engine)
|
|
with postgres_schema_engine.begin() as connection:
|
|
connection.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO feishu_event_receipts (
|
|
event_key,
|
|
source,
|
|
event_id,
|
|
message_id,
|
|
received_at
|
|
)
|
|
VALUES (
|
|
'postgres-content-key',
|
|
'webhook',
|
|
'event-before',
|
|
'message-stable',
|
|
'2026-07-27 00:00:00'
|
|
)
|
|
"""
|
|
)
|
|
)
|
|
connection.execute(text("DELETE FROM alembic_version"))
|
|
before = schema_tool.audit_engine(postgres_schema_engine)
|
|
|
|
with postgres_schema_engine.begin() as connection:
|
|
connection.execute(
|
|
text(
|
|
"""
|
|
UPDATE feishu_event_receipts
|
|
SET event_id = 'event-after'
|
|
WHERE event_key = 'postgres-content-key'
|
|
"""
|
|
)
|
|
)
|
|
changed = schema_tool.audit_engine(postgres_schema_engine)
|
|
|
|
assert changed.impacted_table_row_counts == before.impacted_table_row_counts
|
|
assert (
|
|
changed.impacted_table_content_digests
|
|
!= before.impacted_table_content_digests
|
|
)
|
|
assert changed.fingerprint != before.fingerprint
|
|
with pytest.raises(schema_tool.ReconciliationError, match="fingerprint changed"):
|
|
schema_tool.apply_baseline(
|
|
postgres_schema_engine,
|
|
before.fingerprint,
|
|
)
|
|
with postgres_schema_engine.connect() as connection:
|
|
assert schema_tool._revision_rows(connection) == ()
|
|
|
|
|
|
def test_postgres_advisory_transaction_lock_excludes_second_connection(
|
|
postgres_schema_engine: Engine,
|
|
) -> None:
|
|
with postgres_schema_engine.begin() as connection:
|
|
connection.execute(text("CREATE TABLE projects (id INTEGER PRIMARY KEY)"))
|
|
|
|
with (
|
|
postgres_schema_engine.connect() as first,
|
|
postgres_schema_engine.connect() as second,
|
|
):
|
|
first_transaction = first.begin()
|
|
second_transaction = second.begin()
|
|
try:
|
|
assert first.scalar(text("SELECT pg_backend_pid()")) != second.scalar(
|
|
text("SELECT pg_backend_pid()")
|
|
)
|
|
schema_tool._acquire_postgres_lock(first)
|
|
schema_tool._lock_impacted_postgres_tables(first)
|
|
second_acquired = second.scalar(
|
|
text("SELECT pg_try_advisory_xact_lock(:lock_key)"),
|
|
{"lock_key": schema_tool.ADVISORY_LOCK_KEY},
|
|
)
|
|
assert second_acquired is False
|
|
second.exec_driver_sql("SET LOCAL lock_timeout = '250ms'")
|
|
with pytest.raises(DBAPIError):
|
|
second.exec_driver_sql(
|
|
"LOCK TABLE projects IN ROW EXCLUSIVE MODE NOWAIT"
|
|
)
|
|
second_transaction.rollback()
|
|
|
|
first_transaction.commit()
|
|
second_transaction = second.begin()
|
|
second_acquired_after_release = second.scalar(
|
|
text("SELECT pg_try_advisory_xact_lock(:lock_key)"),
|
|
{"lock_key": schema_tool.ADVISORY_LOCK_KEY},
|
|
)
|
|
assert second_acquired_after_release is True
|
|
finally:
|
|
if first_transaction.is_active:
|
|
first_transaction.rollback()
|
|
if second_transaction.is_active:
|
|
second_transaction.rollback()
|