diff --git a/.github/workflows/migration-validation.yml b/.github/workflows/migration-validation.yml index 292bea6e85..1960d8b83b 100644 --- a/.github/workflows/migration-validation.yml +++ b/.github/workflows/migration-validation.yml @@ -58,8 +58,8 @@ jobs: run: | uv run pytest src/backend/tests/unit/alembic/test_migration_execution.py -x -v - background-hard-proof: - name: Background Execution Hard-Proof (real Postgres + Redis) + background-real-service: + name: Background Execution Real-Service Tests (real Postgres + Redis) runs-on: ubuntu-latest services: @@ -104,12 +104,12 @@ jobs: run: | uv sync --extra postgresql - - name: Run hard-proof tests + - name: Run real-service tests env: LANGFLOW_TEST_DATABASE_URI: "postgresql://langflow:langflow@localhost:5432/langflow" # pragma: allowlist secret LANGFLOW_TEST_REDIS_URL: "redis://localhost:6379/0" run: | - uv run pytest src/backend/tests/unit/background_execution -m hard_proof -x -v + uv run pytest src/backend/tests/unit/background_execution -m real_services -x -v validate-migration: name: Migration Pattern Validation diff --git a/Makefile b/Makefile index 8270bd03d3..78446d00bd 100644 --- a/Makefile +++ b/Makefile @@ -168,12 +168,12 @@ unit_tests: ## run unit tests unit_tests_looponfail: @make unit_tests args="-f" -hard_proof_tests: ## run real-instance hard-proof tests (needs LANGFLOW_TEST_DATABASE_URI + LANGFLOW_TEST_REDIS_URL) +real_services_tests: ## run tests that need real service instances (needs LANGFLOW_TEST_DATABASE_URI + LANGFLOW_TEST_REDIS_URL) @uv sync --frozen uv run pytest src/backend/tests/unit \ --ignore=src/backend/tests/integration \ --ignore=src/backend/tests/unit/template \ - -m hard_proof -ra $(args) + -m real_services -ra $(args) lfx_tests: ## run lfx package unit tests @echo 'Running LFX Package Tests...' diff --git a/pyproject.toml b/pyproject.toml index 29ad5e633c..5930487940 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -205,7 +205,7 @@ markers = [ "integration: Integration tests", "slow: Slow-running tests", "security: Security regression tests (IDOR, auth, access control)", - "hard_proof: Real-instance proof tests (real SQLite + real Postgres + real Redis)" + "real_services: Tests that need real service instances (real SQLite + real Postgres + real Redis)" ] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" diff --git a/src/backend/tests/conftest.py b/src/backend/tests/conftest.py index fb101447f2..30cd6cb0e1 100644 --- a/src/backend/tests/conftest.py +++ b/src/backend/tests/conftest.py @@ -111,7 +111,7 @@ def pytest_configure(config): config.addinivalue_line("markers", "api_key_required: run only if the api key is set in the environment variables") config.addinivalue_line( "markers", - "hard_proof: Real-instance proof tests (real SQLite + real Postgres + real Redis)", + "real_services: Tests that need real service instances (real SQLite + real Postgres + real Redis)", ) data_path = Path(__file__).parent.absolute() / "data" diff --git a/src/backend/tests/unit/api/v2/test_workflow_agui.py b/src/backend/tests/unit/api/v2/test_workflow_agui.py index 1e3b345ee8..6aa7c257e6 100644 --- a/src/backend/tests/unit/api/v2/test_workflow_agui.py +++ b/src/backend/tests/unit/api/v2/test_workflow_agui.py @@ -1235,7 +1235,7 @@ class TestStopWorkflowEndToEnd: The deterministic CANCELLED outcome of a stop is proven without an HTTP-level race in ``services/background_execution/test_service.py::test_stop_cancels_job`` and - ``background_execution/test_facade_hard_proof.py::test_stop_signal_cancels_run`` + ``background_execution/test_facade_real_services.py::test_stop_signal_cancels_run`` (sqlite + postgres), so this test deliberately does not assert the exact terminal label under the unsynchronized fast-flow vs stop race. """ diff --git a/src/backend/tests/unit/background_execution/conftest.py b/src/backend/tests/unit/background_execution/conftest.py index 0010be7122..50e77792b0 100644 --- a/src/backend/tests/unit/background_execution/conftest.py +++ b/src/backend/tests/unit/background_execution/conftest.py @@ -1,14 +1,14 @@ -"""Real-instance fixtures for the background-execution hard-proof tier. +"""Real-instance fixtures for the background-execution real-service tier. -These fixtures back ``@pytest.mark.hard_proof`` tests with REAL engines, never +These fixtures back ``@pytest.mark.real_services`` tests with REAL engines, never fakes: -- ``hard_proof_db_url`` parametrizes over real SQLite (always) and real Postgres +- ``real_services_db_url`` parametrizes over real SQLite (always) and real Postgres (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`` +- ``real_services_redis_url`` yields a real Redis URL from ``LANGFLOW_TEST_REDIS_URL`` (skips when unset), flushed clean before and after each test. -- ``hard_proof_job_service`` runs the real Alembic migrations against - ``hard_proof_db_url`` and binds ``session_scope()`` to it, so store methods are +- ``real_services_job_service`` runs the real Alembic migrations against + ``real_services_db_url`` and binds ``session_scope()`` to it, so store methods are exercised on both real SQLite and real Postgres. """ @@ -40,7 +40,7 @@ def _async_pg_url(raw: str) -> str: @pytest.fixture(params=["sqlite", "postgres"]) -async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[str, None]: +async def real_services_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[str, None]: """Yield a real, connectable async DB URL. - sqlite: a real temp-file SQLite database (always runs). @@ -63,10 +63,10 @@ async def hard_proof_db_url(request: pytest.FixtureRequest) -> AsyncGenerator[st @pytest.fixture -async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobService, None]: +async def real_services_job_service(real_services_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) + Runs the real Alembic migrations against ``real_services_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 @@ -83,7 +83,7 @@ async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobSe 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 + settings_service.settings.database_url = real_services_db_url db_service = DatabaseServiceFactory().create(settings_service) manager.services[ServiceType.DATABASE_SERVICE] = db_service @@ -100,7 +100,7 @@ async def hard_proof_job_service(hard_proof_db_url: str) -> AsyncGenerator[JobSe @pytest.fixture -async def hard_proof_redis_url() -> AsyncGenerator[str, None]: +async def real_services_redis_url() -> AsyncGenerator[str, None]: """Yield a real Redis URL from ``LANGFLOW_TEST_REDIS_URL``, flushed clean. Skips when ``LANGFLOW_TEST_REDIS_URL`` is unset (CI sets it via the redis:7 diff --git a/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py b/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py index d63d2e2864..8a1309d1aa 100644 --- a/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py +++ b/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py @@ -19,7 +19,7 @@ from fastapi.sse import format_sse_event from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter from langflow.services.background_execution.service import BackgroundExecutionService -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services def _live_agui_run_started_bytes(seq: int) -> bytes: diff --git a/src/backend/tests/unit/background_execution/test_bounded_concurrency.py b/src/backend/tests/unit/background_execution/test_bounded_concurrency.py index f467c4faf4..290e85683b 100644 --- a/src/backend/tests/unit/background_execution/test_bounded_concurrency.py +++ b/src/backend/tests/unit/background_execution/test_bounded_concurrency.py @@ -1,4 +1,4 @@ -"""BOUNDED-CONCURRENCY hard proof. +"""BOUNDED-CONCURRENCY real service. Submit N+k jobs to the real ``InProcessExecutor`` configured for max concurrency N. A barrier holds every started job so the peak in-flight count is observable; @@ -15,7 +15,7 @@ import threading import pytest from langflow.services.background_execution.executor import InProcessExecutor -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services class _Probe: diff --git a/src/backend/tests/unit/background_execution/test_executor_hard_proof.py b/src/backend/tests/unit/background_execution/test_executor_real_services.py similarity index 98% rename from src/backend/tests/unit/background_execution/test_executor_hard_proof.py rename to src/backend/tests/unit/background_execution/test_executor_real_services.py index 043ec3a480..96b3b30cf9 100644 --- a/src/backend/tests/unit/background_execution/test_executor_hard_proof.py +++ b/src/backend/tests/unit/background_execution/test_executor_real_services.py @@ -1,4 +1,4 @@ -"""Hard-proof the in-process executor's worker resilience. +"""Verify the in-process executor's worker resilience. These tests drive the REAL ``InProcessExecutor`` (no mocks of our logic) and assert the worker pool survives the two ways a job can end abnormally: @@ -21,7 +21,7 @@ import asyncio import pytest from langflow.services.background_execution.executor import InProcessExecutor -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services async def test_pool_survives_repeated_job_cancellation(monkeypatch): diff --git a/src/backend/tests/unit/background_execution/test_facade_hard_proof.py b/src/backend/tests/unit/background_execution/test_facade_real_services.py similarity index 92% rename from src/backend/tests/unit/background_execution/test_facade_hard_proof.py rename to src/backend/tests/unit/background_execution/test_facade_real_services.py index 20c1884529..89299370eb 100644 --- a/src/backend/tests/unit/background_execution/test_facade_hard_proof.py +++ b/src/backend/tests/unit/background_execution/test_facade_real_services.py @@ -1,7 +1,7 @@ -"""Hard-proof the durable background path on real SQLite AND real Postgres. +"""Verify the durable background path on real SQLite AND real Postgres. The Runner + InMemoryLiveBus + facade durable replay all sit on ``JobService``, -whose ``session_scope()`` the ``hard_proof_job_service`` fixture binds to a real, +whose ``session_scope()`` the ``real_services_job_service`` fixture binds to a real, migrated DB parametrized over sqlite and postgres. These tests drive the real Runner with a scripted frame source (legitimate test input, not a mock of our logic) and assert durable persistence, ordering, terminal state, reattach replay, @@ -23,7 +23,7 @@ from langflow.services.database.models.jobs.model import JobStatus, SignalType if TYPE_CHECKING: from collections.abc import AsyncIterator -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services def _frame(event_type: str, data: dict) -> tuple[bytes, str]: @@ -35,8 +35,8 @@ def _runner(job_service, bus, job_id, source) -> JobRunner: return JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source) -async def test_durable_persistence_and_terminal_state(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_durable_persistence_and_terminal_state(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -60,8 +60,8 @@ async def test_durable_persistence_and_terminal_state(hard_proof_job_service): assert [e.seq for e in events] == [1, 2, 3] # contiguous, monotonic -async def test_reattach_replays_durable_after_completion(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_reattach_replays_durable_after_completion(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -88,8 +88,8 @@ async def test_reattach_replays_durable_after_completion(hard_proof_job_service) assert b"end_vertex" in bodies -async def test_reattach_from_midpoint_has_no_gap(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_reattach_from_midpoint_has_no_gap(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -110,8 +110,8 @@ async def test_reattach_from_midpoint_has_no_gap(hard_proof_job_service): assert seqs == [2, 3] -async def test_stop_signal_cancels_run(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_stop_signal_cancels_run(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) # A STOP written before the run makes the first boundary poll cancel it. @@ -128,8 +128,8 @@ async def test_stop_signal_cancels_run(hard_proof_job_service): assert job.status == JobStatus.CANCELLED -async def test_sweep_orphans_fails_in_progress(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_sweep_orphans_fails_in_progress(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) await job_service.update_job_status(job_id, JobStatus.IN_PROGRESS) @@ -161,7 +161,7 @@ def _echo_input_factory(*, request, **_kwargs): return _source -async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_service): +async def test_submit_persists_request_for_faithful_requeue(real_services_job_service): """``submit`` persists the request body on the job row (job_metadata.request). This is the durable record the startup sweep reads to replay original inputs. @@ -171,7 +171,7 @@ async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_servi from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service original_input = f"original-input-{uuid4()}" flow_id = uuid4() @@ -197,7 +197,7 @@ async def test_submit_persists_request_for_faithful_requeue(hard_proof_job_servi assert persisted == request -async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_job_service): +async def test_submit_redacts_inline_globals_from_persisted_request(real_services_job_service): """Inline ``globals`` (request-level secrets) must NOT land plaintext on the row. ``submit`` persists the request body for faithful replay, but request-level @@ -212,7 +212,7 @@ async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_j from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service flow_id = uuid4() secret = f"sk-secret-{uuid4()}" @@ -239,7 +239,7 @@ async def test_submit_redacts_inline_globals_from_persisted_request(hard_proof_j assert request["globals"] == {"OPENAI_API_KEY": secret} -async def test_requeued_queued_job_replays_original_input(hard_proof_job_service): +async def test_requeued_queued_job_replays_original_input(real_services_job_service): """A QUEUED job that survives a restart re-runs with its ORIGINAL input. Models "queued, never started, then the process died": the durable row is @@ -254,7 +254,7 @@ async def test_requeued_queued_job_replays_original_input(hard_proof_job_service from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service original_input = f"original-input-{uuid4()}" user_id = uuid4() flow_id = uuid4() @@ -295,7 +295,7 @@ async def test_requeued_queued_job_replays_original_input(hard_proof_job_service assert echoed == [original_input] -async def test_job_timeout_marks_timed_out(hard_proof_job_service): +async def test_job_timeout_marks_timed_out(real_services_job_service): """A run that overruns ``background_job_timeout`` ends TIMED_OUT. The runner wraps the drive in ``asyncio.wait_for(timeout=...)``; @@ -306,7 +306,7 @@ async def test_job_timeout_marks_timed_out(hard_proof_job_service): """ import asyncio - job_service = hard_proof_job_service + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -330,7 +330,7 @@ async def test_job_timeout_marks_timed_out(hard_proof_job_service): assert job.status == JobStatus.TIMED_OUT -async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_job_service): +async def test_events_reattach_after_restart_returns_on_terminal_job(real_services_job_service): """Reattaching to an already-terminal job after a restart MUST NOT hang. Models a process restart: a job runs to COMPLETED, persisting durable @@ -348,7 +348,7 @@ async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_ from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service user_id = uuid4() flow_id = uuid4() job_id = uuid4() @@ -381,7 +381,7 @@ async def test_events_reattach_after_restart_returns_on_terminal_job(hard_proof_ assert b"end" in body -async def test_events_replay_frames_are_sse_framed(hard_proof_job_service): +async def test_events_replay_frames_are_sse_framed(real_services_job_service): """Durable replay must emit SSE-framed bytes byte-compatible with live frames. Live frames are pre-SSE-framed with a ``data:`` line followed by an ``id:`` @@ -396,7 +396,7 @@ async def test_events_replay_frames_are_sse_framed(hard_proof_job_service): from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service user_id = uuid4() flow_id = uuid4() job_id = uuid4() @@ -423,7 +423,7 @@ async def test_events_replay_frames_are_sse_framed(hard_proof_job_service): assert f"id: {seq}".encode() in frame, f"replayed frame missing id: {frame!r}" -async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service): +async def test_executor_stop_applies_terminal_reconcile(real_services_job_service): """``executor.stop()`` lets an in-flight stopped job's reconcile land. Drives the REAL runner on the executor. The job blocks mid-run; a STOP signal @@ -440,7 +440,7 @@ async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service): from langflow.services.background_execution.executor import InProcessExecutor - job_service = hard_proof_job_service + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -476,7 +476,7 @@ async def test_executor_stop_applies_terminal_reconcile(hard_proof_job_service): assert job.status == JobStatus.CANCELLED -async def test_stop_poll_only_on_durable_frames(hard_proof_job_service): +async def test_stop_poll_only_on_durable_frames(real_services_job_service): """The runner polls the STOP signal only on DURABLE frames, not every token. Polling ``unconsumed_signals`` (a DB read) on every ephemeral token is wasted @@ -486,7 +486,7 @@ async def test_stop_poll_only_on_durable_frames(hard_proof_job_service): durable frames, and assert the poll count tracks the durable frames, not the token flood. Real SQLite and Postgres. """ - job_service = hard_proof_job_service + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -521,7 +521,7 @@ async def test_stop_poll_only_on_durable_frames(hard_proof_job_service): ) -async def test_stop_signal_is_marked_consumed(hard_proof_job_service): +async def test_stop_signal_is_marked_consumed(real_services_job_service): """When the runner acts on a STOP, the signal row is stamped ``consumed_at``. Otherwise the execution_signals table grows unbounded and, worse, a @@ -530,7 +530,7 @@ async def test_stop_signal_is_marked_consumed(hard_proof_job_service): (2) a fresh run of the SAME job_id afterwards does NOT instantly cancel — it completes, because the stale STOP was consumed. Real SQLite and Postgres. """ - job_service = hard_proof_job_service + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) await job_service.write_signal(job_id, SignalType.STOP) @@ -579,7 +579,7 @@ def _side_effect_factory(*, request, **_kwargs): # noqa: ARG001 return _source -async def test_concurrent_sweep_runs_queued_job_exactly_once(hard_proof_job_service): +async def test_concurrent_sweep_runs_queued_job_exactly_once(real_services_job_service): """Two startup sweepers sharing one DB must run a QUEUED job EXACTLY once. Models two uvicorn workers booting against the same database, each calling @@ -594,7 +594,7 @@ async def test_concurrent_sweep_runs_queued_job_exactly_once(hard_proof_job_serv from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service user_id = uuid4() flow_id = uuid4() job_id = uuid4() diff --git a/src/backend/tests/unit/background_execution/test_hard_proof_ci_wiring.py b/src/backend/tests/unit/background_execution/test_hard_proof_ci_wiring.py deleted file mode 100644 index f4306aeaae..0000000000 --- a/src/backend/tests/unit/background_execution/test_hard_proof_ci_wiring.py +++ /dev/null @@ -1,43 +0,0 @@ -"""Verify CI wires a real redis:7 service and runs the hard-proof tier.""" - -from __future__ import annotations - -from pathlib import Path - -import pytest -import yaml - -_REPO_ROOT = Path(__file__).resolve().parents[5] -_WORKFLOW = _REPO_ROOT / ".github" / "workflows" / "migration-validation.yml" - - -@pytest.fixture -def workflow() -> dict: - assert _WORKFLOW.exists(), f"workflow not found at {_WORKFLOW}" - return yaml.safe_load(_WORKFLOW.read_text(encoding="utf-8")) - - -def _hard_proof_job(workflow: dict) -> dict | None: - return next((j for name, j in workflow["jobs"].items() if "hard" in name and "proof" in name), None) - - -@pytest.mark.no_blockbuster -def test_hard_proof_job_exists(workflow: dict) -> None: - assert _hard_proof_job(workflow) is not None, f"no hard-proof job; jobs: {list(workflow['jobs'])}" - - -@pytest.mark.no_blockbuster -def test_hard_proof_job_has_postgres_and_redis_services(workflow: dict) -> None: - hard_proof = _hard_proof_job(workflow) - assert hard_proof is not None - images = {svc["image"] for svc in hard_proof["services"].values()} - assert any(img.startswith("postgres:") for img in images), images - assert any(img.startswith("redis:") for img in images), images - - -@pytest.mark.no_blockbuster -def test_hard_proof_job_sets_both_test_uris() -> None: - text = _WORKFLOW.read_text(encoding="utf-8") - assert "LANGFLOW_TEST_DATABASE_URI" in text - assert "LANGFLOW_TEST_REDIS_URL" in text - assert "-m hard_proof" in text or "hard_proof_tests" in text diff --git a/src/backend/tests/unit/background_execution/test_hard_proof_make_target.py b/src/backend/tests/unit/background_execution/test_hard_proof_make_target.py deleted file mode 100644 index d66c7ae820..0000000000 --- a/src/backend/tests/unit/background_execution/test_hard_proof_make_target.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Verify the Makefile exposes a hard_proof_tests target wired to -m hard_proof.""" - -from __future__ import annotations - -from pathlib import Path - -import pytest - -_REPO_ROOT = Path(__file__).resolve().parents[5] -_MAKEFILE = _REPO_ROOT / "Makefile" - - -@pytest.mark.no_blockbuster -def test_makefile_has_hard_proof_target() -> None: - assert _MAKEFILE.exists(), f"Makefile not found at {_MAKEFILE}" - text = _MAKEFILE.read_text(encoding="utf-8") - assert "hard_proof_tests:" in text, "missing hard_proof_tests make target" - target_block = text.split("hard_proof_tests:", 1)[1].split("\n\n", 1)[0] - assert "-m hard_proof" in target_block, "hard_proof_tests target must select the hard_proof marker" - assert "src/backend/tests/unit" in target_block, "target must run the backend unit tree" diff --git a/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py b/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py index f00a4333c4..35cf05ef52 100644 --- a/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py +++ b/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py @@ -33,7 +33,7 @@ from langflow.services.background_execution.live_bus import InMemoryLiveBus from langflow.services.background_execution.runner import JobRunner from langflow.services.database.models.jobs.model import JobStatus -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services # --------------------------------------------------------------------------- # @@ -146,12 +146,12 @@ def _echo_factory(*, request, **_kwargs): return _source -async def test_durability_delta_off_loses_job_on_restart_on_recovers(hard_proof_job_service): +async def test_durability_delta_off_loses_job_on_restart_on_recovers(real_services_job_service): """Durability OFF: a restart finds NO recoverable row -> job lost. ON: recovered + re-run.""" from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service # --- OFF: the in-memory-only path (pre-fix). The submit never wrote a durable # QUEUED row — the job lived only in the dead process's executor queue. A @@ -229,9 +229,9 @@ def _runner(job_service, job_id, source): return JobRunner(job_service=job_service, live_bus=InMemoryLiveBus(), adapter=adapter, frame_source=source) -async def test_sweep_claim_delta_off_double_runs_on_exactly_once(hard_proof_job_service): +async def test_sweep_claim_delta_off_double_runs_on_exactly_once(real_services_job_service): """Sweep claim OFF: two sweepers both run a QUEUED job (double). ON: exactly once.""" - job_service = hard_proof_job_service + job_service = real_services_job_service # --- OFF: the pre-fix unconditional re-enqueue. Two sweepers each run the job # because neither claims it first. We reproduce by running the runner twice diff --git a/src/backend/tests/unit/background_execution/test_last_event_id_contract.py b/src/backend/tests/unit/background_execution/test_last_event_id_contract.py index 339adbf996..b44aefa665 100644 --- a/src/backend/tests/unit/background_execution/test_last_event_id_contract.py +++ b/src/backend/tests/unit/background_execution/test_last_event_id_contract.py @@ -30,7 +30,7 @@ from langflow.services.background_execution.runner import JobRunner if TYPE_CHECKING: from collections.abc import AsyncIterator -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services class _StubUser: @@ -92,7 +92,7 @@ async def _run_token_flow(job_service, job_id, bus) -> None: await runner.run(job_id=job_id, source_kwargs={}) -async def test_live_id_matches_durable_seq_default(hard_proof_job_service): +async def test_live_id_matches_durable_seq_default(real_services_job_service): """The live id: a client sees == the durable seq the replay resumes after. Drives the real runner (default in-memory bus), captures the live frames, then @@ -100,7 +100,7 @@ async def test_live_id_matches_durable_seq_default(hard_proof_job_service): Pre-fix the live id is the stream-seq (0,1,4,5) while the durable seq is (1,2,3,4), so they diverge and a Last-Event-ID resume is wrong. """ - job_service = hard_proof_job_service + job_service = real_services_job_service user_id = uuid4() job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=user_id) @@ -134,7 +134,7 @@ async def test_live_id_matches_durable_seq_default(hard_proof_job_service): ) -async def test_reattach_no_gap_no_dup_default(hard_proof_job_service): +async def test_reattach_no_gap_no_dup_default(real_services_job_service): """Mid-run Last-Event-ID resume starts exactly at the next milestone. Captures the live id of the ``vertices_sorted`` milestone, reconnects with it @@ -148,7 +148,7 @@ async def test_reattach_no_gap_no_dup_default(hard_proof_job_service): from langflow.services.background_execution.service import BackgroundExecutionService from langflow.services.deps import get_settings_service - job_service = hard_proof_job_service + job_service = real_services_job_service user_id = uuid4() job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=user_id) diff --git a/src/backend/tests/unit/background_execution/test_multi_worker_boot_hard_proof.py b/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py similarity index 95% rename from src/backend/tests/unit/background_execution/test_multi_worker_boot_hard_proof.py rename to src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py index 5e5621f692..35fb3af161 100644 --- a/src/backend/tests/unit/background_execution/test_multi_worker_boot_hard_proof.py +++ b/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py @@ -7,7 +7,7 @@ inject a phantom terminal event into A's live durable log, and race A's real terminal write. With lease+heartbeat liveness the sweep only reconciles GENUINELY orphaned rows (stale/absent heartbeat), so B leaves A's live job alone. -Real SQLite AND real Postgres via the hard-proof fixture. +Real SQLite AND real Postgres via the real-service fixture. """ from __future__ import annotations @@ -26,15 +26,15 @@ from langflow.services.database.models.jobs.model import JobStatus if TYPE_CHECKING: from collections.abc import AsyncIterator -pytestmark = pytest.mark.hard_proof +pytestmark = pytest.mark.real_services def _frame(event_type: str, data: dict) -> tuple[bytes, str]: return (json.dumps({"event": event_type, "data": data}).encode("utf-8"), event_type) -async def test_booting_worker_sweep_spares_live_heartbeated_job(hard_proof_job_service): - job_service = hard_proof_job_service +async def test_booting_worker_sweep_spares_live_heartbeated_job(real_services_job_service): + job_service = real_services_job_service job_id = uuid4() await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) diff --git a/src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py b/src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py new file mode 100644 index 0000000000..b50698abd3 --- /dev/null +++ b/src/backend/tests/unit/background_execution/test_real_services_ci_wiring.py @@ -0,0 +1,43 @@ +"""Verify CI wires a real redis:7 service and runs the real-service tier.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +_REPO_ROOT = Path(__file__).resolve().parents[5] +_WORKFLOW = _REPO_ROOT / ".github" / "workflows" / "migration-validation.yml" + + +@pytest.fixture +def workflow() -> dict: + assert _WORKFLOW.exists(), f"workflow not found at {_WORKFLOW}" + return yaml.safe_load(_WORKFLOW.read_text(encoding="utf-8")) + + +def _real_services_job(workflow: dict) -> dict | None: + return next((j for name, j in workflow["jobs"].items() if "real" in name and "service" in name), None) + + +@pytest.mark.no_blockbuster +def test_real_services_job_exists(workflow: dict) -> None: + assert _real_services_job(workflow) is not None, f"no real-service job; jobs: {list(workflow['jobs'])}" + + +@pytest.mark.no_blockbuster +def test_real_services_job_has_postgres_and_redis_services(workflow: dict) -> None: + real_services = _real_services_job(workflow) + assert real_services is not None + images = {svc["image"] for svc in real_services["services"].values()} + assert any(img.startswith("postgres:") for img in images), images + assert any(img.startswith("redis:") for img in images), images + + +@pytest.mark.no_blockbuster +def test_real_services_job_sets_both_test_uris() -> None: + text = _WORKFLOW.read_text(encoding="utf-8") + assert "LANGFLOW_TEST_DATABASE_URI" in text + assert "LANGFLOW_TEST_REDIS_URL" in text + assert "-m real_services" in text or "real_services_tests" in text diff --git a/src/backend/tests/unit/background_execution/test_hard_proof_harness.py b/src/backend/tests/unit/background_execution/test_real_services_harness.py similarity index 51% rename from src/backend/tests/unit/background_execution/test_hard_proof_harness.py rename to src/backend/tests/unit/background_execution/test_real_services_harness.py index de311ab280..d2fc9346c9 100644 --- a/src/backend/tests/unit/background_execution/test_hard_proof_harness.py +++ b/src/backend/tests/unit/background_execution/test_real_services_harness.py @@ -1,4 +1,4 @@ -"""Phase 0 harness tests: the hard_proof marker + real-instance fixtures. +"""Phase 0 harness tests: the real_services marker + real-instance fixtures. These prove the harness itself works before any background-execution code exists: the marker is registered, the DB fixture backs a real engine on sqlite and @@ -15,20 +15,20 @@ from sqlalchemy.ext.asyncio import create_async_engine @pytest.mark.no_blockbuster -def test_hard_proof_marker_is_registered(request: pytest.FixtureRequest) -> None: - """The hard_proof marker must be registered so --strict-markers accepts it.""" +def test_real_services_marker_is_registered(request: pytest.FixtureRequest) -> None: + """The real_services marker must be registered so --strict-markers accepts it.""" ini_markers = request.config.getini("markers") names = {line.split(":", 1)[0].strip() for line in ini_markers} - assert "hard_proof" in names, f"hard_proof not registered; have: {sorted(names)}" + assert "real_services" in names, f"real_services not registered; have: {sorted(names)}" -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_hard_proof_db_url_yields_real_engine(hard_proof_db_url: str) -> None: - """hard_proof_db_url must produce a URL backing a real, queryable engine.""" - if "postgresql" in hard_proof_db_url: +async def test_real_services_db_url_yields_real_engine(real_services_db_url: str) -> None: + """real_services_db_url must produce a URL backing a real, queryable engine.""" + if "postgresql" in real_services_db_url: assert os.environ.get("LANGFLOW_TEST_DATABASE_URI"), "postgres param ran without URI set" - engine = create_async_engine(hard_proof_db_url) + engine = create_async_engine(real_services_db_url) try: async with engine.connect() as conn: result = await conn.execute(text("SELECT 1")) @@ -37,17 +37,17 @@ async def test_hard_proof_db_url_yields_real_engine(hard_proof_db_url: str) -> N await engine.dispose() -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_hard_proof_redis_url_pings_real_server(hard_proof_redis_url: str) -> None: - """hard_proof_redis_url must back a reachable, real Redis server.""" +async def test_real_services_redis_url_pings_real_server(real_services_redis_url: str) -> None: + """real_services_redis_url must back a reachable, real Redis server.""" from redis.asyncio import StrictRedis assert os.environ.get("LANGFLOW_TEST_REDIS_URL"), "redis fixture ran without URL set" - client = StrictRedis.from_url(hard_proof_redis_url) + client = StrictRedis.from_url(real_services_redis_url) try: assert await client.ping() is True - await client.set("hard_proof:probe", "1") - assert await client.get("hard_proof:probe") == b"1" + await client.set("real_services:probe", "1") + assert await client.get("real_services:probe") == b"1" finally: await client.aclose() diff --git a/src/backend/tests/unit/background_execution/test_real_services_make_target.py b/src/backend/tests/unit/background_execution/test_real_services_make_target.py new file mode 100644 index 0000000000..8770dc935e --- /dev/null +++ b/src/backend/tests/unit/background_execution/test_real_services_make_target.py @@ -0,0 +1,20 @@ +"""Verify the Makefile exposes a real_services_tests target wired to -m real_services.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +_REPO_ROOT = Path(__file__).resolve().parents[5] +_MAKEFILE = _REPO_ROOT / "Makefile" + + +@pytest.mark.no_blockbuster +def test_makefile_has_real_services_target() -> None: + assert _MAKEFILE.exists(), f"Makefile not found at {_MAKEFILE}" + text = _MAKEFILE.read_text(encoding="utf-8") + assert "real_services_tests:" in text, "missing real_services_tests make target" + target_block = text.split("real_services_tests:", 1)[1].split("\n\n", 1)[0] + assert "-m real_services" in target_block, "real_services_tests target must select the real_services marker" + assert "src/backend/tests/unit" in target_block, "target must run the backend unit tree" diff --git a/src/backend/tests/unit/background_execution/test_store_hard_proof.py b/src/backend/tests/unit/background_execution/test_store_real_services.py similarity index 80% rename from src/backend/tests/unit/background_execution/test_store_hard_proof.py rename to src/backend/tests/unit/background_execution/test_store_real_services.py index 596260b305..3b22042d63 100644 --- a/src/backend/tests/unit/background_execution/test_store_hard_proof.py +++ b/src/backend/tests/unit/background_execution/test_store_real_services.py @@ -1,7 +1,7 @@ -"""Hard-proof store tests: JobService methods against REAL SQLite and REAL Postgres. +"""Real-service 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 +``real_services_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 @@ -17,10 +17,10 @@ import pytest from langflow.services.database.models.jobs.model import JobStatus, SignalType -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_set_result_and_error_persist(hard_proof_job_service) -> None: - service = hard_proof_job_service +async def test_set_result_and_error_persist(real_services_job_service) -> None: + service = real_services_job_service job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -32,10 +32,10 @@ async def test_set_result_and_error_persist(hard_proof_job_service) -> None: assert fetched.error == {"type": "boom", "message": "x"} -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_append_and_read_events_ordered(hard_proof_job_service) -> None: - service = hard_proof_job_service +async def test_append_and_read_events_ordered(real_services_job_service) -> None: + service = real_services_job_service job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -48,10 +48,10 @@ async def test_append_and_read_events_ordered(hard_proof_job_service) -> None: assert tail[0].payload == {"i": 2} -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None: - service = hard_proof_job_service +async def test_append_event_seq_is_per_job(real_services_job_service) -> None: + service = real_services_job_service job_a = uuid4() job_b = uuid4() await service.create_job(job_id=job_a, flow_id=uuid4(), user_id=uuid4()) @@ -62,15 +62,15 @@ async def test_append_event_seq_is_per_job(hard_proof_job_service) -> None: assert await service.append_event(job_a, "x", {}) == 2 -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_append_event_concurrent_appends_all_land_gap_free(hard_proof_job_service) -> None: +async def test_append_event_concurrent_appends_all_land_gap_free(real_services_job_service) -> None: """Concurrent appends to one job must ALL land, gap-free (retry on UNIQUE collision). Regression for the seq race: SELECT max(seq)+1 then INSERT collides under contention; without retry the losers' events were silently dropped. """ - service = hard_proof_job_service + service = real_services_job_service job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -84,10 +84,10 @@ async def test_append_event_concurrent_appends_all_land_gap_free(hard_proof_job_ assert len(events) == n -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_write_and_read_signals(hard_proof_job_service) -> None: - service = hard_proof_job_service +async def test_write_and_read_signals(real_services_job_service) -> None: + service = real_services_job_service job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) @@ -101,10 +101,10 @@ async def test_write_and_read_signals(hard_proof_job_service) -> None: assert pending[0].signal_type == SignalType.STOP -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> None: - service = hard_proof_job_service +async def test_sweep_orphans_reconciles_in_progress(real_services_job_service) -> None: + service = real_services_job_service orphan = uuid4() queued = uuid4() await service.create_job(job_id=orphan, flow_id=uuid4(), user_id=uuid4()) @@ -128,11 +128,11 @@ async def test_sweep_orphans_reconciles_in_progress(hard_proof_job_service) -> N assert queued_job.status == JobStatus.QUEUED -@pytest.mark.hard_proof +@pytest.mark.real_services @pytest.mark.no_blockbuster -async def test_get_jobs_by_flow_id_orders_by_created_timestamp(hard_proof_job_service) -> None: +async def test_get_jobs_by_flow_id_orders_by_created_timestamp(real_services_job_service) -> None: """Regression on real engines: ordering by created_timestamp must not raise.""" - service = hard_proof_job_service + service = real_services_job_service flow_id = uuid4() user_id = uuid4() first = uuid4()