diff --git a/tests/e2e/test_startup.py b/tests/e2e/test_startup.py index 955e33af6..939effc6b 100644 --- a/tests/e2e/test_startup.py +++ b/tests/e2e/test_startup.py @@ -16,6 +16,75 @@ import pytest pytestmark = pytest.mark.e2e +@pytest.fixture(scope='session') +def e2e_tmpdir(tmp_path_factory): + # Initial registration is global; keep this module independent of other E2E accounts. + return tmp_path_factory.mktemp('langbot_startup') + + +@pytest.fixture(scope='session') +def e2e_port(): + import socket + + with socket.socket() as listener: + listener.bind(('127.0.0.1', 0)) + return listener.getsockname()[1] + + +@pytest.fixture(scope='session') +def langbot_process(e2e_tmpdir, e2e_port): + from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories + from tests.e2e.utils.process_manager import LangBotProcess, find_project_root + + create_minimal_config(e2e_tmpdir, port=e2e_port) + create_test_directories(e2e_tmpdir) + process = LangBotProcess( + project_root=find_project_root(), + work_dir=e2e_tmpdir, + port=e2e_port, + timeout=180, + collect_coverage=True, + ) + try: + assert process.start(), process.get_logs() + yield process + finally: + process.stop() + coverage_file = process.get_coverage_file() + if coverage_file is not None and coverage_file.exists(): + import shutil + + shutil.copy(coverage_file, find_project_root() / '.coverage.e2e.startup') + + +@pytest.fixture(scope='session') +def e2e_db_path(e2e_tmpdir): + return e2e_tmpdir / 'data/langbot.db' + + +@pytest.fixture +def startup_credentials(e2e_client): + credentials = {'user': 'startup-e2e@example.com', 'password': 'startup-local-test-password'} + state = e2e_client.get('/api/v1/user/init') + assert state.status_code == 200 + if not state.json()['data']['initialized']: + response = e2e_client.post('/api/v1/user/init', json=credentials) + assert response.status_code == 200, response.text + assert response.json()['code'] == 0 + return credentials + + +@pytest.fixture +def startup_headers(e2e_client, startup_credentials): + response = e2e_client.post('/api/v1/user/auth', json=startup_credentials) + assert response.status_code == 200, response.text + headers = {'Authorization': f'Bearer {response.json()["data"]["token"]}'} + bootstrap = e2e_client.get('/api/v1/workspaces/bootstrap', headers=headers) + assert bootstrap.status_code == 200, bootstrap.text + headers['X-Workspace-Id'] = bootstrap.json()['data']['workspaces'][0]['workspace']['uuid'] + return headers + + class TestStartupFlow: """Tests for LangBot startup process.""" @@ -87,22 +156,21 @@ class TestStartupFlow: response = e2e_client.get('/api/v1/pipelines') assert response.status_code == 401 - def test_auth_endpoint(self, e2e_client, e2e_tmpdir): - """Test auth endpoint.""" - # First startup may allow initial setup - response = e2e_client.post( - '/api/v1/user/auth', - json={ - 'user': 'admin', - 'password': 'admin', - }, - ) + def test_auth_endpoint(self, e2e_client, startup_credentials): + """A controlled local account can sign in and use its token.""" + response = e2e_client.post('/api/v1/user/auth', json=startup_credentials) + assert response.status_code == 200, response.text + assert response.json()['code'] == 0 + token = response.json()['data']['token'] + assert isinstance(token, str) and token + bootstrap = e2e_client.get('/api/v1/workspaces/bootstrap', headers={'Authorization': f'Bearer {token}'}) + assert bootstrap.status_code == 200, bootstrap.text + assert bootstrap.json()['data']['workspaces'][0]['membership']['role'] == 'owner' - # Response could be: - # - 200 if auth succeeds - # - 400 if credentials wrong - # - 401 if user not initialized - assert response.status_code in [200, 400, 401] + rejected = e2e_client.post('/api/v1/user/auth', json={**startup_credentials, 'password': 'wrong-password'}) + assert rejected.status_code == 200, rejected.text + assert rejected.json()['code'] == 1 + assert rejected.json().get('data', {}).get('token') is None class TestStartupStages: @@ -127,24 +195,37 @@ class TestStartupStages: cursor.execute('SELECT version_num FROM alembic_version;') version = cursor.fetchone() - assert version is not None, 'Migration version should be set' + from pathlib import Path + + from alembic.script import ScriptDirectory + + migrations = Path(__file__).parents[2] / 'src/langbot/pkg/persistence/alembic' + assert version == (ScriptDirectory(str(migrations)).get_current_head(),) conn.close() - def test_http_controller_initialized(self, e2e_client): - """Verify HTTP controller was initialized.""" - # Multiple endpoints should be available - endpoints = [ - '/api/v1/system/info', - '/api/v1/pipelines', - '/api/v1/provider/providers', - '/api/v1/platform/bots', - ] + def test_http_controller_initialized(self, e2e_client, startup_headers): + """Authenticated business endpoints return their documented collections.""" + info = e2e_client.get('/api/v1/system/info', headers=startup_headers) + assert info.status_code == 200 + assert isinstance(info.json()['data']['version'], str) - for endpoint in endpoints: - response = e2e_client.get(endpoint) - # Should get a real route response, even if auth is required. - assert response.status_code in [200, 401, 403], f'{endpoint} should be registered' + for endpoint, key in [ + ('/api/v1/pipelines', 'pipelines'), + ('/api/v1/provider/providers', 'providers'), + ('/api/v1/platform/bots', 'bots'), + ]: + response = e2e_client.get(endpoint, headers=startup_headers) + assert response.status_code == 200, (endpoint, response.text) + payload = response.json() + assert payload['code'] == 0 + assert isinstance(payload['data'][key], list) + if key == 'providers': + assert [(item['name'], item['requester'], item['api_keys']) for item in payload['data'][key]] == [ + ('LangBot Models', 'space-chat-completions', []) + ] + else: + assert payload['data'][key] == [] class TestMinimalStartupNoLLM: @@ -155,8 +236,14 @@ class TestMinimalStartupNoLLM: response = e2e_client.get('/api/v1/system/info') assert response.status_code == 200 - def test_pipeline_metadata_available(self, e2e_client): - """Pipeline metadata endpoint should work without LLM.""" - # Requires auth, but endpoint should exist - response = e2e_client.get('/api/v1/pipelines/_/metadata') - assert response.status_code in [200, 401] # Not 404 or 500 + def test_pipeline_metadata_available(self, e2e_client, startup_headers): + """All pipeline configuration groups remain readable without an LLM.""" + response = e2e_client.get('/api/v1/pipelines/_/metadata', headers=startup_headers) + assert response.status_code == 200, response.text + payload = response.json() + assert payload['code'] == 0 + configs = payload['data']['configs'] + assert [config['name'] for config in configs] == ['trigger', 'safety', 'ai', 'output'] + ai = configs[2] + runner = next(stage for stage in ai['stages'] if stage['name'] == 'runner') + assert any(item['name'] == 'id' for item in runner['config']) diff --git a/tests/integration/pipeline/test_full_flow.py b/tests/integration/pipeline/test_full_flow.py index 6b2a0cf72..8ce63c4e3 100644 --- a/tests/integration/pipeline/test_full_flow.py +++ b/tests/integration/pipeline/test_full_flow.py @@ -353,6 +353,9 @@ class TestPreProcessorStage: # Check user_message content assert result.new_query.user_message is not None assert result.new_query.user_message.role == 'user' + assert [(part.type, part.text) for part in result.new_query.user_message.content] == [ + ('text', 'test message content') + ] @pytest.mark.usefixtures('load_pipeline_modules') @@ -370,6 +373,14 @@ class TestProcessorStage: # Create query query = text_query('hello') + from langbot_plugin.api.entities.builtin.provider.session import Session + + query.session = Session( + launcher_type=query.launcher_type, + launcher_id=query.launcher_id, + sender_id=query.sender_id, + using_conversation=await pipeline_app.sess_mgr.get_conversation(), + ) query.adapter = adapter query.pipeline_config = create_minimal_pipeline_config() query.resp_messages = [] @@ -390,9 +401,13 @@ class TestProcessorStage: # Collect results using helper results = await collect_processor_results(processor_stage, query, 'MessageProcessor') - assert len(results) >= 1 - # Check that resp_messages was populated - assert len(query.resp_messages) >= 1 + from langbot.pkg.pipeline import entities + + assert pipeline_app.agent_run_orchestrator.observed_queries == [query] + assert [result.result_type for result in results] == [entities.ResultType.CONTINUE] + assert [(message.role, message.content) for message in query.resp_messages] == [ + ('assistant', 'LANGBOT_FAKE_PONG') + ] @pytest.mark.asyncio async def test_processor_prevent_default_without_reply_interrupts(self, pipeline_app, fake_platform_adapter): diff --git a/tests/integration_tests/box/test_box_integration.py b/tests/integration_tests/box/test_box_integration.py index 9358515cd..02f5d4dd3 100644 --- a/tests/integration_tests/box/test_box_integration.py +++ b/tests/integration_tests/box/test_box_integration.py @@ -249,17 +249,68 @@ async def test_timeout_kills_command(box_client: ActionRPCBoxClient): @requires_socket @pytest.mark.asyncio async def test_offline_cannot_reach_network(box_client: ActionRPCBoxClient): - """With network=OFF the sandbox cannot reach the internet.""" - spec = BoxSpec( - cmd='wget -q -O /dev/null --timeout=3 http://1.1.1.1 2>&1; exit $?', - session_id='int-offline', - workdir='/tmp', - network=BoxNetworkMode.OFF, - image=_TEST_IMAGE, + """Only OFF blocks the same reachable private peer and installed HTTP client.""" + server = await box_client.execute( + BoxSpec( + cmd=r"""cat > /tmp/response.sh <<'HTTP' +#!/bin/sh +while IFS= read -r line; do + [ -z "$(printf '%s' "$line" | busybox tr -d '\r')" ] && break +done +printf 'HTTP/1.0 200 OK\r\nContent-Length: 19\r\n\r\nbox-network-control' +HTTP +chmod +x /tmp/response.sh +(while true; do busybox nc -l -p 8080 -e /bin/sh /tmp/response.sh; done >/tmp/http.log 2>&1 < /dev/null &) +sleep 1 && hostname -i""", + session_id='int-network-peer', + workdir='/tmp', + network=BoxNetworkMode.ON, + image=_TEST_IMAGE, + ) ) - result = await box_client.execute(spec) + assert server.status == BoxExecutionStatus.COMPLETED + assert server.exit_code == 0, server.stderr + import ipaddress - assert result.exit_code != 0 + peer = str(ipaddress.IPv4Address(server.stdout.strip().split()[0])) + assert not ipaddress.ip_address(peer).is_loopback + command = f'wget -q -O - --timeout=3 http://{peer}:8080/' + online = await box_client.execute( + BoxSpec( + cmd=command, + session_id='int-online', + workdir='/tmp', + network=BoxNetworkMode.ON, + image=_TEST_IMAGE, + ) + ) + assert online.status == BoxExecutionStatus.COMPLETED + assert online.exit_code == 0, online.stderr + assert online.stdout == 'box-network-control' + + offline = await box_client.execute( + BoxSpec( + cmd=command, + session_id='int-offline', + workdir='/tmp', + network=BoxNetworkMode.OFF, + image=_TEST_IMAGE, + ) + ) + assert offline.status == BoxExecutionStatus.COMPLETED + assert offline.exit_code != 0 + assert offline.stdout == '' + still_online = await box_client.execute( + BoxSpec( + cmd=command, + session_id='int-online', + workdir='/tmp', + network=BoxNetworkMode.ON, + image=_TEST_IMAGE, + ) + ) + assert still_online.exit_code == 0 + assert still_online.stdout == 'box-network-control' # ── 5. Backend unavailable ─────────────────────────────────────────── diff --git a/tests/unit_tests/box/test_box_service.py b/tests/unit_tests/box/test_box_service.py index 1d25299de..301fbd2d5 100644 --- a/tests/unit_tests/box/test_box_service.py +++ b/tests/unit_tests/box/test_box_service.py @@ -434,7 +434,7 @@ async def test_box_service_get_sessions_delegates_to_client(): sessions = await service.get_sessions(_CONTEXT) assert sessions == [{'session_id': 'test-session'}] - client.get_sessions.assert_awaited_once() + client.get_sessions.assert_awaited_once_with(action_context=_ACTION_CONTEXT) @pytest.mark.asyncio @@ -599,12 +599,14 @@ async def test_cloud_box_service_reconnect_restores_runtime_only( connector.reconnect = AsyncMock() service._ensure_default_workspace = Mock() service._verify_cloud_runtime = AsyncMock() + service._purge_attachment_dirs = AsyncMock() monkeypatch.setattr('langbot.pkg.box.service.asyncio.sleep', AsyncMock()) await service._reconnect_loop(connector) connector.reconnect.assert_awaited_once() service._verify_cloud_runtime.assert_awaited_once() + service._purge_attachment_dirs.assert_not_awaited() assert service.available is True @@ -674,6 +676,8 @@ async def test_box_service_allows_host_mount_under_configured_root(tmp_path): assert result['ok'] is True assert backend.start_calls == ['person_test_user'] + assert backend.start_specs[0].host_path == os.path.realpath(host_dir) + assert backend.start_specs[0].host_path_mode == BoxHostMountMode.READ_WRITE @pytest.mark.asyncio @@ -744,16 +748,20 @@ async def test_box_service_rejects_host_mount_outside_allowed_roots(tmp_path): service = BoxService(make_app(logger, [str(allowed_root)]), client=_InProcessBoxRuntimeClient(logger, runtime)) await service.initialize() - with pytest.raises(BoxValidationError): + with pytest.raises(BoxValidationError, match='outside allowed_mount_roots'): await service.execute_spec_payload( { 'cmd': 'pwd', 'host_path': str(disallowed_root), - 'session_id': '12', + 'session_id': 'person_test_user', }, make_query(12), ) + assert backend.start_calls == [] + assert backend.exec_calls == [] + assert list(disallowed_root.iterdir()) == [] + @pytest.mark.asyncio async def test_box_runtime_rejects_host_mount_conflict_in_same_session(tmp_path): @@ -1064,8 +1072,18 @@ def test_builtin_profiles_are_consistent(): @pytest.mark.asyncio -async def test_profile_default_applies_resource_limits(): +async def test_profile_default_applies_resource_limits(monkeypatch): """Default profile resource limits are applied to BoxSpec.""" + profile = BUILTIN_PROFILES['default'].model_copy( + update={ + 'cpus': 0.75, + 'memory_mb': 320, + 'pids_limit': 37, + 'read_only_rootfs': False, + 'workspace_quota_mb': 19, + } + ) + monkeypatch.setitem(BUILTIN_PROFILES, 'default', profile) logger = Mock() backend = FakeBackend(logger) runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300) @@ -1473,8 +1491,14 @@ async def test_rpc_client_get_status(): try: status = await client.get_status(action_context=_ACTION_CONTEXT) - assert 'backend' in status - assert 'active_sessions' in status + assert status['backend']['name'] == 'fake' + assert status['backend']['available'] is True + assert status['active_sessions'] == 0 + await client.execute(BoxSpec(cmd='true', session_id='status-session'), action_context=_ACTION_CONTEXT) + status = await client.get_status(action_context=_ACTION_CONTEXT) + assert status['active_sessions'] == 1 + assert status['managed_processes'] == 0 + assert status['session_ttl_sec'] == 300 finally: server_task.cancel() client_task.cancel() @@ -1515,8 +1539,10 @@ async def test_rpc_client_delete_session(): spec = BoxSpec.model_validate({'cmd': 'echo hi', 'session_id': 'r-del-1'}) await client.execute(spec, action_context=_ACTION_CONTEXT) + physical_session_id = backend.start_calls[0] await client.delete_session('r-del-1', action_context=_ACTION_CONTEXT) + assert backend.stop_calls == [physical_session_id] sessions = await client.get_sessions(action_context=_ACTION_CONTEXT) assert len(sessions) == 0 finally: @@ -1958,13 +1984,26 @@ class TestAttachmentHelpers: class TestInboundOutboundRoundTrip: + @staticmethod + async def _execute_command(command, workspace): + process = await asyncio.create_subprocess_exec( + '/bin/sh', + '-c', + command.replace('/workspace', str(workspace)), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + assert process.returncode == 0, stderr.decode() + return stdout.decode().replace(str(workspace), '/workspace') + def _service(self) -> BoxService: service = BoxService(make_app(Mock()), client=Mock(spec=BoxRuntimeClient)) service._available = True return service @pytest.mark.asyncio - async def test_materialize_inbound_writes_and_describes(self): + async def test_materialize_inbound_writes_and_describes(self, tmp_path): import base64 import langbot_plugin.api.entities.builtin.platform.message as platform_message @@ -1982,14 +2021,9 @@ class TestInboundOutboundRoundTrip: ] ) - # Mock the sandbox write path: echo back the written paths. async def fake_execute_tool(parameters, q): - assert '/workspace/inbox/' in parameters['command'] - return { - 'ok': True, - 'stdout': '["/workspace/inbox/query-42/image_1.png"]', - 'stderr': '', - } + assert q is query + return {'ok': True, 'stdout': await self._execute_command(parameters['command'], tmp_path), 'stderr': ''} service.execute_tool = AsyncMock(side_effect=fake_execute_tool) @@ -1999,6 +2033,7 @@ class TestInboundOutboundRoundTrip: assert d['type'] == 'Image' assert d['path'] == '/workspace/inbox/query-42/image_1.png' assert d['size'] == len(img_bytes) + assert (tmp_path / 'inbox/query-42/image_1.png').read_bytes() == img_bytes @pytest.mark.asyncio async def test_materialize_inbound_noop_without_attachments(self): @@ -2012,45 +2047,40 @@ class TestInboundOutboundRoundTrip: service.execute_tool.assert_not_called() @pytest.mark.asyncio - async def test_collect_outbound_reads_and_clears(self): + async def test_collect_outbound_reads_and_clears(self, tmp_path): service = self._service() query = make_query() - - calls = [] + outbox = tmp_path / 'outbox/query-42' + outbox.mkdir(parents=True) + (outbox / 'out.png').write_bytes(b'ABC') + guard = tmp_path / 'outbox/query-other/protected.txt' + guard.parent.mkdir() + guard.write_bytes(b'other query') async def fake_client_execute(spec, *, action_context=None): - assert action_context.workspace_uuid == query.workspace_uuid - cmd = spec.cmd - calls.append(cmd) - if 'os.scandir' in cmd: - return BoxExecutionResult( - session_id='s', - backend_name='test', - status=BoxExecutionStatus.COMPLETED, - exit_code=0, - stdout='[{"name": "out.png", "b64": "QUJD"}]', - duration_ms=10, - ) - # the rm -rf cleanup call + assert action_context == _ACTION_CONTEXT return BoxExecutionResult( session_id='s', backend_name='test', status=BoxExecutionStatus.COMPLETED, exit_code=0, - stdout='', + stdout=await self._execute_command(spec.cmd, tmp_path), duration_ms=10, ) + async def fake_execute_tool(parameters, q): + assert q is query + return {'ok': True, 'stdout': await self._execute_command(parameters['command'], tmp_path), 'stderr': ''} + service.client.execute = AsyncMock(side_effect=fake_client_execute) - service.execute_tool = AsyncMock(return_value={'ok': True, 'stdout': '', 'stderr': ''}) + service.execute_tool = AsyncMock(side_effect=fake_execute_tool) attachments = await service.collect_outbound_attachments(query) - assert len(attachments) == 1 - assert attachments[0]['type'] == 'Image' - assert attachments[0]['name'] == 'out.png' - # cleanup (rm -rf) must have been issued after a successful collection - service.execute_tool.assert_awaited_once() - assert 'rm -rf' in service.execute_tool.await_args.args[0]['command'] + + assert attachments == [{'type': 'Image', 'name': 'out.png', 'base64': 'data:image/png;base64,QUJD'}] + assert not (outbox / 'out.png').exists() + assert guard.read_bytes() == b'other query' + assert await service.collect_outbound_attachments(query) == [] @pytest.mark.asyncio async def test_collect_outbound_empty_still_clears(self): @@ -2157,6 +2187,7 @@ class TestAttachmentHostPath: stale_dir = os.path.join(ws, 'inbox', 'query-42') os.makedirs(stale_dir, exist_ok=True) pathlib.Path(stale_dir, 'image_1.png').write_bytes(b'STALE-OLD-IMAGE') + pathlib.Path(stale_dir, 'leftover-secret.txt').write_bytes(b'STALE-OTHER-FILE') new = b'\x89PNG\r\n\x1a\n NEW' b64 = 'data:image/png;base64,' + base64.b64encode(new).decode() @@ -2170,6 +2201,7 @@ class TestAttachmentHostPath: assert host_bytes == new # No leftover content from the stale image. assert b'STALE-OLD-IMAGE' not in host_bytes + assert sorted(os.listdir(stale_dir)) == ['image_1.png'] @pytest.mark.asyncio async def test_inbound_host_replaces_query_symlink_without_touching_other_workspace(self, tmp_path): diff --git a/tests/unit_tests/platform/test_officialaccount_eba_adapter.py b/tests/unit_tests/platform/test_officialaccount_eba_adapter.py index f6d55aa4a..65c812139 100644 --- a/tests/unit_tests/platform/test_officialaccount_eba_adapter.py +++ b/tests/unit_tests/platform/test_officialaccount_eba_adapter.py @@ -137,10 +137,7 @@ async def test_officialaccount_message_converter_maps_components_to_passive_text ) ) - assert 'hi' in content - assert '[Image]' in content - assert '[File: a.txt]' in content - assert 'quoted' in content + assert content == 'hi\n[Image]\n[File: a.txt]\nquoted' @pytest.mark.asyncio diff --git a/tests/unit_tests/platform/test_openclaw_weixin_client.py b/tests/unit_tests/platform/test_openclaw_weixin_client.py index 6c8dcadc5..11f0779ab 100644 --- a/tests/unit_tests/platform/test_openclaw_weixin_client.py +++ b/tests/unit_tests/platform/test_openclaw_weixin_client.py @@ -11,12 +11,20 @@ from langbot.libs.openclaw_weixin_api.client import ( from langbot.libs.openclaw_weixin_api.types import ApiError -def test_cdn_crypto_helpers_round_trip(): - original = b'tenant-media' * 128 +def test_cdn_crypto_helpers_round_trip(monkeypatch): + # AES-128-ECB with PKCS7 padding, independently generated using OpenSSL. + key = bytes.fromhex('2b7e151628aed2a6abf7158809cf4f3c') + original = bytes.fromhex('6bc1bee22e409f96e93d7e117393172a') + expected_ciphertext = bytes.fromhex('3ad77bb40d7a3660a89ecaf32466ef97a254be88e037ddd9d79fb6411c3f9df8') + monkeypatch.setattr('langbot.libs.openclaw_weixin_api.client.os.urandom', lambda size: key) - aes_key_hex, _encoded_key, encrypted, _raw_md5 = _encrypt_cdn_payload(original) + aes_key_hex, encoded_key, encrypted, raw_md5 = _encrypt_cdn_payload(original) - assert _decrypt_cdn_payload(encrypted, bytes.fromhex(aes_key_hex)) == original + assert aes_key_hex == '2b7e151628aed2a6abf7158809cf4f3c' + assert encoded_key == 'MmI3ZTE1MTYyOGFlZDJhNmFiZjcxNTg4MDljZjRmM2M=' + assert encrypted == expected_ciphertext + assert raw_md5 == 'f2ecfb66e72f88f217e659ac75b1075f' + assert _decrypt_cdn_payload(expected_ciphertext, key) == original @pytest.mark.asyncio diff --git a/tests/unit_tests/platform/test_slack_eba_adapter.py b/tests/unit_tests/platform/test_slack_eba_adapter.py index 8c6e7decd..faa1816b2 100644 --- a/tests/unit_tests/platform/test_slack_eba_adapter.py +++ b/tests/unit_tests/platform/test_slack_eba_adapter.py @@ -164,12 +164,7 @@ async def test_slack_message_converter_maps_common_components_to_text(): ) ) - assert 'hi' in text - assert '<@U-2>' in text - assert '' in text - assert 'https://example.test/a.png' in text - assert 'a.txt' in text - assert 'quoted' in text + assert text == 'hi\n<@U-2>\n\nhttps://example.test/a.png\na.txt\nquoted' @pytest.mark.asyncio diff --git a/tests/unit_tests/platform/test_telegram_eba_adapter.py b/tests/unit_tests/platform/test_telegram_eba_adapter.py index eca17ca4a..24761fa8e 100644 --- a/tests/unit_tests/platform/test_telegram_eba_adapter.py +++ b/tests/unit_tests/platform/test_telegram_eba_adapter.py @@ -604,6 +604,17 @@ async def test_telegram_platform_apis_call_underlying_bot_methods(): assert invite['invite_link'] == 'https://t.me/+abc' assert await PLATFORM_API_MAP['answer_callback_query'](bot, {'callback_query_id': 'cb'}) == {'ok': True} + bot.pin_chat_message.assert_awaited_once_with(chat_id=1, message_id=2, disable_notification=False) + bot.unpin_chat_message.assert_awaited_once_with(chat_id=1, message_id=2) + bot.unpin_all_chat_messages.assert_awaited_once_with(chat_id=1) + bot.get_chat_administrators.assert_awaited_once_with(chat_id=1) + bot.set_chat_title.assert_awaited_once_with(chat_id=1, title='New') + bot.set_chat_description.assert_awaited_once_with(chat_id=1, description='Desc') + bot.get_chat_member_count.assert_awaited_once_with(chat_id=1) + bot.send_chat_action.assert_awaited_once_with(chat_id=1, action='typing') + bot.create_chat_invite_link.assert_awaited_once_with(chat_id=1, name='invite', expire_date=None, member_limit=None) + bot.answer_callback_query.assert_awaited_once_with(callback_query_id='cb', text=None, show_alert=False, url=None) + @pytest.mark.asyncio async def test_telegram_unmute_member_uses_current_chat_permissions_fields(): diff --git a/tests/unit_tests/platform/test_websocket_adapter_attachments.py b/tests/unit_tests/platform/test_websocket_adapter_attachments.py index 3274626bf..9504ca251 100644 --- a/tests/unit_tests/platform/test_websocket_adapter_attachments.py +++ b/tests/unit_tests/platform/test_websocket_adapter_attachments.py @@ -90,17 +90,23 @@ async def test_image_defaults_to_png(): adapter, _, _ = _make_adapter() chain = [{'type': 'Image', 'path': f'{_UPLOAD_PREFIX}blob'}] await adapter._process_image_components(_make_connection(), chain) - assert chain[0]['base64'].startswith('data:image/png;base64,') + assert chain[0]['base64'] == 'data:image/png;base64,aGVsbG8=' @pytest.mark.asyncio -async def test_voice_uses_guessed_or_wav_mimetype(): +async def test_voice_uses_guessed_or_wav_mimetype(monkeypatch): adapter, storage_mgr, _ = _make_adapter() - chain = [{'type': 'Voice', 'path': f'{_UPLOAD_PREFIX}clip.wav'}] + path = f'{_UPLOAD_PREFIX}clip.unknown' + chain = [{'type': 'Voice', 'path': path}] + guess_type = Mock(return_value=(None, None)) + monkeypatch.setattr('mimetypes.guess_type', guess_type) + await adapter._process_image_components(_make_connection(), chain) - assert chain[0]['base64'].startswith('data:audio/') + + guess_type.assert_called_once_with(path) + assert chain[0]['base64'] == 'data:audio/wav;base64,aGVsbG8=' assert chain[0]['path'] == '' - storage_mgr.delete_scoped_object_key.assert_awaited_once() + storage_mgr.delete_scoped_object_key.assert_awaited_once_with(_CONTEXT, path, expected_owner_type='upload_image') @pytest.mark.asyncio