access-tool/backend/tests/unit/community_manager/actions/test_refresh_external_sources.py at 6cb9f269074b6567b72adcad015581e51ece3cee · OpenBuilders/access-tool · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import pytest
from unittest.mock import AsyncMock, patch

from sqlalchemy.orm import Session

from community_manager.actions.chat import (
    CommunityManagerTaskChatAction,
    CommunityManagerUserChatAction,
)
from core.dtos.chat.rule.whitelist import WhitelistRuleItemsDifferenceDTO
from core.services.chat.rule.whitelist import TelegramChatExternalSourceService
from tests.factories.chat import TelegramChatFactory, TelegramChatUserFactory
from tests.factories.rule.external_source import (
    TelegramChatWhitelistExternalSourceFactory,
)
from tests.factories.rule.group import TelegramChatRuleGroupFactory
from tests.factories.user import UserFactory


@pytest.mark.asyncio
async def test_refresh_external_sources__removed_user_is_kicked(
    db_session: Session,
):
    chat = TelegramChatFactory.create(is_full_control=True)
    group = TelegramChatRuleGroupFactory.create(chat=chat)

    user_stays = UserFactory.create(telegram_id=1001)
    user_removed = UserFactory.create(telegram_id=1002)

    TelegramChatUserFactory.create(chat=chat, user=user_stays, is_managed=True)
    TelegramChatUserFactory.create(chat=chat, user=user_removed, is_managed=True)

    source = TelegramChatWhitelistExternalSourceFactory.create(
        chat=chat,
        group=group,
        content=[1001, 1002],
        is_enabled=True,
        url="https://example.com/api/whitelist",
    )
    db_session.flush()

    mock_validate = AsyncMock(
        return_value=WhitelistRuleItemsDifferenceDTO(
            previous=[1001, 1002],
            current=[1001],
        )
    )

    action = CommunityManagerTaskChatAction(db_session)

    with patch.object(
        TelegramChatExternalSourceService,
        "validate_external_source",
        mock_validate,
    ), patch.object(
        CommunityManagerUserChatAction,
        "kick_chat_member",
        new_callable=AsyncMock,
    ) as mock_kick:
        await action.refresh_external_sources()

        # User 1002 was removed from the API response and should be kicked
        assert mock_kick.call_count == 1
        kicked_member = mock_kick.call_args.args[0]
        assert kicked_member.user.telegram_id == 1002
        assert kicked_member.chat_id == chat.id

    db_session.refresh(source)
    assert source.content == [1001]


@pytest.mark.asyncio
async def test_refresh_external_sources__no_removed_users__no_kicks(
    db_session: Session,
):
    chat = TelegramChatFactory.create(is_full_control=True)
    group = TelegramChatRuleGroupFactory.create(chat=chat)

    user = UserFactory.create(telegram_id=1001)
    TelegramChatUserFactory.create(chat=chat, user=user, is_managed=True)

    TelegramChatWhitelistExternalSourceFactory.create(
        chat=chat,
        group=group,
        content=[1001],
        is_enabled=True,
        url="https://example.com/api/whitelist",
    )
    db_session.flush()

    mock_validate = AsyncMock(
        return_value=WhitelistRuleItemsDifferenceDTO(
            previous=[1001],
            current=[1001],
        )
    )

    action = CommunityManagerTaskChatAction(db_session)

    with patch.object(
        TelegramChatExternalSourceService,
        "validate_external_source",
        mock_validate,
    ), patch.object(
        CommunityManagerUserChatAction,
        "kick_chat_member",
        new_callable=AsyncMock,
    ) as mock_kick:
        await action.refresh_external_sources()

        assert mock_kick.call_count == 0