from contextlib import contextmanager from datetime import datetime from pathlib import Path from types import SimpleNamespace import sqlite3 import pytest from src.database.models import Base, Account, EmailService, RegistrationTask, Sub2ApiService from src.database.session import DatabaseSessionManager from src.database import external_batch_crud import src.core.external_batches.service as external_batch_service_module import src.web.routes.registration as registration_routes from src.core.external_batches.service import ExternalBatchCreateRequest, ExternalBatchService from src.core.external_batches.recovery import recover_interrupted_external_batches def _make_db(name: str): runtime_dir = Path("tests_runtime") runtime_dir.mkdir(exist_ok=True) db_path = runtime_dir / name if db_path.exists(): db_path.unlink() manager = DatabaseSessionManager(f"sqlite:///{db_path}") Base.metadata.create_all(bind=manager.engine) return manager def _seed_temp_mail(session): session.add(EmailService(service_type="temp_mail", name="tm-1", config={"domain": "a.test"}, enabled=True, priority=0)) @contextmanager def _session_context(manager): session = manager.SessionLocal() try: yield session finally: session.close() def _finalize_batch_with_failure(session, service, batch_uuid, failure_reason, item_reason=None): external_batch_crud.update_batch(session, batch_uuid, failure_reason=failure_reason) items = external_batch_crud.list_batch_items(session, batch_uuid) for item in items: external_batch_crud.update_batch_item( session, item.id, status="failed", failure_reason=item_reason or failure_reason, ) return service.recompute_summary(session, batch_uuid) def _run_scenario_and_assert_categories(manager, scenario): service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) scenario(session, batch.batch_uuid) refreshed = service.recompute_summary(session, batch.batch_uuid) return refreshed def test_create_batch_persists_batch_and_items_and_is_idempotent(): manager = _make_db("external_batch_create.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=3, idempotency_key="batch-001", email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=1, interval_max=2, ) with manager.session_scope() as session: first = service.create_batch(session, request) second = service.create_batch(session, request) items = external_batch_crud.list_batch_items(session, first.batch_uuid) assert first.id == second.id assert first.batch_uuid == second.batch_uuid assert len(items) == 3 assert [item.item_index for item in items] == [0, 1, 2] assert all(item.registration_task_uuid for item in items) def test_recompute_summary_marks_completed_partial_and_counts_uploads(): manager = _make_db("external_batch_summary.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) session.add(Sub2ApiService(name="sub2-1", api_url="https://sub2.test", api_key="k1", enabled=True, priority=0)) request = ExternalBatchCreateRequest( count=3, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=True, upload_provider="sub2api", upload_service_id=1, mode="pipeline", concurrency=1, interval_min=1, interval_max=1, ) with manager.session_scope() as session: batch = service.create_batch(session, request) items = external_batch_crud.list_batch_items(session, batch.batch_uuid) external_batch_crud.update_batch_item(session, items[0].id, status="completed", upload_status="success") external_batch_crud.update_batch_item(session, items[1].id, status="failed", failure_reason="no_available_email_service", upload_status="skipped") external_batch_crud.update_batch_item(session, items[2].id, status="completed", upload_status="failed", upload_error="upload_failed") refreshed = service.recompute_summary(session, batch.batch_uuid) assert refreshed.status == "completed_partial" assert refreshed.completed_count == 3 assert refreshed.success_count == 2 assert refreshed.failed_count == 1 assert refreshed.upload_success_count == 1 assert refreshed.upload_failed_count == 1 def test_recovery_marks_interrupted_batches_and_registration_tasks_failed(): manager = _make_db("external_batch_recovery.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="parallel", concurrency=2, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) items = external_batch_crud.list_batch_items(session, batch.batch_uuid) external_batch_crud.update_batch(session, batch.batch_uuid, status="running") external_batch_crud.update_batch_item(session, items[0].id, status="running") external_batch_crud.update_batch_item(session, items[1].id, status="pending") session.query(RegistrationTask).filter(RegistrationTask.task_uuid == items[0].registration_task_uuid).update({"status": "running"}) session.query(RegistrationTask).filter(RegistrationTask.task_uuid == items[1].registration_task_uuid).update({"status": "pending"}) with manager.session_scope() as session: recover_interrupted_external_batches(session) refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) refreshed_items = external_batch_crud.list_batch_items(session, batch.batch_uuid) reg_tasks = {task.task_uuid: task for task in session.query(RegistrationTask).all()} assert refreshed.status == "failed" assert refreshed.failure_reason == "service_restarted" assert refreshed.failure_category == "transient" assert all(item.status == "failed" for item in refreshed_items) assert all(item.failure_reason == "service_restarted" for item in refreshed_items) assert all(reg_tasks[item.registration_task_uuid].status == "failed" for item in refreshed_items) def test_run_batch_pipeline_completes_items_and_uploads(monkeypatch): manager = _make_db("external_batch_run_pipeline.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) session.add(Sub2ApiService(name="sub2-1", api_url="https://sub2.test", api_key="k1", enabled=True, priority=0)) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=True, upload_provider="sub2api", upload_service_id=1, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) async def _fake_run_registration_task(task_uuid, *_args, **_kwargs): with manager.session_scope() as session: task = session.query(RegistrationTask).filter(RegistrationTask.task_uuid == task_uuid).first() task.status = "completed" task.result = {"email": f"{task_uuid}@example.com", "account_id": f"acc-{task_uuid}"} session.add( Account( email=f"{task_uuid}@example.com", password="pw", account_id=f"acc-{task_uuid}", email_service="temp_mail", email_service_id="svc-1", status="active", source="register", ) ) monkeypatch.setattr(registration_routes, "run_registration_task", _fake_run_registration_task) monkeypatch.setattr( external_batch_service_module, "upload_registered_account", lambda account, target: SimpleNamespace(success=True, message="ok", account=account, target=target), ) import asyncio asyncio.run(service.run_batch(batch.batch_uuid)) with manager.session_scope() as session: refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) items = external_batch_crud.list_batch_items(session, batch.batch_uuid) assert refreshed.status == "completed" assert refreshed.success_count == 2 assert refreshed.upload_success_count == 2 assert all(item.status == "completed" for item in items) assert all(item.upload_status == "success" for item in items) def test_run_batch_marks_prefailed_item_without_calling_registration(monkeypatch): manager = _make_db("external_batch_prefailed_item.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) item = external_batch_crud.list_batch_items(session, batch.batch_uuid)[0] external_batch_crud.update_batch_item(session, item.id, failure_reason="no_available_email_service") task_uuid = item.registration_task_uuid monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) calls = {"count": 0} async def _fake_run_registration_task(*_args, **_kwargs): calls["count"] += 1 monkeypatch.setattr(registration_routes, "run_registration_task", _fake_run_registration_task) import asyncio asyncio.run(service.run_batch(batch.batch_uuid)) with manager.session_scope() as session: refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) item = external_batch_crud.list_batch_items(session, batch.batch_uuid)[0] task = session.query(RegistrationTask).filter(RegistrationTask.task_uuid == task_uuid).first() assert calls["count"] == 0 assert refreshed.status == "failed" assert item.status == "failed" assert item.failure_reason == "no_available_email_service" assert task.status == "failed" assert task.error_message == "no_available_email_service" def test_run_batch_uses_account_id_fallback_for_upload_lookup(monkeypatch): manager = _make_db("external_batch_account_id_fallback.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) session.add(Sub2ApiService(name="sub2-1", api_url="https://sub2.test", api_key="k1", enabled=True, priority=0)) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=True, upload_provider="sub2api", upload_service_id=1, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) async def _fake_run_registration_task(task_uuid, *_args, **_kwargs): with manager.session_scope() as session: task = session.query(RegistrationTask).filter(RegistrationTask.task_uuid == task_uuid).first() task.status = "completed" task.result = {"account_id": f"acc-{task_uuid}"} session.add( Account( email=f"stored-{task_uuid}@example.com", password="pw", account_id=f"acc-{task_uuid}", email_service="temp_mail", email_service_id="svc-1", status="active", source="register", ) ) seen = {} def _fake_upload(account, target): seen["email"] = account.email if account else None seen["provider"] = target.provider if target else None return SimpleNamespace(success=False, message="upload_failed") monkeypatch.setattr(registration_routes, "run_registration_task", _fake_run_registration_task) monkeypatch.setattr(external_batch_service_module, "upload_registered_account", _fake_upload) import asyncio asyncio.run(service.run_batch(batch.batch_uuid)) with manager.session_scope() as session: refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) item = external_batch_crud.list_batch_items(session, batch.batch_uuid)[0] assert seen["provider"] == "sub2api" assert seen["email"].startswith("stored-") assert refreshed.status == "completed" assert refreshed.upload_failed_count == 1 assert item.upload_status == "failed" assert item.upload_error == "upload_failed" def test_request_cancel_rejects_finished_batch(): manager = _make_db("external_batch_cancel_finished.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) external_batch_crud.update_batch(session, batch.batch_uuid, status="completed") with manager.session_scope() as session: with pytest.raises(ValueError, match="already finished"): service.request_cancel(session, batch.batch_uuid) def test_recompute_summary_classifies_service_restarted_as_transient(): manager = _make_db("external_batch_failure_category_service_restarted.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) refreshed = _finalize_batch_with_failure(session, service, batch.batch_uuid, "service_restarted") assert refreshed.failure_category == "transient" def test_recompute_summary_classifies_config_reasons_as_config(): manager = _make_db("external_batch_failure_category_config.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) reason = "no enabled upload services for provider sub2api" with manager.session_scope() as session: batch = service.create_batch(session, request) refreshed = _finalize_batch_with_failure(session, service, batch.batch_uuid, reason) assert refreshed.failure_category == "config" def test_recompute_summary_classifies_business_reasons_as_business(): manager = _make_db("external_batch_failure_category_business.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) reason = "outlook requested_service_id cannot be reused when count > 1" with manager.session_scope() as session: batch = service.create_batch(session, request) refreshed = _finalize_batch_with_failure(session, service, batch.batch_uuid, reason) assert refreshed.failure_category == "business" def test_recompute_summary_classifies_no_available_email_service_as_transient(): manager = _make_db("external_batch_failure_category_no_available.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) refreshed = _finalize_batch_with_failure( session, service, batch.batch_uuid, None, item_reason="no_available_email_service", ) assert refreshed.failure_category == "transient" def test_recompute_summary_defaults_unknown_failure_to_transient(): manager = _make_db("external_batch_failure_category_unknown.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) reason = "registration_failed" with manager.session_scope() as session: batch = service.create_batch(session, request) refreshed = _finalize_batch_with_failure(session, service, batch.batch_uuid, reason) assert refreshed.failure_category == "transient" def test_recompute_summary_uses_upload_error_as_transient_fallback(): manager = _make_db("external_batch_failure_category_upload_error.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) session.add(Sub2ApiService(name="sub2-upload-fallback", api_url="https://sub2.test", api_key="k1", enabled=True, priority=0)) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=True, upload_provider="sub2api", upload_service_id=1, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) for item in external_batch_crud.list_batch_items(session, batch.batch_uuid): external_batch_crud.update_batch_item( session, item.id, status="failed", upload_status="failed", upload_error="provider timeout", ) refreshed = service.recompute_summary(session, batch.batch_uuid) assert refreshed.status == "failed" assert refreshed.failure_category == "transient" def test_recompute_summary_prefers_batch_failure_reason_for_classification(): manager = _make_db("external_batch_failure_category_priority.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=2, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) external_batch_crud.update_batch( session, batch.batch_uuid, failure_reason="unsupported upload provider: custom", ) for item in external_batch_crud.list_batch_items(session, batch.batch_uuid): external_batch_crud.update_batch_item( session, item.id, status="failed", failure_reason="outlook requested_service_id cannot be reused when count > 1", ) refreshed = service.recompute_summary(session, batch.batch_uuid) assert refreshed.status == "failed" assert refreshed.failure_category == "config" def test_non_failed_statuses_keep_null_failure_category(): manager = _make_db("external_batch_failure_category_non_failed.db") scenarios = [ ("pending", lambda session, batch_uuid: None), ( "running", lambda session, batch_uuid: ( external_batch_crud.update_batch( session, batch_uuid, started_at=datetime.utcnow() ), external_batch_crud.update_batch_item( session, external_batch_crud.list_batch_items(session, batch_uuid)[0].id, status="running", ), ), ), ( "completed", lambda session, batch_uuid: [ external_batch_crud.update_batch_item(session, item.id, status="completed") for item in external_batch_crud.list_batch_items(session, batch_uuid) ], ), ( "completed_partial", lambda session, batch_uuid: ( external_batch_crud.update_batch_item( session, external_batch_crud.list_batch_items(session, batch_uuid)[0].id, status="completed", ), external_batch_crud.update_batch_item( session, external_batch_crud.list_batch_items(session, batch_uuid)[1].id, status="failed", failure_reason="upload_failed", ), ), ), ( "cancelled", lambda session, batch_uuid: ( external_batch_crud.update_batch(session, batch_uuid, cancel_requested=True), [ external_batch_crud.update_batch_item( session, item.id, status="cancelled", ) for item in external_batch_crud.list_batch_items(session, batch_uuid) ], ), ), ] for expected_status, scenario in scenarios: refreshed = _run_scenario_and_assert_categories(manager, scenario) assert refreshed.status == expected_status assert refreshed.failure_category is None def test_sqlite_migration_adds_failure_category_column(): runtime_dir = Path("tests_runtime") runtime_dir.mkdir(exist_ok=True) db_path = runtime_dir / "external_batch_failure_category_migration.db" if db_path.exists(): db_path.unlink() conn = sqlite3.connect(db_path) columns = """ id INTEGER PRIMARY KEY, batch_uuid VARCHAR(36) NOT NULL, status VARCHAR(32) NOT NULL, failure_reason VARCHAR(255), created_at DATETIME, updated_at DATETIME """ conn.execute(f"CREATE TABLE external_registration_batches ({columns})") conn.commit() conn.close() manager = DatabaseSessionManager(f"sqlite:///{db_path}") manager.migrate_tables() manager.migrate_tables() with sqlite3.connect(db_path) as conn: info = conn.execute("PRAGMA table_info('external_registration_batches')").fetchall() column_names = [row[1] for row in info] assert column_names.count("failure_category") == 1 def test_create_external_registration_batch_replay_returns_persisted_failure_category(monkeypatch): manager = _make_db("external_batch_failure_category_replay.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key="replay-key", email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) external_batch_crud.update_batch( session, batch.batch_uuid, status="failed", failure_reason="unsupported upload provider: custom", failure_category="config", ) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) replay = external_batch_service_module.create_external_registration_batch( { "count": 1, "idempotency_key": "replay-key", "email": {"type": "temp_mail"}, "upload": {"enabled": False}, "execution": {"mode": "pipeline", "concurrency": 1, "interval_min": 0, "interval_max": 0}, }, background_tasks=None, ) assert replay["idempotent_replay"] is True assert replay["failure_category"] == "config" def test_get_external_registration_batch_status_backfills_missing_failure_category(monkeypatch): manager = _make_db("external_batch_failure_category_backfill_get.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) external_batch_crud.update_batch( session, batch.batch_uuid, status="failed", failure_reason="unsupported upload provider: custom", failure_category=None, ) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) response = external_batch_service_module.get_external_registration_batch_status(batch.batch_uuid) with manager.session_scope() as session: refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) assert response["failure_category"] == "config" assert refreshed.failure_category == "config" def test_create_external_registration_batch_replay_backfills_missing_failure_category(monkeypatch): manager = _make_db("external_batch_failure_category_backfill_replay.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key="replay-backfill-key", email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) external_batch_crud.update_batch( session, batch.batch_uuid, status="failed", failure_reason="unsupported upload provider: custom", failure_category=None, ) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) replay = external_batch_service_module.create_external_registration_batch( { "count": 1, "idempotency_key": "replay-backfill-key", "email": {"type": "temp_mail"}, "upload": {"enabled": False}, "execution": {"mode": "pipeline", "concurrency": 1, "interval_min": 0, "interval_max": 0}, }, background_tasks=None, ) with manager.session_scope() as session: refreshed = external_batch_crud.get_batch_by_uuid(session, batch.batch_uuid) assert replay["idempotent_replay"] is True assert replay["failure_category"] == "config" assert refreshed.failure_category == "config" def test_get_external_registration_batch_status_backfills_from_item_failure_reason(monkeypatch): manager = _make_db("external_batch_failure_category_backfill_item_reason.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) item = external_batch_crud.list_batch_items(session, batch.batch_uuid)[0] external_batch_crud.update_batch( session, batch.batch_uuid, status="failed", failure_reason=None, failure_category=None, ) external_batch_crud.update_batch_item( session, item.id, status="failed", failure_reason="no_available_email_service", ) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) response = external_batch_service_module.get_external_registration_batch_status(batch.batch_uuid) assert response["failure_category"] == "transient" def test_get_external_registration_batch_status_backfills_from_upload_error(monkeypatch): manager = _make_db("external_batch_failure_category_backfill_upload_error.db") service = ExternalBatchService() with manager.session_scope() as session: _seed_temp_mail(session) request = ExternalBatchCreateRequest( count=1, idempotency_key=None, email_type="temp_mail", email_service_id=None, upload_enabled=False, upload_provider=None, upload_service_id=None, mode="pipeline", concurrency=1, interval_min=0, interval_max=0, ) with manager.session_scope() as session: batch = service.create_batch(session, request) item = external_batch_crud.list_batch_items(session, batch.batch_uuid)[0] external_batch_crud.update_batch( session, batch.batch_uuid, status="failed", failure_reason=None, failure_category=None, ) external_batch_crud.update_batch_item( session, item.id, status="failed", upload_status="failed", upload_error="provider timeout", ) monkeypatch.setattr("src.database.session.get_db", lambda: _session_context(manager)) response = external_batch_service_module.get_external_registration_batch_status(batch.batch_uuid) assert response["failure_category"] == "transient" def test_create_external_registration_batch_marks_replay_true_when_create_batch_returns_replay(monkeypatch): fake_batch = SimpleNamespace( batch_uuid="race-batch", status="failed", requested_count=1, completed_count=1, success_count=0, failed_count=1, upload_success_count=0, upload_failed_count=0, failure_reason="unsupported upload provider: custom", failure_category=None, created_at=None, started_at=None, completed_at=None, recent_errors=[], _idempotent_replay=True, ) @contextmanager def _noop_db(): yield object() monkeypatch.setattr("src.database.session.get_db", _noop_db) monkeypatch.setattr(external_batch_crud, "get_batch_by_idempotency_key", lambda db, key: None) monkeypatch.setattr(ExternalBatchService, "create_batch", lambda self, db, request: fake_batch) def _ensure(self, db, batch): batch.failure_category = "config" return batch monkeypatch.setattr(ExternalBatchService, "ensure_failure_category", _ensure) response = external_batch_service_module.create_external_registration_batch( { "count": 1, "idempotency_key": "race-key", "email": {"type": "temp_mail"}, "upload": {"enabled": False}, "execution": {"mode": "pipeline", "concurrency": 1, "interval_min": 0, "interval_max": 0}, }, background_tasks=None, ) assert response["idempotent_replay"] is True assert response["failure_category"] == "config" def test_create_external_registration_batch_marks_replay_true_for_non_failed_detached_batch(monkeypatch): fake_batch = SimpleNamespace( batch_uuid="race-batch-pending", status="pending", requested_count=1, completed_count=0, success_count=0, failed_count=0, upload_success_count=0, upload_failed_count=0, failure_reason=None, failure_category=None, created_at=None, started_at=None, completed_at=None, recent_errors=[], _idempotent_replay=True, ) @contextmanager def _noop_db(): yield object() monkeypatch.setattr("src.database.session.get_db", _noop_db) monkeypatch.setattr(external_batch_crud, "get_batch_by_idempotency_key", lambda db, key: None) monkeypatch.setattr(ExternalBatchService, "create_batch", lambda self, db, request: fake_batch) response = external_batch_service_module.create_external_registration_batch( { "count": 1, "idempotency_key": "race-key-pending", "email": {"type": "temp_mail"}, "upload": {"enabled": False}, "execution": {"mode": "pipeline", "concurrency": 1, "interval_min": 0, "interval_max": 0}, }, background_tasks=None, ) assert response["idempotent_replay"] is True assert response["failure_category"] is None def test_create_external_registration_batch_keeps_replay_flag_when_backfill_returns_new_batch(monkeypatch): replay_batch = SimpleNamespace( batch_uuid="race-batch-failed", status="failed", requested_count=1, completed_count=1, success_count=0, failed_count=1, upload_success_count=0, upload_failed_count=0, failure_reason="unsupported upload provider: custom", failure_category=None, created_at=None, started_at=None, completed_at=None, recent_errors=[], _idempotent_replay=True, ) updated_batch = SimpleNamespace( batch_uuid="race-batch-failed", status="failed", requested_count=1, completed_count=1, success_count=0, failed_count=1, upload_success_count=0, upload_failed_count=0, failure_reason="unsupported upload provider: custom", failure_category="config", created_at=None, started_at=None, completed_at=None, recent_errors=[], ) @contextmanager def _noop_db(): yield object() monkeypatch.setattr("src.database.session.get_db", _noop_db) monkeypatch.setattr(external_batch_crud, "get_batch_by_idempotency_key", lambda db, key: None) monkeypatch.setattr(ExternalBatchService, "create_batch", lambda self, db, request: replay_batch) monkeypatch.setattr(ExternalBatchService, "ensure_failure_category", lambda self, db, batch: updated_batch) response = external_batch_service_module.create_external_registration_batch( { "count": 1, "idempotency_key": "race-key-failed", "email": {"type": "temp_mail"}, "upload": {"enabled": False}, "execution": {"mode": "pipeline", "concurrency": 1, "interval_min": 0, "interval_max": 0}, }, background_tasks=None, ) assert response["idempotent_replay"] is True assert response["failure_category"] == "config"