143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
import asyncio
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from lead_scoring.db.models.outbox_event import OutboxEvent
|
|
from lead_scoring.db.models.scoring_run import ScoringRun
|
|
from lead_scoring.execution import recover_interrupted_runs
|
|
|
|
|
|
def get_alembic_config(database_url: str) -> Config:
|
|
root_dir = Path(__file__).resolve().parent.parent
|
|
config = Config(str(root_dir / "alembic.ini"))
|
|
config.attributes["sqlalchemy_url"] = database_url
|
|
return config
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_terminalizes_expired_run_once_with_atomic_outbox(
|
|
temp_postgres_db: str,
|
|
) -> None:
|
|
await asyncio.to_thread(command.upgrade, get_alembic_config(temp_postgres_db), "head")
|
|
engine = create_async_engine(temp_postgres_db)
|
|
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
run_id = str(uuid.uuid4())
|
|
created_at = datetime(2026, 8, 27, 12, 0, tzinfo=UTC)
|
|
|
|
async with session_factory() as session:
|
|
session.add(
|
|
ScoringRun(
|
|
id=run_id,
|
|
tenant_id="tenant-recovery",
|
|
subject_ref="lead-recovery",
|
|
scoring_config_version="1.0.0",
|
|
idempotency_key="recovery-key",
|
|
payload_hash="0" * 64,
|
|
status="running",
|
|
snapshot_data={},
|
|
confidence_threshold=0.8,
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
lease_expires_at=created_at + timedelta(seconds=30),
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
recovery_time = created_at + timedelta(seconds=31)
|
|
assert await recover_interrupted_runs(session_factory, clock=lambda: recovery_time) == 1
|
|
assert await recover_interrupted_runs(session_factory, clock=lambda: recovery_time) == 0
|
|
|
|
async with session_factory() as session:
|
|
run = await session.get(ScoringRun, run_id)
|
|
assert run is not None
|
|
assert run.status == "failed"
|
|
assert run.error_code == "worker_interrupted"
|
|
assert run.retryable is True
|
|
assert run.completed_at == recovery_time
|
|
|
|
events = list(
|
|
(await session.execute(select(OutboxEvent).where(OutboxEvent.run_id == run_id)))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assert len(events) == 1
|
|
assert events[0].payload["data"] == {
|
|
"status": "failed",
|
|
"rule_score": None,
|
|
"ai_score": None,
|
|
"final_score": None,
|
|
"effective_score": None,
|
|
"confidence": None,
|
|
"category": "cold",
|
|
"explanation": "Không thể chấm điểm: worker_interrupted",
|
|
"recommended_action": "Kiểm tra lỗi và thử chấm điểm lại.",
|
|
"scoring_mode": None,
|
|
"degraded": None,
|
|
"error_code": "worker_interrupted",
|
|
"retryable": True,
|
|
}
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_rolls_back_run_transition_when_outbox_insert_fails(
|
|
temp_postgres_db: str,
|
|
) -> None:
|
|
await asyncio.to_thread(command.upgrade, get_alembic_config(temp_postgres_db), "head")
|
|
engine = create_async_engine(temp_postgres_db)
|
|
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
run_id = str(uuid.uuid4())
|
|
created_at = datetime(2026, 8, 27, 13, 0, tzinfo=UTC)
|
|
|
|
async with session_factory() as session:
|
|
session.add(
|
|
ScoringRun(
|
|
id=run_id,
|
|
tenant_id="tenant-atomicity",
|
|
subject_ref="lead-atomicity",
|
|
scoring_config_version="1.0.0",
|
|
idempotency_key="atomicity-key",
|
|
payload_hash="1" * 64,
|
|
status="running",
|
|
snapshot_data={},
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
lease_expires_at=created_at,
|
|
)
|
|
)
|
|
session.add(
|
|
OutboxEvent(
|
|
id=str(uuid.uuid4()),
|
|
event_type="lead_scoring.run.terminal",
|
|
event_version="1.0.0",
|
|
run_id=run_id,
|
|
payload={},
|
|
occurred_at=created_at,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
with pytest.raises(IntegrityError):
|
|
await recover_interrupted_runs(
|
|
session_factory,
|
|
clock=lambda: created_at + timedelta(seconds=1),
|
|
)
|
|
|
|
async with session_factory() as session:
|
|
run = await session.get(ScoringRun, run_id)
|
|
assert run is not None
|
|
assert run.status == "running"
|
|
assert run.error_code is None
|
|
assert run.retryable is None
|
|
|
|
await engine.dispose()
|