Files
Nexus/tests/chain/test_batch_job_flow.py
2026-07-08 22:31:31 +08:00

127 lines
3.5 KiB
Python

"""CH-BAT-001 — batch category job → Redis live → MySQL finalize."""
from __future__ import annotations
import asyncio
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from server.application.services.server_batch_service import get_batch_job, start_batch_job
from server.infrastructure.database.server_batch_job_repo import ServerBatchJobRepositoryImpl
from server.infrastructure.database.server_repo import ServerRepositoryImpl
pytestmark = pytest.mark.chain
class _MemRedis:
"""Minimal Redis stub for server_batch_store."""
def __init__(self) -> None:
self._data: dict[str, str] = {}
self._seq = 0
async def incr(self, key: str) -> int:
self._seq += 1
self._data[key] = str(self._seq)
return self._seq
async def set(self, key: str, value: str, ex: int | None = None) -> None:
self._data[key] = value
async def get(self, key: str) -> str | None:
return self._data.get(key)
async def scan(self, cursor: int, match: str | None = None, count: int = 50):
keys = [
k
for k in self._data
if k.startswith("server_batch:") and k != "server_batch:next_id"
]
return 0, keys
class _SessionCtx:
def __init__(self, session):
self._session = session
async def __aenter__(self):
return self._session
async def __aexit__(self, exc_type, exc, tb) -> None:
if exc:
await self._session.rollback()
@pytest.mark.asyncio
async def test_batch_category_job_completes_and_persists(
db_session,
test_admin,
test_server,
monkeypatch,
):
mem = _MemRedis()
monkeypatch.setattr(
"server.infrastructure.redis.server_batch_store.get_redis",
lambda: mem,
)
monkeypatch.setattr(
"server.application.services.server_batch_service.AsyncSessionLocal",
lambda: _SessionCtx(db_session),
)
monkeypatch.setattr(
"server.application.services.server_batch_service._notify_progress",
AsyncMock(),
)
monkeypatch.setattr(
"server.application.services.server_batch_service._notify_complete",
AsyncMock(),
)
captured: list = []
def capture_create_task(coro):
captured.append(coro)
return MagicMock()
class _AsyncioProxy:
def __getattr__(self, name):
return getattr(asyncio, name)
def create_task(self, coro):
return capture_create_task(coro)
monkeypatch.setattr(
"server.application.services.server_batch_service.asyncio",
_AsyncioProxy(),
)
started = await start_batch_job(
op="category",
server_ids=[test_server.id],
operator=test_admin["admin"].username,
params={"category": "chain-batch-cat"},
)
assert started["status"] == "running"
assert len(captured) == 1
await captured[0]
detail = await get_batch_job(started["job_id"])
assert detail is not None
assert detail["status"] == "completed"
assert detail["success"] == 1
assert detail["failed"] == 0
repo = ServerBatchJobRepositoryImpl(db_session)
row = await repo.get_by_id(started["job_id"])
assert row is not None
assert row.status == "completed"
assert json.loads(row.server_ids) == [test_server.id]
srv_repo = ServerRepositoryImpl(db_session)
updated = await srv_repo.get_by_id(test_server.id)
assert updated is not None
assert updated.category == "chain-batch-cat"