diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index aaee59549..51653e2f3 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -84,6 +84,93 @@ jobs: echo "" >> $GITHUB_STEP_SUMMARY echo "Test Status: ${{ job.status }}" >> $GITHUB_STEP_SUMMARY + postgres-contracts: + name: PostgreSQL Contract Tests + runs-on: ubuntu-latest + timeout-minutes: 15 + # Only this disposable, loopback-bound CI service uses trust authentication. + # Admission fixtures create a separate unprivileged role with no password; + # the literal postgres@ URL lets that fixture switch to the runtime role. + env: + TEST_POSTGRES_URL: postgresql+asyncpg://postgres@localhost:5432/langbot_test + LANGBOT_ADMISSION_TEST_URL: postgresql+asyncpg://postgres@localhost:5432/langbot_admission_test + services: + postgres: + image: pgvector/pgvector:pg16 + env: + POSTGRES_USER: postgres + POSTGRES_DB: langbot_test + POSTGRES_HOST_AUTH_METHOD: trust + ports: + - '127.0.0.1:5432:5432' + options: >- + --health-cmd="pg_isready -U postgres -d langbot_test" + --health-interval=5s + --health-timeout=5s + --health-retries=5 + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install uv + uses: astral-sh/setup-uv@v4 + + - name: Install dependencies + run: uv sync --dev --frozen + + - name: Create isolated admission test database + run: >- + docker exec "${{ job.services.postgres.id }}" + psql -U postgres -d postgres -v ON_ERROR_STOP=1 + -c 'CREATE DATABASE langbot_admission_test;' + + - name: Run required PostgreSQL contracts + run: | + uv run pytest \ + tests/integration/persistence/test_pipeline_admission_postgres.py \ + tests/integration/persistence/test_rag_document_identity.py \ + tests/integration/persistence/test_rag_interrupted_ingestion.py \ + tests/integration/persistence/test_migration_branch_convergence.py \ + tests/integration/persistence/test_runner_timestamps_postgres.py \ + -k postgres -q --tb=short -rs --durations=10 \ + --junitxml=postgres-contract-results.xml + python - <<'PY' + from collections import Counter + import xml.etree.ElementTree as ET + + cases = list(ET.parse('postgres-contract-results.xml').getroot().iter('testcase')) + expected = { + 'test_pipeline_admission_postgres': 27, + 'test_rag_document_identity': 34, + 'test_rag_interrupted_ingestion': 22, + 'test_migration_branch_convergence': 7, + 'test_runner_timestamps_postgres': 32, + } + counts = Counter(case.get('classname', '').rsplit('.', 1)[-1] for case in cases) + for module, minimum in expected.items(): + assert counts[module] >= minimum, f'{module}: expected at least {minimum} cases, got {counts[module]}' + unsuccessful = [ + case.get('name') for case in cases + if any(case.find(tag) is not None for tag in ('skipped', 'failure', 'error')) + ] + assert not unsuccessful, f'PostgreSQL contracts must pass without skips: {unsuccessful}' + print(f'PostgreSQL contracts: {len(cases)} passed, 0 skipped; {dict(counts)}') + PY + + - name: Upload PostgreSQL contract results + if: always() + uses: actions/upload-artifact@v4 + with: + name: postgres-contract-results + path: postgres-contract-results.xml + if-no-files-found: ignore + e2e: name: E2E Startup Tests runs-on: ubuntu-latest @@ -148,9 +235,16 @@ jobs: coverage: name: Coverage Gate runs-on: ubuntu-latest - needs: [test, integration] + needs: [test, integration, postgres-contracts] + if: always() steps: + - name: Verify required test jobs + run: | + test "${{ needs.test.result }}" = success + test "${{ needs.integration.result }}" = success + test "${{ needs.postgres-contracts.result }}" = success + - name: Checkout code uses: actions/checkout@v4 diff --git a/docs/POSTGRES_CONTRACT_TESTS.md b/docs/POSTGRES_CONTRACT_TESTS.md new file mode 100644 index 000000000..5286da109 --- /dev/null +++ b/docs/POSTGRES_CONTRACT_TESTS.md @@ -0,0 +1,81 @@ +# Disposable PostgreSQL contract tests + +The fast integration suite deliberately runs without an external database. At +the current baseline, 83 cases in three persistence modules require PostgreSQL: + +| Module | PostgreSQL cases | Contract | +| --- | ---: | --- | +| `test_pipeline_admission_postgres.py` | 27 | Real row-lock admission, tenant isolation, rollback, cancellation and commit reconciliation | +| `test_rag_document_identity.py` | 34 | Persisted engine document identity, scoped deletion and migration convergence | +| `test_rag_interrupted_ingestion.py` | 22 | Interrupted/cancelled ingestion, delayed SDK writes, recovery and retained uploads | + +Another 39 PostgreSQL cases are marked slow and deselected by Fast Integration +Tests rather than reported as skipped: + +| Module | PostgreSQL cases | Contract | +| --- | ---: | --- | +| `test_migration_branch_convergence.py` | 7 | Real historical/populated upgrade branches, certification backfill and merge-only downgrade | +| `test_runner_timestamps_postgres.py` | 32 | Non-superuser timestamp/lease/retention contracts against both metadata and published-migration schemas | + +The PostgreSQL Contract Tests job in `.github/workflows/run-tests.yml` provisions +a dedicated `pgvector/pgvector:pg16` service, sets both required URL variables, +and runs all 122 PostgreSQL cases. Admission uses a separate database from RAG, +convergence and timestamp tests: its populated `public` schema must not affect +other fixtures' schema discovery through their `schema,public` search path. +Its JUnit check requires each module's baseline +case count and rejects skips, failures and errors. Results are uploaded even when +the job fails. SQLite variants continue running in Fast Integration Tests. + +## Run locally + +Use a new disposable database. The admission fixture drops and recreates the +entire `public` schema and creates a test runtime role. Never use a development, +staging or production database, or a shared database containing useful data. + +For example, start an isolated container bound only to the local loopback address: + +```bash +docker run --rm -d --name langbot-contract-pg \ + -e POSTGRES_USER=postgres -e POSTGRES_DB=langbot_test \ + -e POSTGRES_HOST_AUTH_METHOD=trust \ + -p 127.0.0.1:55432:5432 pgvector/pgvector:pg16 +docker exec langbot-contract-pg pg_isready -U postgres -d langbot_test +``` + +Wait until `pg_isready` reports that the server is accepting connections, then: + +```bash +docker exec langbot-contract-pg psql -U postgres -d postgres \ + -v ON_ERROR_STOP=1 -c 'CREATE DATABASE langbot_admission_test;' +uv sync --dev --frozen +export TEST_POSTGRES_URL=postgresql+asyncpg://postgres@localhost:55432/langbot_test +export LANGBOT_ADMISSION_TEST_URL=postgresql+asyncpg://postgres@localhost:55432/langbot_admission_test +uv run pytest \ + tests/integration/persistence/test_pipeline_admission_postgres.py \ + tests/integration/persistence/test_rag_document_identity.py \ + tests/integration/persistence/test_rag_interrupted_ingestion.py \ + tests/integration/persistence/test_migration_branch_convergence.py \ + tests/integration/persistence/test_runner_timestamps_postgres.py \ + -k postgres -q --tb=short -rs --durations=10 \ + --junitxml=postgres-contract-results.xml +docker stop langbot-contract-pg +``` + +Trust authentication is limited to this throwaway, loopback-bound service. The +admission fixture needs the literal `postgres@` URL to derive a separate +unprivileged runtime-role connection. RAG fixtures create and clean up their own +unique schemas, as do convergence fixtures. Timestamp fixtures create a unique +non-superuser, non-RLS-bypass role and owned schema for each case and remove both +afterward. The admin connection therefore needs role/schema creation privileges. +These settings do not belong in an application configuration. + +## Other skip categories + +- The monitoring PostgreSQL unit regression is separately selected by the + Migrations (PostgreSQL) job in `.github/workflows/test-migrations.yml` +- For plugin identity, compiled CLI and embedded SeekDB prerequisites, see + [the dependency contract setup in PR #2636](https://github.com/langbot-app/LangBot/pull/2636). + +These remaining prerequisites should be reported explicitly rather than treated +as passing tests. The four chat-handler logging tests now initialize the normal +application import graph before importing the handler and no longer skip. diff --git a/tests/README.md b/tests/README.md index 3a110b9dc..f3acd40b3 100644 --- a/tests/README.md +++ b/tests/README.md @@ -291,6 +291,11 @@ Note: PostgreSQL tests are **not** included in fast integration gate because the - Are marked with `@pytest.mark.slow` - Need `TEST_POSTGRES_URL` environment variable +A separate **PostgreSQL Contract Tests** job provisions a disposable database for +pipeline admission, RAG recovery, migration convergence, and timestamp/lease contracts. +See [the PostgreSQL contract guide](../docs/POSTGRES_CONTRACT_TESTS.md) for scope, +local commands, and remaining optional prerequisites. + CI workflow `.github/workflows/test-migrations.yml` runs: - SQLite tests in `test-migrations-sqlite` job (fast, no external services) - PostgreSQL tests in `test-migrations-postgres` job (uses PostgreSQL service container) diff --git a/tests/integration/persistence/test_runner_timestamps_postgres.py b/tests/integration/persistence/test_runner_timestamps_postgres.py index 445d3b3e0..6cc988b7b 100644 --- a/tests/integration/persistence/test_runner_timestamps_postgres.py +++ b/tests/integration/persistence/test_runner_timestamps_postgres.py @@ -34,7 +34,11 @@ TABLES = [AgentRun.__table__, AgentRunEvent.__table__, AgentRuntime.__table__, E def _published_schema(connection): with Operations.context(MigrationContext.configure(connection)): - for revision in ('58846a8d7a81_add_event_log_and_transcript_tables', '8d3a1f2c4b6e_add_agent_run_ledger'): + for revision in ( + '58846a8d7a81_add_event_log_and_transcript_tables', + '8d3a1f2c4b6e_add_agent_run_ledger', + '0036_execution_host_leases', + ): importlib.import_module(f'langbot.pkg.persistence.alembic.versions.{revision}').upgrade() @@ -119,6 +123,7 @@ async def test_run_lifecycle_deadline_events_and_stats(pg_engine): bot_id=None, event_type='message.received', source='debug', + data={}, delivery=SimpleNamespace(model_dump=lambda **kwargs: {}), ), binding=SimpleNamespace(binding_id='binding', agent_id='agent', processor_id='agent', processor_type='agent'), diff --git a/tests/integration/pipeline/test_full_flow.py b/tests/integration/pipeline/test_full_flow.py index 1a51dd9e4..6b2a0cf72 100644 --- a/tests/integration/pipeline/test_full_flow.py +++ b/tests/integration/pipeline/test_full_flow.py @@ -2,7 +2,8 @@ Pipeline full-flow integration tests. Tests real pipeline stages with fake runner/provider. -Validates message processing through PreProcessor, Processor, and SendResponseBackStage. +Validates RuntimePipeline dispatch through PreProcessor, Processor, ResponseWrapper, +and SendResponseBackStage, with external services replaced by deterministic doubles. Uses RuntimePipeline directly (not PipelineManager) to avoid DB dependency. @@ -13,6 +14,7 @@ from __future__ import annotations import pytest import asyncio +from types import SimpleNamespace from unittest.mock import AsyncMock, Mock from tests.factories import FakeApp, text_query, mock_platform_adapter @@ -23,76 +25,20 @@ from tests.factories.platform import FakePlatform pytestmark = pytest.mark.integration -# ============== FIXTURE FOR SYS.MODULES ISOLATION ============== +# ============== NORMAL APPLICATION IMPORT BOOTSTRAP ============== @pytest.fixture(scope='module') -def mock_circular_import_chain(): +def load_pipeline_modules(): + """Initialize the real import graph before importing individual stages. + + This matches normal startup and the pipeline unit-test bootstrap. No + sys.modules/package attributes are replaced, so later tests keep real + module identities and the complete stage registry. """ - Break circular import chain for pipeline modules using isolated_sys_modules. + from importlib import import_module - Chain: pipeline → core.app → http_controller → groups/plugins - - We mock minimal modules to allow importing RuntimePipeline, StageInstContainer, - and stage classes without triggering full application initialization. - - After mocking, we import the stage modules so decorators register them. - """ - from tests.utils.import_isolation import isolated_sys_modules, MockLifecycleControlScope - - # Mock core.entities with LifecycleControlScope enum - mock_core_entities = Mock() - mock_core_entities.LifecycleControlScope = MockLifecycleControlScope - - # Mock core.app - Application class is referenced but not instantiated - mock_core_app = Mock() - - # Mock utils.importutil to avoid unrelated import-time registrations. - mock_importutil = Mock() - mock_importutil.import_modules_in_pkg = lambda pkg: None - mock_importutil.import_modules_in_pkgs = lambda pkgs: None - - # Modules to clear (force re-import after mocking) - clear = [ - 'langbot.pkg.pipeline.stage', - 'langbot.pkg.pipeline.entities', - 'langbot.pkg.pipeline.pipelinemgr', - 'langbot.pkg.pipeline.preproc.preproc', - 'langbot.pkg.pipeline.process.process', - 'langbot.pkg.pipeline.process.handler', - 'langbot.pkg.pipeline.process.handlers.chat', - 'langbot.pkg.pipeline.process.handlers.command', - 'langbot.pkg.pipeline.respback.respback', - ] - - with isolated_sys_modules( - mocks={ - 'langbot.pkg.core.entities': mock_core_entities, - 'langbot.pkg.core.app': mock_core_app, - 'langbot.pkg.utils.importutil': mock_importutil, - 'langbot.pkg.pipeline.controller': Mock(), - 'langbot.pkg.pipeline.pipelinemgr': Mock(), - }, - clear=clear, - ): - # Import stage modules AFTER clearing so decorators register them - from importlib import import_module - - # Import stage base first - import_module('langbot.pkg.pipeline.stage') - - # Import entities - import_module('langbot.pkg.pipeline.entities') - - # Import specific stages to register them - import_module('langbot.pkg.pipeline.preproc.preproc') - import_module('langbot.pkg.pipeline.process.process') - import_module('langbot.pkg.pipeline.respback.respback') - - # Import pipelinemgr for RuntimePipeline - import_module('langbot.pkg.pipeline.pipelinemgr') - - yield + import_module('langbot.pkg.pipeline.pipelinemgr') # ============== FAKE RUNNER ============== @@ -172,35 +118,18 @@ def pipeline_app(): """ app = FakeApp() - # Session/conversation mocks for PreProcessor - mock_session = Mock() - mock_session.launcher_type = Mock() - mock_session.launcher_type.value = 'person' - mock_session.launcher_id = 12345 - mock_session.sender_id = 12345 - mock_session.use_prompt_name = 'default' - mock_session.using_conversation = None + # Real SDK entities keep plugin event validation meaningful. + from langbot_plugin.api.entities.builtin.provider.session import Conversation, Session, LauncherTypes + from langbot_plugin.api.entities.builtin.provider.prompt import Prompt - # Create a simple class to mimic Prompt behavior - class MockPrompt: - def __init__(self, name, messages): - self.name = name - self.messages = messages - - def copy(self): - return MockPrompt(self.name, list(self.messages)) - - # Create real lists for messages - prompt_messages_list = [] - messages_list = [] - - mock_prompt = MockPrompt('default', prompt_messages_list) - mock_conversation = Mock() - mock_conversation.prompt = mock_prompt - mock_conversation.messages = messages_list - mock_conversation.uuid = 'test-conversation-uuid' - mock_conversation.update_time = None - mock_conversation.create_time = None + mock_session = Session(launcher_type=LauncherTypes.PERSON, launcher_id=12345, sender_id=12345) + mock_conversation = Conversation( + prompt=Prompt(name='default', messages=[]), + messages=[], + pipeline_uuid='test-pipeline-uuid', + bot_uuid='test-bot-uuid', + uuid='test-conversation-uuid', + ) async def get_scoped_session(query): context = query._execution_context @@ -223,6 +152,18 @@ def pipeline_app(): # Tool manager mock app.tool_mgr.get_all_tools = AsyncMock(return_value=[]) + app.tool_mgr.get_resolved_tool_catalog = AsyncMock(return_value=[]) + app.workspace_service = AsyncMock() + app.workspace_service.get_execution_binding.return_value = SimpleNamespace( + instance_uuid='test-instance', + workspace_uuid='test-workspace', + placement_generation=1, + ) + app.query_pool.remove_query = AsyncMock(return_value=True) + app.bot_service = AsyncMock() + app.bot_service.get_bot.return_value = {'name': 'Test bot'} + app.monitoring_service = AsyncMock() + app.monitoring_service.record_message.return_value = 'test-monitoring-message' # Telemetry mock (required by ChatMessageHandler) app.telemetry = Mock() @@ -251,7 +192,10 @@ def set_fake_runner(pipeline_app): orchestrator = Mock() orchestrator.try_claim_steering_from_query = AsyncMock(return_value=False) + orchestrator.observed_queries = [] + async def run_from_query(query): + orchestrator.observed_queries.append(query) async for result in runner.run(query): yield result @@ -283,6 +227,7 @@ def create_minimal_pipeline_config(): 'misc': { 'at-sender': False, 'quote-origin': False, + 'track-function-calls': False, 'exception-handling': 'show-hint', 'failure-hint': 'Request failed.', }, @@ -320,7 +265,7 @@ async def collect_processor_results(processor, query, stage_name): # ============== TESTS ============== -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.mark.usefixtures('load_pipeline_modules') class TestPipelineStageChainReal: """Tests for real pipeline stage chain.""" @@ -348,7 +293,7 @@ class TestPipelineStageChainReal: assert 'SendResponseBackStage' in stage.preregistered_stages -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.mark.usefixtures('load_pipeline_modules') class TestPreProcessorStage: """Tests for PreProcessor stage alone.""" @@ -410,7 +355,7 @@ class TestPreProcessorStage: assert result.new_query.user_message.role == 'user' -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.mark.usefixtures('load_pipeline_modules') class TestProcessorStage: """Tests for MessageProcessor stage.""" @@ -517,7 +462,7 @@ class TestProcessorStage: assert query.resp_messages[0] == reply_chain -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.mark.usefixtures('load_pipeline_modules') class TestRunnerExceptionFlow: """Tests for runner exception handling.""" @@ -638,7 +583,7 @@ class TestRunnerExceptionFlow: assert results[0].user_notice is None -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.mark.usefixtures('load_pipeline_modules') class TestSendResponseBackStage: """Tests for SendResponseBackStage.""" @@ -768,136 +713,220 @@ class TestSendResponseBackStage: pipeline_app.plugin_connector.notify_plugin_diagnostic.assert_not_called() -@pytest.mark.usefixtures('mock_circular_import_chain') +@pytest.fixture +async def runtime_pipeline_factory(pipeline_app, load_pipeline_modules): + """Use real dispatch and stages; fake only the surrounding services.""" + from langbot.pkg.pipeline import pipelinemgr + from langbot.pkg.pipeline.preproc.preproc import PreProcessor + from langbot.pkg.pipeline.process.process import Processor + from langbot.pkg.pipeline.wrapper.wrapper import ResponseWrapper + from langbot.pkg.pipeline.respback.respback import SendResponseBackStage + + async def create(query): + context = query._execution_context + query.instance_uuid = context.instance_uuid + query.workspace_uuid = context.workspace_uuid + query.placement_generation = context.placement_generation + query.resp_message_chain = [] + query.adapter.on_monitoring_message_created = AsyncMock() + stage_specs = [ + ('PreProcessor', PreProcessor), + ('MessageProcessor', Processor), + ('ResponseWrapper', ResponseWrapper), + ('SendResponseBackStage', SendResponseBackStage), + ] + containers = [] + for name, cls in stage_specs: + instance = cls(pipeline_app) + await instance.initialize(query.pipeline_config) + containers.append(pipelinemgr.StageInstContainer(name, instance)) + entity = SimpleNamespace( + uuid=query.pipeline_uuid, + workspace_uuid=context.workspace_uuid, + name='Contract pipeline', + config=query.pipeline_config, + extensions_preferences={'enable_all_plugins': False, 'plugins': [{'author': 'test', 'name': 'observer'}]}, + ) + return pipelinemgr.RuntimePipeline(pipeline_app, entity, containers, context) + + return create + + +@pytest.fixture +def passthrough_plugin(pipeline_app): + """Return real SDK contexts, like a plugin which does not intercept a call.""" + from langbot_plugin.api.entities.context import EventContext + + def context_for(event, _bound_plugins): + return EventContext.from_event(event) + + pipeline_app.plugin_connector.emit_event.side_effect = context_for + return context_for + + +@pytest.mark.usefixtures('load_pipeline_modules') class TestStageChainIntegration: - """Tests for full stage chain (PreProcessor -> Processor -> SendResponseBackStage).""" + """Runtime dispatch must preserve input, output, plugin boundaries, and cleanup.""" @pytest.mark.asyncio - async def test_full_chain_text_message_flow(self, pipeline_app, fake_platform_adapter, set_fake_runner): - """ - Full chain: text message -> PreProcessor -> Processor -> SendResponseBackStage. - - Validates: - - PreProcessor sets up session, user_message - - Processor calls runner and populates resp_messages - - SendResponseBackStage calls adapter.reply_message - """ - from langbot.pkg.pipeline import entities - from langbot.pkg.pipeline.preproc import preproc - from langbot.pkg.pipeline.process import process - from langbot.pkg.pipeline.respback import respback - + async def test_full_chain_text_message_flow( + self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin + ): adapter, platform = fake_platform_adapter - - # Set fake runner - fake_runner = FakeRunner().returns('LANGBOT_FAKE_PONG') - set_fake_runner(fake_runner) - - # Create query - config = create_minimal_pipeline_config() - query = text_query('ping') + set_fake_runner(FakeRunner.returns('LANGBOT_FAKE_PONG')) + query = text_query('ping', pipeline_config=create_minimal_pipeline_config()) + query.pipeline_config['output']['misc']['quote-origin'] = True query.adapter = adapter - query.pipeline_config = config - query.resp_messages = [] - query.resp_message_chain = [] + runtime = await runtime_pipeline_factory(query) - # Mock plugin_connector for PreProcessor and Processor events - mock_event_ctx_preproc = Mock() - mock_event_ctx_preproc.event = Mock() - mock_event_ctx_preproc.event.default_prompt = [] - mock_event_ctx_preproc.event.prompt = [] + await runtime.run(query) - mock_event_ctx_processor = Mock() - mock_event_ctx_processor.is_prevented_default = Mock(return_value=False) - mock_event_ctx_processor.event = Mock() - mock_event_ctx_processor.event.user_message_alter = None - - pipeline_app.plugin_connector.emit_event = AsyncMock() - pipeline_app.plugin_connector.emit_event.side_effect = [ - mock_event_ctx_preproc, # PreProcessor PromptPreProcessing - mock_event_ctx_processor, # Processor NormalMessageReceived + assert pipeline_app.agent_run_orchestrator.observed_queries == [query] + assert str(query.user_message.get_content_platform_message_chain()) == 'ping' + assert [message.content for message in query.resp_messages] == ['LANGBOT_FAKE_PONG'] + assert [str(chain) for chain in query.resp_message_chain] == ['LANGBOT_FAKE_PONG'] + outbound = platform.get_outbound_messages() + assert len(outbound) == 1, pipeline_app.logger.error.call_args_list + assert outbound[0]['source'] is query.message_event + assert str(outbound[0]['message']) == 'LANGBOT_FAKE_PONG' + assert outbound[0]['quote_origin'] is True + assert platform.get_outbound_chunks() == [] + assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list + pipeline_app.logger.error.assert_not_called() + emitted = pipeline_app.plugin_connector.emit_event.await_args_list + assert [call.args[0].event_name for call in emitted] == [ + 'PersonMessageReceived', + 'PromptPreProcessing', + 'PersonNormalMessageReceived', + 'NormalMessageResponded', ] + assert [call.args[1] for call in emitted] == [['test/observer']] * 4 + pipeline_app.query_pool.remove_query.assert_awaited_once_with(query) - # Create stages - preproc_stage = preproc.PreProcessor(pipeline_app) - processor_stage = process.Processor(pipeline_app) - await processor_stage.initialize(config) - respback_stage = respback.SendResponseBackStage(pipeline_app) + @pytest.mark.asyncio + @pytest.mark.parametrize( + 'blocked_event', ['PersonMessageReceived', 'PersonNormalMessageReceived', 'NormalMessageResponded'] + ) + async def test_chain_stops_on_interrupt( + self, + pipeline_app, + fake_platform_adapter, + set_fake_runner, + runtime_pipeline_factory, + passthrough_plugin, + blocked_event, + ): + adapter, platform = fake_platform_adapter + set_fake_runner(FakeRunner.returns('MUST_NOT_BE_DELIVERED')) + query = text_query('hello', pipeline_config=create_minimal_pipeline_config()) + query.adapter = adapter + runtime = await runtime_pipeline_factory(query) - # Run PreProcessor - result1 = await preproc_stage.process(query, 'PreProcessor') - assert result1.result_type.value == entities.ResultType.CONTINUE.value - query = result1.new_query + def intercept(event, bound_plugins): + context = passthrough_plugin(event, bound_plugins) + if event.event_name == blocked_event: + context.prevent_default() + return context - # Run Processor - results = await collect_processor_results(processor_stage, query, 'MessageProcessor') - assert len(results) >= 1 + pipeline_app.plugin_connector.emit_event.side_effect = intercept + await runtime.run(query) - # Build resp_message_chain from resp_messages + event_names = [call.args[0].event_name for call in pipeline_app.plugin_connector.emit_event.await_args_list] + assert event_names[-1] == blocked_event + expected_runner_queries = [query] if blocked_event == 'NormalMessageResponded' else [] + assert pipeline_app.agent_run_orchestrator.observed_queries == expected_runner_queries + assert query.resp_message_chain == [] + assert platform.get_outbound_messages() == [] + assert platform.get_outbound_chunks() == [] + assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list + pipeline_app.logger.error.assert_not_called() + pipeline_app.query_pool.remove_query.assert_awaited_once_with(query) + + @pytest.mark.asyncio + async def test_wrapper_plugin_reply_reaches_adapter( + self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin + ): from tests.factories.message import text_chain - for resp_msg in query.resp_messages: - if resp_msg.content: - query.resp_message_chain.append(text_chain(resp_msg.content)) + adapter, platform = fake_platform_adapter + set_fake_runner(FakeRunner.returns('original runner answer')) + query = text_query('hello', pipeline_config=create_minimal_pipeline_config()) + query.adapter = adapter + runtime = await runtime_pipeline_factory(query) - # Run SendResponseBackStage - result3 = await respback_stage.process(query, 'SendResponseBackStage') - assert result3.result_type.value == entities.ResultType.CONTINUE.value + def alter_response(event, bound_plugins): + context = passthrough_plugin(event, bound_plugins) + if event.event_name == 'NormalMessageResponded': + context.event.reply_message_chain = text_chain('plugin replacement') + return context + + pipeline_app.plugin_connector.emit_event.side_effect = alter_response + await runtime.run(query) - # Verify adapter was called outbound = platform.get_outbound_messages() - assert len(outbound) >= 1 + assert len(outbound) == 1, pipeline_app.logger.error.call_args_list + assert str(outbound[0]['message']) == 'plugin replacement' + assert [str(chain) for chain in query.resp_message_chain] == ['plugin replacement'] + pipeline_app.query_pool.remove_query.assert_awaited_once_with(query) @pytest.mark.asyncio - async def test_chain_stops_on_interrupt(self, pipeline_app, fake_platform_adapter): - """ - Chain should stop when a stage returns INTERRUPT. - - PreProcessor returns CONTINUE, Processor returns INTERRUPT (prevent_default). - """ - from langbot.pkg.pipeline import entities - from langbot.pkg.pipeline.preproc import preproc - from langbot.pkg.pipeline.process import process - + async def test_runner_failure_delivers_hint_and_cleans_up( + self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin + ): adapter, platform = fake_platform_adapter - - # Create query - query = text_query('hello') + set_fake_runner(FakeRunner.raises(RuntimeError('provider unavailable'))) + query = text_query('hello', pipeline_config=create_minimal_pipeline_config()) query.adapter = adapter - query.pipeline_config = create_minimal_pipeline_config() + runtime = await runtime_pipeline_factory(query) - # Mock plugin_connector - PreProcessor continues, Processor interrupts - mock_event_ctx_preproc = Mock() - mock_event_ctx_preproc.event = Mock() - mock_event_ctx_preproc.event.default_prompt = [] - mock_event_ctx_preproc.event.prompt = [] + await runtime.run(query) - mock_event_ctx_processor = Mock() - mock_event_ctx_processor.is_prevented_default = Mock(return_value=True) - mock_event_ctx_processor.event = Mock() - mock_event_ctx_processor.event.reply_message_chain = None - - pipeline_app.plugin_connector.emit_event = AsyncMock() - pipeline_app.plugin_connector.emit_event.side_effect = [ - mock_event_ctx_preproc, # PreProcessor PromptPreProcessing - mock_event_ctx_processor, # Processor NormalMessageReceived + outbound = platform.get_outbound_messages() + assert len(outbound) == 1, pipeline_app.logger.error.call_args_list + assert str(outbound[0]['message']) == 'Request failed.' + assert query.variables['_monitoring_has_error'] is True + assert query.resp_messages == [] + assert query.resp_message_chain == [] + assert [call.args[0].event_name for call in pipeline_app.plugin_connector.emit_event.await_args_list] == [ + 'PersonMessageReceived', + 'PromptPreProcessing', + 'PersonNormalMessageReceived', ] + pipeline_app.query_pool.remove_query.assert_awaited_once_with(query) - # Create stages - preproc_stage = preproc.PreProcessor(pipeline_app) - processor_stage = process.Processor(pipeline_app) - await processor_stage.initialize(query.pipeline_config) + @pytest.mark.asyncio + async def test_streaming_chain_delivers_each_chunk_with_one_response_id( + self, pipeline_app, set_fake_runner, runtime_pipeline_factory, passthrough_plugin + ): + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk - # Run PreProcessor - result1 = await preproc_stage.process(query, 'PreProcessor') - assert result1.result_type.value == entities.ResultType.CONTINUE.value - query = result1.new_query + class StreamingRunner(FakeRunner): + async def run(self, query): + yield MessageChunk(role='assistant', content='first', all_content='first', is_final=False) + yield MessageChunk(role='assistant', content=' second', all_content='first second', is_final=True) - # Run Processor - should INTERRUPT - results = await collect_processor_results(processor_stage, query, 'MessageProcessor') + platform = FakePlatform(stream_output_supported=True) + adapter = mock_platform_adapter(platform) + adapter.create_message_card = AsyncMock(side_effect=platform.create_message_card) + set_fake_runner(StreamingRunner) + query = text_query('stream please', pipeline_config=create_minimal_pipeline_config()) + query.adapter = adapter + runtime = await runtime_pipeline_factory(query) - assert len(results) == 1 - assert results[0].result_type.value == entities.ResultType.INTERRUPT.value + await runtime.run(query) - # Chain stops here - no resp_messages - assert len(query.resp_messages) == 0 + chunks = platform.get_outbound_chunks() + assert [(str(chunk['message']), chunk['is_final']) for chunk in chunks] == [ + ('first', False), + ('first second', True), + ], pipeline_app.logger.error.call_args_list + assert all(chunk['source'] is query.message_event for chunk in chunks) + response_ids = [chunk['bot_message'].resp_message_id for chunk in chunks] + assert response_ids[0] and response_ids == [response_ids[0]] * 2 + adapter.create_message_card.assert_awaited_once_with(response_ids[0], query.message_event) + assert [str(chain) for chain in query.resp_message_chain] == ['first second'] + assert platform.get_outbound_messages() == [] + assert pipeline_app.agent_run_orchestrator.observed_queries == [query] + assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list + pipeline_app.logger.error.assert_not_called() + pipeline_app.query_pool.remove_query.assert_awaited_once_with(query) diff --git a/tests/unit_tests/api/service/test_bot_service.py b/tests/unit_tests/api/service/test_bot_service.py index da2bedff6..96c6d24f0 100644 --- a/tests/unit_tests/api/service/test_bot_service.py +++ b/tests/unit_tests/api/service/test_bot_service.py @@ -867,7 +867,8 @@ class TestBotServiceSendMessage: with pytest.raises(Exception, match='Invalid message_chain format'): await service.send_message(WORKSPACE_UUID, 'bot-uuid', 'group', '123', {'invalid': 'format'}) - async def test_send_message_valid_call(self): + @pytest.mark.parametrize(('target_type', 'target_id'), [('group', 123), ('person', 'user-456')]) + async def test_send_message_valid_call(self, target_type, target_id): """Sends message through adapter when all valid.""" # Setup ap = SimpleNamespace() @@ -881,14 +882,18 @@ class TestBotServiceSendMessage: service = BotService(ap) service.get_bot = AsyncMock(return_value={'uuid': 'bot-uuid'}) - # Execute with valid message chain format - message_chain_data = {'messages': [{'type': 'text', 'data': {'text': 'Hello'}}]} + from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain - # Patch the import location - the module imports inside the function - with patch('langbot_plugin.api.entities.builtin.platform.message.MessageChain') as MockMessageChain: - mock_chain = Mock() - MockMessageChain.model_validate = Mock(return_value=mock_chain) - await service.send_message(WORKSPACE_UUID, 'bot-uuid', 'group', '123', message_chain_data) + expected_chain = MessageChain([Plain(text='Hello')]) + message_chain_data = expected_chain.model_dump() + await service.send_message(WORKSPACE_UUID, 'bot-uuid', target_type, target_id, message_chain_data) - # Verify adapter.send_message was called - runtime_bot.adapter.send_message.assert_called_once_with('group', '123', mock_chain) + service.get_bot.assert_awaited_once_with(WORKSPACE_UUID, 'bot-uuid', include_secret=False) + ap.platform_mgr.get_bot_by_uuid.assert_awaited_once_with(WORKSPACE_UUID, 'bot-uuid') + runtime_bot.adapter.send_message.assert_awaited_once() + sent_target_type, sent_target_id, sent_chain = runtime_bot.adapter.send_message.await_args.args + assert sent_target_type == target_type + assert sent_target_id == str(target_id) + assert isinstance(sent_chain, MessageChain) + assert sent_chain.model_dump() == expected_chain.model_dump() + assert str(sent_chain) == 'Hello' diff --git a/tests/unit_tests/api/service/test_user_service.py b/tests/unit_tests/api/service/test_user_service.py index add2ea7b3..493c033a2 100644 --- a/tests/unit_tests/api/service/test_user_service.py +++ b/tests/unit_tests/api/service/test_user_service.py @@ -16,6 +16,10 @@ from __future__ import annotations import pytest import jwt import datetime + +import sqlalchemy +from argon2 import PasswordHasher +from argon2.exceptions import VerifyMismatchError from unittest.mock import AsyncMock, Mock from types import SimpleNamespace @@ -510,10 +514,17 @@ class TestUserServiceResetPassword: service = UserService(ap) # Execute - await service.reset_password('test@example.com', 'new_password') + await service.reset_password(' Test@Example.COM ', 'new_password') - # Verify - execute_async was called with update - ap.persistence_mgr.execute_async.assert_called_once() + ap.persistence_mgr.execute_async.assert_awaited_once() + statement = ap.persistence_mgr.execute_async.await_args.args[0] + assert isinstance(statement, sqlalchemy.sql.dml.Update) + assert statement.table.name == User.__tablename__ + params = statement.compile().params + assert params['normalized_email_1'] == 'test@example.com' + assert PasswordHasher().verify(params['password'], 'new_password') + with pytest.raises(VerifyMismatchError): + PasswordHasher().verify(params['password'], 'old-password') class TestUserServiceChangePassword: diff --git a/tests/unit_tests/api/service/test_webhook_service.py b/tests/unit_tests/api/service/test_webhook_service.py index 64f58de05..fcf44c094 100644 --- a/tests/unit_tests/api/service/test_webhook_service.py +++ b/tests/unit_tests/api/service/test_webhook_service.py @@ -377,104 +377,59 @@ class TestWebhookServiceGetWebhook: class TestWebhookServiceUpdateWebhook: """Tests for update_webhook method.""" - async def test_update_webhook_name_only(self): - """Updates only the name field.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result()) + @pytest.mark.parametrize( + 'changes', + [ + {'name': 'Updated Name'}, + {'url': 'https://updated.example.com'}, + {'description': 'Updated description'}, + {'enabled': False}, + { + 'name': 'All Updated', + 'url': 'https://all.updated.example.com', + 'description': 'All updated description', + 'enabled': False, + }, + ], + ids=['name-only', 'url-only', 'description-only', 'enabled-only', 'all-fields'], + ) + async def test_update_webhook_persists_requested_fields(self, tenant_webhook_service, changes): + """Persist requested fields without changing omitted fields or another Workspace.""" + original = { + 'name': 'Original Name', + 'url': 'https://original.example.com', + 'description': 'Original description', + 'enabled': True, + } + created = await tenant_webhook_service.create_webhook(ISOLATION_WORKSPACE_A, **original) + other = await tenant_webhook_service.create_webhook(ISOLATION_WORKSPACE_B, **original) - service = WebhookService(ap) + assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_A, created['id'], **changes) is True - # Execute - await service.update_webhook(WORKSPACE_UUID, 1, name='Updated Name') + updated = await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True) + assert {field: updated[field] for field in original} == original | changes + assert updated['id'] == created['id'] + assert updated['workspace_uuid'] == ISOLATION_WORKSPACE_A + assert ( + await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_B, other['id'], include_secret=True) == other + ) - # Verify - ap.persistence_mgr.execute_async.assert_called_once() - - async def test_update_webhook_url_only(self): - """Updates only the url field.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result()) - - service = WebhookService(ap) - - # Execute - await service.update_webhook(WORKSPACE_UUID, 1, url='http://updated.example.com') - - # Verify - ap.persistence_mgr.execute_async.assert_called_once() - - async def test_update_webhook_description_only(self): - """Updates only the description field.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result()) - - service = WebhookService(ap) - - # Execute - await service.update_webhook(WORKSPACE_UUID, 1, description='Updated description') - - # Verify - ap.persistence_mgr.execute_async.assert_called_once() - - async def test_update_webhook_enabled_only(self): - """Updates only the enabled field.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result()) - - service = WebhookService(ap) - - # Execute - await service.update_webhook(WORKSPACE_UUID, 1, enabled=False) - - # Verify - ap.persistence_mgr.execute_async.assert_called_once() - - async def test_update_webhook_all_fields(self): - """Updates all fields at once.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result()) - - service = WebhookService(ap) - - # Execute - await service.update_webhook( - WORKSPACE_UUID, - 1, - name='All Updated', - url='http://all.updated.com', - description='All updated description', + async def test_update_webhook_no_fields(self, tenant_webhook_service): + """An empty update reports scoped existence and leaves persisted values unchanged.""" + created = await tenant_webhook_service.create_webhook( + ISOLATION_WORKSPACE_A, + 'Unchanged', + 'https://unchanged.example.com', + description='Keep this description', enabled=False, ) - # Verify - ap.persistence_mgr.execute_async.assert_called_once() - - async def test_update_webhook_no_fields(self): - """Does nothing when no fields provided.""" - # Setup - ap = SimpleNamespace() - ap.persistence_mgr = SimpleNamespace() - existing = _create_mock_webhook(webhook_id=1) - ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_mock_result(first_item=existing)) - ap.persistence_mgr.serialize_model = Mock(return_value={'id': 1}) - - service = WebhookService(ap) - - # Execute - no update parameters - await service.update_webhook(WORKSPACE_UUID, 1) - - # No write is issued; one scoped existence lookup is performed. - ap.persistence_mgr.execute_async.assert_called_once() + assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_A, created['id']) is True + assert ( + await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True) + == created + ) + assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_B, created['id']) is False class TestWebhookServiceDeleteWebhook: @@ -695,6 +650,10 @@ async def test_update_and_delete_are_scoped(tenant_webhook_service): name='new', enabled=False, ) + updated = await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True) + assert updated['name'] == 'new' + assert updated['enabled'] is False + assert updated['url'] == 'https://a.invalid/old' assert await tenant_webhook_service.get_enabled_webhooks(ISOLATION_WORKSPACE_A) == [] assert await tenant_webhook_service.delete_webhook(ISOLATION_WORKSPACE_A, created['id']) assert await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id']) is None diff --git a/tests/unit_tests/pipeline/test_chat_handler_logging.py b/tests/unit_tests/pipeline/test_chat_handler_logging.py index 6ae85558f..83769bfae 100644 --- a/tests/unit_tests/pipeline/test_chat_handler_logging.py +++ b/tests/unit_tests/pipeline/test_chat_handler_logging.py @@ -2,14 +2,11 @@ from __future__ import annotations from unittest.mock import Mock -import pytest import langbot_plugin.api.entities.builtin.provider.message as provider_message -# TODO: unskip once the handler ↔ app circular import is resolved -pytest.skip( - 'circular import in handler ↔ app; will be unblocked once resolved', - allow_module_level=True, -) +# Initialize the application import graph before importing a leaf handler. This +# matches normal startup and avoids handler -> app -> handler partial imports. +from langbot.pkg.core import app # noqa: F401 from langbot.pkg.pipeline.process.handler import MessageHandler # noqa: E402