from __future__ import annotations from unittest.mock import AsyncMock, MagicMock, patch import pytest from yuxi.channels.adapters.nextcloudtalk.normalizer import normalize_inbound from yuxi.channels.exceptions import ChannelNotConnectedError from yuxi.channels.models import ( ChannelType, ChatType, EventType, MessageType, ) class TestNormalizeInbound: def test_normal_text_message(self): raw = { "id": 1234, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "Hello World", "messageType": "comment", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.identity.channel_id == "nextcloud-talk" assert msg.identity.channel_type == ChannelType.NEXTCLOUDTALK assert msg.identity.channel_user_id == "user-1" assert msg.identity.channel_chat_id == "conv-abc" assert msg.identity.channel_message_id == "1234" assert msg.content == "Hello World" assert msg.message_type == MessageType.TEXT assert msg.chat_type == ChatType.DIRECT def test_normal_command_message(self): raw = { "id": 1235, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "/help", "messageType": "command", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.message_type == MessageType.COMMAND assert msg.content == "/help" def test_message_with_reply(self): raw = { "id": 1236, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "reply text", "messageType": "comment", "parent": {"id": 1234}, } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.reply_to_message_id == "1234" def test_message_with_attachment(self): raw = { "id": 1237, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "check this file", "messageType": "comment", "messageParameters": { "file1": { "type": "file", "link": "https://nc.example.com/f/123", "name": "document.pdf", "mimetype": "application/pdf", "size": 1024, }, }, } msg = normalize_inbound(raw, "nextcloud-talk") assert len(msg.attachments) == 1 assert msg.attachments[0].type == "file" assert msg.attachments[0].filename == "document.pdf" assert msg.attachments[0].url == "https://nc.example.com/f/123" assert msg.attachments[0].size_bytes == 1024 def test_system_message_user_joined(self): raw = { "id": 1238, "token": "conv-abc", "actorType": "users", "actorId": "user-2", "actorDisplayName": "New User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "user_added", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.MEMBER_JOINED assert msg.content == "user_added" def test_system_message_user_removed(self): raw = { "id": 1239, "token": "conv-abc", "actorType": "users", "actorId": "user-3", "actorDisplayName": "Left User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "user_removed", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.MEMBER_LEFT def test_system_message_message_edited(self): raw = { "id": 1242, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "message_edited", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.MESSAGE_UPDATED def test_system_message_recording_started(self): raw = { "id": 1243, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "recording_started", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT assert msg.content == "recording_started" def test_system_message_breakout_rooms_started(self): raw = { "id": 1244, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "breakout_rooms_started", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT def test_system_message_poll_voted(self): raw = { "id": 1245, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "poll_voted", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT def test_system_message_reaction_revoked(self): raw = { "id": 1246, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "reaction_revoked", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT def test_unknown_system_message_defaults_to_system_event(self): raw = { "id": 1247, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "unknown_event_type", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT def test_group_chat_message(self): raw = { "id": 1240, "token": "group-conv-1", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "Hello Group", "messageType": "comment", } msg = normalize_inbound(raw, "nextcloud-talk", room_type=2) assert msg.chat_type == ChatType.GROUP def test_metadata_fields(self): raw = { "id": 1241, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "test", "messageType": "comment", "referenceId": "ref-123", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.metadata["nc_message_type"] == "comment" assert msg.metadata["nc_actor_type"] == "users" assert msg.metadata["nc_actor_display_name"] == "Test User" assert msg.metadata["reference_id"] == "ref-123" class TestFormatOutbound: def test_format_text_message(self): from yuxi.channels.adapters.nextcloudtalk.formatter import format_outbound from yuxi.channels.models import ChannelIdentity, ChannelResponse response = ChannelResponse( identity=ChannelIdentity( channel_id="nextcloud-talk", channel_type=ChannelType.NEXTCLOUDTALK, channel_user_id="bot", channel_chat_id="conv-abc", channel_message_id="msg-1", ), content="Hello from bot", message_type=MessageType.TEXT, ) payloads = format_outbound(response, "ForcePilot Bot") assert isinstance(payloads, list) payload = payloads[0] assert payload["token"] == "conv-abc" assert payload["message"] == "Hello from bot" assert payload["actorDisplayName"] == "ForcePilot Bot" assert "referenceId" in payload assert payload["referenceId"].startswith("fp-") assert len(payload["referenceId"]) > 20 def test_format_with_reply(self): from yuxi.channels.adapters.nextcloudtalk.formatter import format_outbound from yuxi.channels.models import ChannelIdentity, ChannelResponse response = ChannelResponse( identity=ChannelIdentity( channel_id="nextcloud-talk", channel_type=ChannelType.NEXTCLOUDTALK, channel_user_id="bot", channel_chat_id="conv-abc", ), content="Reply text", reply_to_message_id="1234", ) payloads = format_outbound(response, "ForcePilot Bot") payload = payloads[0] assert payload["replyTo"] == "1234" def test_format_with_reference_id(self): from yuxi.channels.adapters.nextcloudtalk.formatter import format_outbound from yuxi.channels.models import ChannelIdentity, ChannelResponse response = ChannelResponse( identity=ChannelIdentity( channel_id="nextcloud-talk", channel_type=ChannelType.NEXTCLOUDTALK, channel_user_id="bot", channel_chat_id="conv-abc", ), content="test", metadata={"reference_id": "custom-ref-1"}, ) payloads = format_outbound(response, "ForcePilot Bot") payload = payloads[0] assert payload["referenceId"] == "custom-ref-1" def test_media_message_raises(self): from yuxi.channels.adapters.nextcloudtalk.formatter import format_outbound from yuxi.channels.models import ChannelIdentity, ChannelResponse response = ChannelResponse( identity=ChannelIdentity( channel_id="nextcloud-talk", channel_type=ChannelType.NEXTCLOUDTALK, channel_user_id="bot", channel_chat_id="conv-abc", ), content="image caption", message_type=MessageType.IMAGE, ) with pytest.raises(ValueError, match="Media messages must use send_media"): format_outbound(response, "ForcePilot Bot") class TestProbe: def test_resolve_api_version_v18(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({"version": "18.0.0"}) == "v4" def test_resolve_api_version_v20(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({"version": "20.1.0"}) == "v4" def test_resolve_api_version_v14(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({"version": "14.0.0"}) == "v3" def test_resolve_api_version_v12(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({"version": "12.0.0"}) == "v3" def test_resolve_api_version_legacy(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({"version": "10.0.0"}) == "v1" def test_resolve_api_version_unknown(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_version assert resolve_api_version({}) == "v1" def test_resolve_api_base(self): from yuxi.channels.adapters.nextcloudtalk.probe import resolve_api_base assert resolve_api_base("v4") == "/ocs/v2.php/apps/spreed/api/v4" assert resolve_api_base("v3") == "/ocs/v2.php/apps/spreed/api/v3" assert resolve_api_base("v1") == "/ocs/v2.php/apps/spreed/api/v1" class TestSession: def test_resolve_dm_route(self): from yuxi.channels.adapters.nextcloudtalk.session import resolve_session_route result = resolve_session_route("dm-token", {"type": 1}, "default") assert "dm:dm-token" in result["route"] def test_resolve_group_route(self): from yuxi.channels.adapters.nextcloudtalk.session import resolve_session_route result = resolve_session_route("group-token", {"type": 2}, "default") assert "group:group-token" in result["route"] def test_resolve_chat_type(self): from yuxi.channels.adapters.nextcloudtalk.session import resolve_chat_type assert resolve_chat_type(1) == ChatType.DIRECT assert resolve_chat_type(2) == ChatType.GROUP assert resolve_chat_type(3) == ChatType.GROUP class TestSecurity: def test_dm_policy_open(self): from yuxi.channels.adapters.nextcloudtalk.security import check_dm_policy assert check_dm_policy("any-user", {"dm_policy": "open"}) is True def test_dm_policy_disabled(self): from yuxi.channels.adapters.nextcloudtalk.security import check_dm_policy assert check_dm_policy("any-user", {"dm_policy": "disabled"}) is False def test_dm_policy_allowlist_match(self): from yuxi.channels.adapters.nextcloudtalk.security import check_dm_policy assert check_dm_policy("user-1", {"dm_policy": "allowlist", "allow_from": ["nc:user-1"]}) is True def test_dm_policy_allowlist_no_match(self): from yuxi.channels.adapters.nextcloudtalk.security import check_dm_policy assert check_dm_policy("user-2", {"dm_policy": "allowlist", "allow_from": ["nc:user-1"]}) is False def test_group_policy_open(self): from yuxi.channels.adapters.nextcloudtalk.security import check_group_policy assert check_group_policy("token-1", "user-1", {"group_policy": "open"}) is True def test_group_policy_disabled(self): from yuxi.channels.adapters.nextcloudtalk.security import check_group_policy assert check_group_policy("token-1", "user-1", {"group_policy": "disabled"}) is False def test_group_policy_allowlist_match(self): from yuxi.channels.adapters.nextcloudtalk.security import check_group_policy config = { "group_policy": "allowlist", "groups": {"token-1": {"allow_from": ["nc:user-1"]}}, } assert check_group_policy("token-1", "user-1", config) is True def test_group_policy_allowlist_no_match(self): from yuxi.channels.adapters.nextcloudtalk.security import check_group_policy config = { "group_policy": "allowlist", "groups": {"token-1": {"allow_from": ["nc:user-1"]}}, } assert check_group_policy("token-1", "user-2", config) is False def test_group_policy_global_allowlist_match(self): from yuxi.channels.adapters.nextcloudtalk.security import check_group_policy config = { "group_policy": "allowlist", "groups": {"token-1": {"allow_from": ["nc:user-1"]}}, "group_allow_from": ["nc:user-3"], } assert check_group_policy("token-1", "user-3", config) is True assert check_group_policy("token-1", "user-2", config) is False def test_pairing(self): from yuxi.channels.adapters.nextcloudtalk.security import check_dm_policy, pair_user, unpair_user pair_user("paired-user") assert check_dm_policy("paired-user", {"dm_policy": "pairing"}) is True unpair_user("paired-user") assert check_dm_policy("paired-user", {"dm_policy": "pairing"}) is False class TestSendRetry: @pytest.mark.asyncio async def test_retry_config_defaults(self): from yuxi.channels.adapters.nextcloudtalk.send import _retry_config rc = _retry_config({}) assert rc["max_attempts"] == 3 assert rc["min_delay_ms"] == 400 assert rc["max_delay_ms"] == 30000 assert rc["jitter"] == 0.1 @pytest.mark.asyncio async def test_retry_config_custom(self): from yuxi.channels.adapters.nextcloudtalk.send import _retry_config config = {"retry": {"attempts": 5, "min_delay_ms": 100, "max_delay_ms": 60000, "jitter": 0.2}} rc = _retry_config(config) assert rc["max_attempts"] == 5 assert rc["min_delay_ms"] == 100 assert rc["max_delay_ms"] == 60000 assert rc["jitter"] == 0.2 @pytest.mark.asyncio async def test_with_retry_success_first_attempt(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry from yuxi.channels.models import DeliveryResult call_count = 0 async def _do(): nonlocal call_count call_count += 1 return DeliveryResult(success=True, message_id="msg-1") result = await _with_retry(_do, {}) assert result.success is True assert result.message_id == "msg-1" assert call_count == 1 @pytest.mark.asyncio async def test_with_retry_success_after_failure(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry from yuxi.channels.models import DeliveryResult call_count = 0 async def _do(): nonlocal call_count call_count += 1 if call_count < 2: raise ConnectionError("transient") return DeliveryResult(success=True, message_id="msg-1") result = await _with_retry(_do, {"retry": {"attempts": 3, "min_delay_ms": 10, "max_delay_ms": 100}}) assert result.success is True assert call_count == 2 @pytest.mark.asyncio async def test_with_retry_all_attempts_fail(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry from yuxi.channels.exceptions import DeliveryFailedError async def _do(): raise ConnectionError("always fails") with pytest.raises(DeliveryFailedError): await _with_retry(_do, {"retry": {"attempts": 2, "min_delay_ms": 10, "max_delay_ms": 100}}) @pytest.mark.asyncio async def test_with_retry_http_401_raises(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry from yuxi.channels.exceptions import ChannelAuthenticationError import aiohttp async def _do(): resp = MagicMock() resp.status = 401 raise aiohttp.ClientResponseError( request_info=MagicMock(), history=(), status=401, message="Unauthorized", ) with pytest.raises(ChannelAuthenticationError): await _with_retry(_do, {}) @pytest.mark.asyncio async def test_with_retry_http_404_returns_failure(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry import aiohttp async def _do(): resp = MagicMock() resp.status = 404 raise aiohttp.ClientResponseError( request_info=MagicMock(), history=(), status=404, message="Not Found", ) result = await _with_retry(_do, {}) assert result.success is False assert "HTTP 404" in result.error @pytest.mark.asyncio async def test_with_retry_http_429_retries_then_raises(self): from yuxi.channels.adapters.nextcloudtalk.send import _with_retry from yuxi.channels.exceptions import ChannelRateLimitError import aiohttp async def _do(): resp = MagicMock() resp.status = 429 resp.headers = {"Retry-After": "1"} raise aiohttp.ClientResponseError( request_info=MagicMock(), history=(), status=429, message="Rate Limited", headers={"Retry-After": "1"}, ) with pytest.raises(ChannelRateLimitError): await _with_retry(_do, {"retry": {"attempts": 1, "min_delay_ms": 10, "max_delay_ms": 100}}) class TestClient: def test_session_property_raises_channel_not_connected(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") with pytest.raises(ChannelNotConnectedError): _ = client.session @pytest.mark.asyncio async def test_session_property_after_start(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") await client.start() try: session = client.session assert session is not None finally: await client.stop() @pytest.mark.asyncio async def test_client_user_agent_header(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") await client.start() try: headers = client.session.headers assert "User-Agent" in headers finally: await client.stop() def test_auth_headers(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") headers = client.auth_headers() assert "Authorization" in headers assert headers["Authorization"].startswith("Basic ") def test_api_url(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") assert client.api_url("/ocs/v2.php/cloud/capabilities") == "https://nc.example.com/ocs/v2.php/cloud/capabilities" class TestSecurityLazyLoading: def test_lazy_loading_no_immediate_file_read(self, tmp_path, monkeypatch): import importlib import yuxi.channels.adapters.nextcloudtalk.security as sec_mod monkeypatch.setattr(sec_mod, "_PERSIST_PATH", str(tmp_path / ".paired_users.json")) importlib.reload(sec_mod) paired_file = tmp_path / ".paired_users.json" assert not paired_file.exists() or paired_file.stat().st_size == 0 class TestPollingMonitor: def test_update_conversations(self): from yuxi.channels.adapters.nextcloudtalk.monitor_polling import PollingMonitor from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") monitor = PollingMonitor(client, "/ocs/v2.php/apps/spreed/api/v4", "nc", 1.0, 30) monitor.update_conversations({"new-token": {"type": 1, "name": "test"}}) assert "new-token" in monitor._conversations assert monitor._conversations["new-token"]["type"] == 1 class TestDownloadMediaValidation: @pytest.mark.asyncio async def test_download_media_rejects_external_url(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.server_url = "https://nc.example.com" with pytest.raises(ValueError, match="same Nextcloud instance"): await adapter.download_media("https://evil.com/f/123") @pytest.mark.asyncio async def test_download_media_rejects_path_traversal(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.server_url = "https://nc.example.com" with pytest.raises(ValueError, match="path traversal"): await adapter.download_media("https://nc.example.com/../etc/passwd") class TestChannelType: def test_nextcloudtalk_in_enum(self): assert ChannelType.NEXTCLOUDTALK == "nextcloud-talk" assert ChannelType("nextcloud-talk") == ChannelType.NEXTCLOUDTALK class TestRoomInfo: def test_map_room_type_direct(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _map_room_type assert _map_room_type(1) == "direct" assert _map_room_type(5) == "direct" assert _map_room_type(6) == "direct" def test_map_room_type_group(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _map_room_type assert _map_room_type(2) == "group" assert _map_room_type(3) == "group" assert _map_room_type(4) == "group" def test_map_room_type_none(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _map_room_type assert _map_room_type(None) is None assert _map_room_type(99) is None def test_kind_result_direct(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _kind_result result = _kind_result("direct") assert result["kind"] == "direct" assert result["chat_type"] == ChatType.DIRECT def test_kind_result_group(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _kind_result result = _kind_result("group") assert result["kind"] == "group" assert result["chat_type"] == ChatType.GROUP def test_kind_result_fallback(self): from yuxi.channels.adapters.nextcloudtalk.room_info import _kind_result result = _kind_result(None) assert result["kind"] is None assert result["source"] == "cache_fallback" def test_invalidate_room_cache(self): from yuxi.channels.adapters.nextcloudtalk.room_info import ( _room_kind_cache, invalidate_room_cache, ) _room_kind_cache["test-token"] = (0, "direct") invalidate_room_cache("test-token") assert "test-token" not in _room_kind_cache _room_kind_cache["a"] = (0, "group") _room_kind_cache["b"] = (0, "direct") invalidate_room_cache() assert len(_room_kind_cache) == 0 class TestReactionOperations: @pytest.mark.asyncio async def test_remove_reaction_returns_delivery_result(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.delete = AsyncMock(return_value={ "ocs": {"meta": {"statuscode": 200}, "data": {}}, }) result = await adapter.remove_reaction("chat-1", "msg-1", "👍") assert result.success is True @pytest.mark.asyncio async def test_get_reactions_returns_metadata(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.get = AsyncMock(return_value={ "ocs": {"meta": {"statuscode": 200}, "data": {"👍": 3, "❤️": 1}}, }) result = await adapter.get_reactions("chat-1", "msg-1") assert result.success is True assert "reactions" in result.metadata class TestCredentialSource: def test_resolve_with_source_config(self): from yuxi.channels.adapters.nextcloudtalk.credentials import resolve_credential_with_source value, source = resolve_credential_with_source("direct_value", None, None) assert value == "direct_value" assert source == "config" def test_resolve_with_source_env(self): from yuxi.channels.adapters.nextcloudtalk.credentials import resolve_credential_with_source value, source = resolve_credential_with_source(None, None, "PATH") assert value assert source == "env" def test_resolve_with_source_none(self): from yuxi.channels.adapters.nextcloudtalk.credentials import resolve_credential_with_source value, source = resolve_credential_with_source(None, None, "NONEXISTENT_ENV_VAR_12345") assert value == "" assert source == "none" class TestDMConfig: def test_resolve_dm_config_exact_match(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_config config = {"dms": {"user-1": {"enabled": False, "systemPrompt": "custom"}}} dm_cfg = resolve_dm_config("user-1", config) assert dm_cfg["enabled"] is False assert dm_cfg["systemPrompt"] == "custom" def test_resolve_dm_config_wildcard(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_config config = {"dms": {"*": {"enabled": False}}} dm_cfg = resolve_dm_config("unknown-user", config) assert dm_cfg["enabled"] is False def test_resolve_dm_config_merge(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_config config = { "dms": { "*": {"enabled": True, "systemPrompt": "base"}, "user-1": {"systemPrompt": "override"}, } } dm_cfg = resolve_dm_config("user-1", config) assert dm_cfg["enabled"] is True assert dm_cfg["systemPrompt"] == "override" def test_resolve_dm_enabled(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_enabled config = {"dms": {"user-1": {"enabled": False}}} assert resolve_dm_enabled("user-1", config) is False assert resolve_dm_enabled("user-2", {"dms": {}}) is True def test_resolve_dm_system_prompt(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_system_prompt config = {"dms": {"user-1": {"system_prompt": "hello"}}} assert resolve_dm_system_prompt("user-1", config) == "hello" assert resolve_dm_system_prompt("user-2", config) is None def test_resolve_dm_skills(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_dm_skills config = {"dms": {"user-1": {"skills": ["skill-a", "skill-b"]}}} assert resolve_dm_skills("user-1", config) == ["skill-a", "skill-b"] class TestHistoryLimit: def test_resolve_history_limit_group_default(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_history_limit assert resolve_history_limit("token-1", {}, "group") == 0 def test_resolve_history_limit_group_config(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_history_limit assert resolve_history_limit("token-1", {"historyLimit": 50}, "group") == 50 def test_resolve_history_limit_group_room(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_history_limit config = {"rooms": {"token-1": {"historyLimit": 25}}} assert resolve_history_limit("token-1", config, "group") == 25 def test_resolve_history_limit_dm(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_history_limit config = {"dms": {"user-1": {"historyLimit": 10}}} assert resolve_history_limit("user-1", config, "direct") == 10 def test_resolve_history_limit_dm_global(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_history_limit assert resolve_history_limit("user-1", {"dmHistoryLimit": 30}, "direct") == 30 class TestDoctorDiagnostics: def test_check_credentials_pass(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_credentials result = _check_credentials({"bot_user": "bot", "app_password": "secret"}) assert result["status"] == "pass" def test_check_credentials_fail(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_credentials result = _check_credentials({"bot_user": "", "app_password": ""}) assert result["status"] == "fail" def test_check_server_url_pass(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_server_url result = _check_server_url({"server_url": "https://nc.example.com"}) assert result["status"] == "pass" def test_check_server_url_fail_missing(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_server_url result = _check_server_url({"server_url": ""}) assert result["status"] == "fail" def test_check_room_configs_empty(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_room_configs result = _check_room_configs({}) assert result["detail"] == "No per-room configuration" def test_check_dm_configs_empty(self): from yuxi.channels.adapters.nextcloudtalk.doctor import _check_dm_configs result = _check_dm_configs({}) assert result["detail"] == "No per-DM configuration" class TestStatusSnapshot: def test_build_snapshot_basic(self): from yuxi.channels.adapters.nextcloudtalk.status import build_snapshot from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({"server_url": "https://nc.example.com", "bot_user": "bot", "app_password": "pw"}) snapshot = build_snapshot(adapter) assert snapshot.account_id == "nextcloud-talk" assert snapshot.configured is True assert snapshot.base_url == "https://nc.example.com" assert snapshot.bot_token_source == "config" def test_build_snapshot_not_configured(self): from yuxi.channels.adapters.nextcloudtalk.status import build_snapshot from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({}) snapshot = build_snapshot(adapter) assert snapshot.configured is False assert snapshot.status_state == "not-configured" class TestApprovalAuth: def test_check_approval_open(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import check_approval_required result = check_approval_required("user-1", {"dm_policy": "open"}) assert result["approval_required"] is False def test_check_approval_disabled(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import check_approval_required result = check_approval_required("user-1", {"dm_policy": "disabled"}) assert result["approval_required"] is True assert result["action"] == "block" def test_check_approval_allowlist_match(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import check_approval_required result = check_approval_required("nc:user-1", {"dm_policy": "allowlist", "allow_from": ["nc:user-1"]}) assert result["approval_required"] is False def test_check_approval_allowlist_no_match(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import check_approval_required result = check_approval_required("nc:user-2", {"dm_policy": "allowlist", "allow_from": ["nc:user-1"]}) assert result["approval_required"] is True def test_check_approval_pairing_not_paired(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import check_approval_required result = check_approval_required("user-1", {"dm_policy": "pairing"}, paired_users=set()) assert result["approval_required"] is True assert result["action"] == "pair" def test_resolve_approver_id(self): from yuxi.channels.adapters.nextcloudtalk.approval_auth import resolve_approver_id assert resolve_approver_id("nc:user-1") == "nc:user-1" assert resolve_approver_id("nextcloudtalk:user-1") == "nc:user-1" class TestSecretContract: def test_list_secret_targets(self): from yuxi.channels.adapters.nextcloudtalk.secret_contract import list_secret_targets targets = list_secret_targets() assert len(targets) >= 4 assert any(t["key"] == "app_password" and t["secret"] for t in targets) def test_verify_secret_contract_valid(self): from yuxi.channels.adapters.nextcloudtalk.secret_contract import verify_secret_contract result = verify_secret_contract({ "bot_user": "bot", "app_password": "secret", "server_url": "https://nc.example.com", }) assert result["valid"] is True def test_verify_secret_contract_missing_creds(self): from yuxi.channels.adapters.nextcloudtalk.secret_contract import verify_secret_contract result = verify_secret_contract({"bot_user": ""}) assert result["valid"] is False assert any("bot_user" in issue or "app_password" in issue for issue in result["issues"]) def test_collect_runtime_secrets(self): from yuxi.channels.adapters.nextcloudtalk.secret_contract import collect_runtime_secrets secrets = collect_runtime_secrets({ "bot_user": "bot", "app_password": "secret", "api_user": "api", "api_password": "api_secret", }) assert secrets["bot_user"]["present"] is True assert secrets["app_password"]["present"] is True assert secrets["app_password"]["value"] == "***" class TestConfigSchema: def test_validate_config_minimal(self): from yuxi.channels.adapters.nextcloudtalk.config_schema import validate_config config = validate_config({ "serverUrl": "https://nc.example.com", "botUser": "bot", "appPassword": "secret", }) assert config.server_url == "https://nc.example.com" assert config.bot_user == "bot" assert config.dm_policy == "pairing" def test_validate_config_invalid_url(self): import pytest from yuxi.channels.adapters.nextcloudtalk.config_schema import validate_config with pytest.raises(ValueError, match="server_url"): validate_config({"serverUrl": "ftp://invalid", "botUser": "bot", "appPassword": "secret"}) def test_validate_config_invalid_dm_policy(self): import pytest from yuxi.channels.adapters.nextcloudtalk.config_schema import validate_config with pytest.raises(ValueError, match="dm_policy"): validate_config({ "serverUrl": "https://nc.example.com", "botUser": "bot", "appPassword": "secret", "dmPolicy": "unknown", }) def test_validate_config_with_rooms(self): from yuxi.channels.adapters.nextcloudtalk.config_schema import validate_config config = validate_config({ "serverUrl": "https://nc.example.com", "botUser": "bot", "appPassword": "secret", "rooms": {"room-1": {"enabled": False, "systemPrompt": "custom"}}, "dms": {"user-1": {"enabled": True}}, }) assert "room-1" in config.rooms assert config.rooms["room-1"].enabled is False def test_validate_config_aliases(self): from yuxi.channels.adapters.nextcloudtalk.config_schema import validate_config config = validate_config({ "serverUrl": "https://nc.example.com", "botUser": "bot", "appPassword": "secret", "historyLimit": 50, "dmHistoryLimit": 30, "responsePrefix": "[Bot]", }) assert config.history_limit == 50 assert config.dm_history_limit == 30 assert config.response_prefix == "[Bot]" class TestNormalizerPinAndForward: def test_pin_event_detection(self): raw = { "id": 1250, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "message_pinned", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT assert msg.content == "message_pinned" def test_unpin_event_detection(self): raw = { "id": 1251, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "", "messageType": "system", "systemMessage": "message_unpinned", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.event_type == EventType.SYSTEM_EVENT def test_forwarded_message_metadata(self): raw = { "id": 1252, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "check this {mention-forwarded}", "messageType": "comment", "forwardActorDisplayName": "Original Sender", "forwardActorId": "user-2", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.metadata["nc_forwarded"] is True assert msg.metadata["nc_forward_actor_display_name"] == "Original Sender" assert msg.metadata["nc_forward_actor_id"] == "user-2" def test_non_forwarded_message(self): raw = { "id": 1253, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "normal message", "messageType": "comment", } msg = normalize_inbound(raw, "nextcloud-talk") assert "nc_forwarded" not in msg.metadata class TestAliasSupport: def test_adapter_aliases_in_meta(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter assert "nc-talk" in NextcloudTalkAdapter.meta.aliases assert "nc" in NextcloudTalkAdapter.meta.aliases assert "nextcloud-talk" in NextcloudTalkAdapter.meta.aliases def test_adapter_registered_with_deprecated_alias(self): from yuxi.channels.registry import BUILTIN_ADAPTERS, _BUILTIN_ADAPTER_ALIASES assert "nextcloud-talk" in BUILTIN_ADAPTERS assert _BUILTIN_ADAPTER_ALIASES.get("nc-talk") == "nextcloud-talk" assert _BUILTIN_ADAPTER_ALIASES.get("nc") == "nextcloud-talk" assert _BUILTIN_ADAPTER_ALIASES.get("nextcloud-talk") == "nextcloud-talk" class TestHMACSigning: def test_compute_bot_signature(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature sig = _compute_bot_signature(b"test body", "random-hex-32bytes-nonce", "my-secret") assert isinstance(sig, str) assert len(sig) > 0 def test_compute_bot_signature_deterministic(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature sig1 = _compute_bot_signature(b"hello", "fixed-random", "secret") sig2 = _compute_bot_signature(b"hello", "fixed-random", "secret") assert sig1 == sig2 def test_compute_bot_signature_different_body_different_sig(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature sig1 = _compute_bot_signature(b"hello", "fixed-random", "secret") sig2 = _compute_bot_signature(b"world", "fixed-random", "secret") assert sig1 != sig2 def test_compute_bot_signature_different_secret_different_sig(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature sig1 = _compute_bot_signature(b"hello", "fixed-random", "secret1") sig2 = _compute_bot_signature(b"hello", "fixed-random", "secret2") assert sig1 != sig2 def test_compute_bot_signature_different_nonce_different_sig(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature sig1 = _compute_bot_signature(b"hello", "random-aaa", "secret") sig2 = _compute_bot_signature(b"hello", "random-bbb", "secret") assert sig1 != sig2 def test_verify_hmac_signature_valid(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature, verify_hmac_signature random_hex = "test-random-nonce" sig = _compute_bot_signature(b"test body", random_hex, "my-secret") assert verify_hmac_signature(b"test body", random_hex, sig, "my-secret") is True def test_verify_hmac_signature_invalid(self): from yuxi.channels.adapters.nextcloudtalk.client import verify_hmac_signature assert verify_hmac_signature(b"test body", "random", "invalid-sig", "my-secret") is False def test_verify_hmac_signature_empty_random(self): from yuxi.channels.adapters.nextcloudtalk.client import verify_hmac_signature assert verify_hmac_signature(b"test body", "", "some-sig", "my-secret") is False def test_verify_hmac_signature_empty_header(self): from yuxi.channels.adapters.nextcloudtalk.client import verify_hmac_signature assert verify_hmac_signature(b"test body", "random", "", "my-secret") is False def test_verify_hmac_signature_empty_secret(self): from yuxi.channels.adapters.nextcloudtalk.client import verify_hmac_signature assert verify_hmac_signature(b"test body", "random", "some-sig", "") is False def test_client_sign_body_with_secret(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass", bot_secret="secret") headers = client._sign_body(b"hello") assert "X-Nextcloud-Talk-Bot-Random" in headers assert "X-Nextcloud-Talk-Bot-Signature" in headers assert len(headers["X-Nextcloud-Talk-Bot-Random"]) == 64 def test_client_sign_body_without_secret(self): from yuxi.channels.adapters.nextcloudtalk.client import NextcloudTalkClient client = NextcloudTalkClient("https://nc.example.com", "bot", "pass") headers = client._sign_body(b"hello") assert headers == {} class TestWebhookSignature: @pytest.mark.asyncio async def test_verify_webhook_signature_valid(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({"app_password": "my-secret"}) body = b'{"test": "data"}' random_hex = "test-nonce-for-webhook" sig = _compute_bot_signature(body, random_hex, "my-secret") result = await adapter.verify_webhook_signature( {"X-Nextcloud-Talk-Bot-Signature": sig, "X-Nextcloud-Talk-Bot-Random": random_hex}, body ) assert result is True @pytest.mark.asyncio async def test_verify_webhook_signature_invalid(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({"app_password": "my-secret"}) result = await adapter.verify_webhook_signature( {"X-Nextcloud-Talk-Bot-Signature": "bad-signature", "X-Nextcloud-Talk-Bot-Random": "random"}, b"body" ) assert result is False @pytest.mark.asyncio async def test_verify_webhook_signature_no_header(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({"app_password": "my-secret"}) result = await adapter.verify_webhook_signature({}, b"body") assert result is False @pytest.mark.asyncio async def test_verify_webhook_signature_no_secret(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({}) result = await adapter.verify_webhook_signature( {"X-Nextcloud-Talk-Bot-Signature": "some-sig", "X-Nextcloud-Talk-Bot-Random": "random"}, b"body" ) assert result is False @pytest.mark.asyncio async def test_verify_webhook_signature_lowercase_header(self): from yuxi.channels.adapters.nextcloudtalk.client import _compute_bot_signature from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter({"app_password": "my-secret"}) body = b"hello" random_hex = "lowercase-test-nonce" sig = _compute_bot_signature(body, random_hex, "my-secret") result = await adapter.verify_webhook_signature( {"x-nextcloud-talk-bot-signature": sig, "x-nextcloud-talk-bot-random": random_hex}, body ) assert result is True class TestBackendSourceValidation: def test_validate_backend_source_match(self): from yuxi.channels.adapters.nextcloudtalk.security import validate_backend_source assert validate_backend_source( {"X-Nextcloud-Talk-Backend": "https://nc.example.com/ocs/v2.php"}, "https://nc.example.com", ) is True def test_validate_backend_source_mismatch(self): from yuxi.channels.adapters.nextcloudtalk.security import validate_backend_source assert validate_backend_source( {"X-Nextcloud-Talk-Backend": "https://evil.com"}, "https://nc.example.com", ) is False def test_validate_backend_source_missing_header(self): from yuxi.channels.adapters.nextcloudtalk.security import validate_backend_source assert validate_backend_source({}, "https://nc.example.com") is False def test_validate_backend_source_lowercase_header(self): from yuxi.channels.adapters.nextcloudtalk.security import validate_backend_source assert validate_backend_source( {"x-nextcloud-talk-backend": "https://nc.example.com"}, "https://nc.example.com", ) is True class TestAS2Normalization: def test_as2_basic_create_activity(self): raw = { "@context": "https://www.w3.org/ns/activitystreams", "type": "Create", "actor": {"id": "user-1", "name": "Test User"}, "object": {"id": "msg-100", "content": "AS2 message"}, "target": {"id": "conv-abc"}, "published": 1700000000, } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.identity.channel_user_id == "user-1" assert msg.identity.channel_chat_id == "conv-abc" assert msg.identity.channel_message_id == "msg-100" assert msg.content == "AS2 message" assert msg.metadata["nc_actor_display_name"] == "Test User" assert msg.metadata["as2_type"] == "Create" assert msg.event_type == EventType.MESSAGE_RECEIVED def test_as2_delete_activity(self): raw = { "@context": "https://www.w3.org/ns/activitystreams", "type": "Delete", "actor": {"id": "user-1", "name": "Test User"}, "object": {"id": "msg-101"}, "published": 1700000000, "messageType": "system", "systemMessage": "message_deleted", "token": "conv-abc", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.metadata["as2_type"] == "Delete" assert msg.event_type == EventType.MESSAGE_DELETED def test_as2_with_string_actor(self): raw = { "@context": "https://www.w3.org/ns/activitystreams", "type": "Create", "actor": "user-1", "actorId": "user-1", "actorDisplayName": "Test User", "object": "msg-100", "id": "msg-100", "token": "conv-abc", "timestamp": 1700000000, "message": "simple message", "messageType": "comment", } msg = normalize_inbound(raw, "nextcloud-talk") assert msg.content == "simple message" assert msg.metadata["as2_type"] == "Create" def test_non_as2_message_no_as2_metadata(self): raw = { "id": 2000, "token": "conv-abc", "actorType": "users", "actorId": "user-1", "actorDisplayName": "Test User", "timestamp": 1700000000, "message": "regular NC message", "messageType": "comment", } msg = normalize_inbound(raw, "nextcloud-talk") assert "as2_type" not in msg.metadata assert msg.content == "regular NC message" class TestDedupGuardEmptyValues: def test_claim_returns_true_when_token_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() assert guard.claim("", "msg-1") is True def test_claim_returns_true_when_message_id_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() assert guard.claim("token-1", "") is True def test_commit_noop_when_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() guard.commit("", "msg-1") guard.commit("token-1", "") assert guard.stats()["committed"] == 0 def test_release_noop_when_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() guard.release("", "msg-1") guard.release("token-1", "") assert guard.stats()["pending"] == 0 def test_is_invalid_returns_false_when_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() assert guard.is_invalid("", "msg-1") is False assert guard.is_invalid("token-1", "") is False def test_is_duplicate_returns_false_when_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() assert guard.is_duplicate("", "msg-1") is False assert guard.is_duplicate("token-1", "") is False def test_mark_invalid_noop_when_empty(self): from yuxi.channels.adapters.nextcloudtalk.dedup import NextcloudTalkDedupGuard guard = NextcloudTalkDedupGuard() guard.mark_invalid("", "msg-1") guard.mark_invalid("token-1", "") assert guard.stats()["invalid"] == 0 class TestMatchSource: def test_match_source_direct(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_room_config config = {"rooms": {"room-1": {"enabled": True}, "*": {"enabled": False}}} result = resolve_room_config("room-1", config) assert result["_match_source"] == "direct" assert result["enabled"] is True def test_match_source_wildcard(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_room_config config = {"rooms": {"*": {"enabled": False}}} result = resolve_room_config("unknown-room", config) assert result["_match_source"] == "wildcard" assert result["enabled"] is False def test_match_source_none(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_room_config result = resolve_room_config("room-1", {}) assert result["_match_source"] == "none" def test_match_source_direct_overrides_wildcard(self): from yuxi.channels.adapters.nextcloudtalk.config import resolve_room_config config = { "rooms": { "room-1": {"enabled": True, "systemPrompt": "custom"}, "*": {"enabled": False, "systemPrompt": "base"}, } } result = resolve_room_config("room-1", config) assert result["_match_source"] == "direct" assert result["enabled"] is True assert result["systemPrompt"] == "custom" class TestActivityRecording: def test_initial_activity_stats(self): from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() stats = adapter.get_activity_stats() assert stats["inbound_count"] == 0 assert stats["outbound_count"] == 0 assert stats["last_inbound_seconds_ago"] is None assert stats["last_outbound_seconds_ago"] is None class TestPollOperations: @pytest.mark.asyncio async def test_create_poll_success(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post = AsyncMock(return_value={ "ocs": {"meta": {"status": "ok"}, "data": {"id": "poll-1"}}, }) result = await adapter.create_poll("chat-1", "Favorite color?", ["Red", "Blue", "Green"]) assert result.success is True assert result.message_id == "poll-1" @pytest.mark.asyncio async def test_create_poll_failure(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post = AsyncMock(return_value={ "ocs": {"meta": {"status": "error", "message": "Invalid options"}}, }) result = await adapter.create_poll("chat-1", "?", []) assert result.success is False class TestVoiceOperations: @pytest.mark.asyncio async def test_send_voice_delegates_to_send_media(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post_form = AsyncMock(return_value={ "ocs": {"meta": {"statuscode": 201}, "data": {"id": "voice-msg-1"}}, }) result = await adapter.send_voice("chat-1", b"fake-audio-data") assert result.success is True assert result.message_id == "voice-msg-1" @pytest.mark.asyncio async def test_transcribe_voice_success(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.get = AsyncMock(return_value={ "ocs": {"meta": {"status": "ok"}, "data": {"transcription": "Hello world"}}, }) result = await adapter.transcribe_voice("chat-1", "msg-1") assert result.success is True assert result.metadata.get("transcription") == "Hello world" @pytest.mark.asyncio async def test_transcribe_voice_not_available(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.get = AsyncMock(return_value={ "ocs": {"meta": {"status": "error"}}, }) result = await adapter.transcribe_voice("chat-1", "msg-1") assert result.success is False class TestImageAnalysis: @pytest.mark.asyncio async def test_analyze_image_success(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.get = AsyncMock(return_value={ "ocs": {"meta": {"status": "ok"}, "data": {"analysis": "A cat sitting on a chair"}}, }) result = await adapter.analyze_image("chat-1", "msg-1") assert result.success is True assert result.metadata.get("analysis") == "A cat sitting on a chair" @pytest.mark.asyncio async def test_analyze_image_with_prompt(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post = AsyncMock(return_value={ "ocs": {"meta": {"status": "ok"}, "data": {"result": "The image contains a dog"}}, }) result = await adapter.analyze_image("chat-1", "msg-1", prompt="What animal is this?") assert result.success is True assert result.metadata.get("analysis") == "The image contains a dog" class TestWebhookURLFallback: def test_public_url_defaults_to_localhost(self): from yuxi.channels.adapters.nextcloudtalk.webhook_server import NextcloudTalkWebhookServer adapter = MagicMock() adapter.channel_id = "nextcloud-talk" adapter.config = {} server = NextcloudTalkWebhookServer(adapter, host="0.0.0.0", port=8788) assert server.public_url == "http://localhost:8788/nextcloud-talk-webhook" def test_public_url_uses_configured_host(self): from yuxi.channels.adapters.nextcloudtalk.webhook_server import NextcloudTalkWebhookServer adapter = MagicMock() adapter.channel_id = "nextcloud-talk" adapter.config = {} server = NextcloudTalkWebhookServer(adapter, host="192.168.1.100", port=9090, path="/webhook") assert server.public_url == "http://192.168.1.100:9090/webhook" def test_public_url_omits_standard_ports(self): from yuxi.channels.adapters.nextcloudtalk.webhook_server import NextcloudTalkWebhookServer adapter = MagicMock() adapter.channel_id = "nextcloud-talk" adapter.config = {} server = NextcloudTalkWebhookServer(adapter, host="example.com", port=80) assert server.public_url == "http://example.com/nextcloud-talk-webhook" server443 = NextcloudTalkWebhookServer(adapter, host="example.com", port=443) assert server443.public_url == "http://example.com/nextcloud-talk-webhook" class TestApprovalOperations: @pytest.mark.asyncio async def test_request_approval_success(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post = AsyncMock(return_value={ "ocs": {"meta": {"status": "ok"}, "data": {"id": "approval-1"}}, }) result = await adapter.request_approval("chat-1", "exec_command", {"cmd": "ls"}) assert result.success is True assert result.message_id == "approval-1" @pytest.mark.asyncio async def test_request_approval_failure(self): from unittest.mock import AsyncMock, MagicMock from yuxi.channels.adapters.nextcloudtalk.adapter import NextcloudTalkAdapter adapter = NextcloudTalkAdapter() adapter._client = MagicMock() adapter._client.post = AsyncMock(return_value={ "ocs": {"meta": {"status": "error", "message": "Denied"}}, }) result = await adapter.request_approval("chat-1", "exec_command", {}) assert result.success is False class TestDebugAccounts: def test_debug_env_var_defaults_false(self): import os old_val = os.environ.pop("OPENCLAW_DEBUG_NEXTCLOUD_TALK_ACCOUNTS", None) try: import importlib import yuxi.channels.adapters.nextcloudtalk.adapter as mod importlib.reload(mod) assert mod._DEBUG_ACCOUNTS is False finally: if old_val is not None: os.environ["OPENCLAW_DEBUG_NEXTCLOUD_TALK_ACCOUNTS"] = old_val def test_debug_env_var_enabled(self, monkeypatch): monkeypatch.setenv("OPENCLAW_DEBUG_NEXTCLOUD_TALK_ACCOUNTS", "1") import importlib import yuxi.channels.adapters.nextcloudtalk.adapter as mod importlib.reload(mod) assert mod._DEBUG_ACCOUNTS is True