From 82e500c19a0a04ca08c70323f3354d2b77c765fb Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Wed, 3 Jun 2026 23:26:17 -0300 Subject: [PATCH] test(jobs): hard-proof JobService store methods on real sqlite and postgres --- .../unit/background_execution/conftest.py | 45 ++++++- .../test_store_hard_proof.py | 122 ++++++++++++++++++ 2 files changed, 164 insertions(+), 3 deletions(-) create mode 100644 src/backend/tests/unit/background_execution/test_store_hard_proof.py diff --git a/src/backend/tests/unit/background_execution/conftest.py b/src/backend/tests/unit/background_execution/conftest.py index 3418c81c1d..0010be7122 100644 --- a/src/backend/tests/unit/background_execution/conftest.py +++ b/src/backend/tests/unit/background_execution/conftest.py @@ -7,9 +7,9 @@ fakes: (only when ``LANGFLOW_TEST_DATABASE_URI`` is set; CI always sets it). - ``hard_proof_redis_url`` yields a real Redis URL from ``LANGFLOW_TEST_REDIS_URL`` (skips when unset), flushed clean before and after each test. - -Phase 0 keeps these minimal: they yield connectable URLs. Later phases extend -``hard_proof_db_url`` to create the background-execution tables. +- ``hard_proof_job_service`` runs the real Alembic migrations against + ``hard_proof_db_url`` and binds ``session_scope()`` to it, so store methods are + exercised on both real SQLite and real Postgres. """ from __future__ import annotations @@ -25,6 +25,8 @@ import pytest if TYPE_CHECKING: from collections.abc import AsyncGenerator + from langflow.services.jobs.service import JobService + def _async_pg_url(raw: str) -> str: """Normalize a Postgres URL to the async asyncpg dialect (the driver we install).""" @@ -60,6 +62,43 @@ async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[st yield _async_pg_url(raw) +@pytest.fixture +async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobService, None]: + """Yield a JobService whose ``session_scope()`` is bound to a real, migrated DB. + + Runs the real Alembic migrations against ``hard_proof_db_url`` (sqlite + postgres) + through the production ``DatabaseService`` path, then swaps that service into the + service manager so ``JobService``'s ``session_scope()`` resolves to it. This proves + the store methods on BOTH real SQLite and real Postgres, and that the migrations + themselves apply on both engines. + """ + from langflow.services.database.factory import DatabaseServiceFactory + from langflow.services.deps import get_settings_service + from langflow.services.jobs.service import JobService + from lfx.services.manager import get_service_manager + from lfx.services.schema import ServiceType + + manager = get_service_manager() + settings_service = get_settings_service() + original_url = settings_service.settings.database_url + original_db_service = manager.services.pop(ServiceType.DATABASE_SERVICE, None) + + settings_service.settings.database_url = hard_proof_db_url + db_service = DatabaseServiceFactory().create(settings_service) + manager.services[ServiceType.DATABASE_SERVICE] = db_service + + try: + await db_service.run_migrations() + yield JobService() + finally: + manager.services.pop(ServiceType.DATABASE_SERVICE, None) + with contextlib.suppress(Exception): + await db_service.teardown() + settings_service.settings.database_url = original_url + if original_db_service is not None: + manager.services[ServiceType.DATABASE_SERVICE] = original_db_service + + @pytest.fixture async def hard_proof_redis_url() -> AsyncGenerator[str, None]: """Yield a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``, flushed clean. diff --git a/src/backend/tests/unit/background_execution/test_store_hard_proof.py b/src/backend/tests/unit/background_execution/test_store_hard_proof.py new file mode 100644 index 0000000000..196269c00c --- /dev/null +++ b/src/backend/tests/unit/background_execution/test_store_hard_proof.py @@ -0,0 +1,122 @@ +"""Hard-proof store tests: JobService methods against REAL SQLite and REAL Postgres. + +These bind ``session_scope()`` to a real, migrated database (via the +``hard_proof_job_service`` fixture) so every store method is exercised on both +engines, not just the temp-file SQLite the ``client`` fixture provides. Running +the real Alembic migrations against both engines here also re-proves the schema +applies on Postgres (JSONB columns, the execution_signal_type_enum, the +uq_job_events_job_id_seq unique constraint). +""" + +from __future__ import annotations + +from uuid import uuid4 + +import pytest +from langflow.services.database.models.jobs.model import JobStatus, SignalType + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_set_result_and_error_persist(hard_proof_job_service) -> None: + service = hard_proof_job_service + job_id = uuid4() + await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) + + await service.set_result(job_id, {"output_text": "hi", "session_id": "s1"}) + await service.set_error(job_id, {"type": "boom", "message": "x"}) + + fetched = await service.get_job_by_job_id(job_id) + assert fetched.result == {"output_text": "hi", "session_id": "s1"} + assert fetched.error == {"type": "boom", "message": "x"} + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_append_and_read_events_ordered(hard_proof_job_service) -> None: + service = hard_proof_job_service + job_id = uuid4() + await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) + + assert await service.append_event(job_id, "e1", {"i": 1}) == 1 + assert await service.append_event(job_id, "e2", {"i": 2}) == 2 + assert await service.append_event(job_id, "e3", {"i": 3}) == 3 + + tail = await service.read_events(job_id, after_seq=1) + assert [e.seq for e in tail] == [2, 3] + assert tail[0].payload == {"i": 2} + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None: + service = hard_proof_job_service + job_a = uuid4() + job_b = uuid4() + await service.create_job(job_id=job_a, flow_id=uuid4(), user_id=uuid4()) + await service.create_job(job_id=job_b, flow_id=uuid4(), user_id=uuid4()) + + assert await service.append_event(job_a, "x", {}) == 1 + assert await service.append_event(job_b, "x", {}) == 1 + assert await service.append_event(job_a, "x", {}) == 2 + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_write_and_read_signals(hard_proof_job_service) -> None: + service = hard_proof_job_service + job_id = uuid4() + await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) + + assert await service.unconsumed_signals(job_id) == [] + signal = await service.write_signal(job_id, SignalType.STOP) + assert signal.signal_type == SignalType.STOP + assert signal.consumed_at is None + + pending = await service.unconsumed_signals(job_id) + assert len(pending) == 1 + assert pending[0].signal_type == SignalType.STOP + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> None: + service = hard_proof_job_service + orphan = uuid4() + queued = uuid4() + await service.create_job(job_id=orphan, flow_id=uuid4(), user_id=uuid4()) + await service.update_job_status(orphan, JobStatus.IN_PROGRESS) + await service.create_job(job_id=queued, flow_id=uuid4(), user_id=uuid4()) + + swept = await service.sweep_orphans() + assert orphan in swept + assert queued not in swept + + orphan_job = await service.get_job_by_job_id(orphan) + assert orphan_job.status == JobStatus.FAILED + assert orphan_job.error == {"type": "worker_lost"} + assert orphan_job.finished_timestamp is not None + + events = await service.read_events(orphan) + assert [e.event_type for e in events] == ["run_failed"] + assert events[0].payload == {"type": "worker_lost"} + + queued_job = await service.get_job_by_job_id(queued) + assert queued_job.status == JobStatus.QUEUED + + +@pytest.mark.hard_proof +@pytest.mark.no_blockbuster +async def test_get_jobs_by_flow_id_orders_by_created_timestamp(hard_proof_job_service) -> None: + """Regression on real engines: ordering by created_timestamp must not raise.""" + service = hard_proof_job_service + flow_id = uuid4() + user_id = uuid4() + first = uuid4() + second = uuid4() + await service.create_job(job_id=first, flow_id=flow_id, user_id=user_id) + await service.create_job(job_id=second, flow_id=flow_id, user_id=user_id) + + jobs = await service.get_jobs_by_flow_id(flow_id, user_id) + returned = {job.job_id for job in jobs} + assert {first, second} <= returned