#
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License 2.0;
# you may not use this file except in compliance with the Elastic License 2.0.
#
"""Tests the Confluence database source class methods"""

import ssl
from contextlib import asynccontextmanager
from copy import copy
from unittest import mock
from unittest.mock import AsyncMock, MagicMock, Mock, patch

import aiohttp
import pytest
from aiohttp import StreamReader
from aiohttp.client_exceptions import ClientResponseError
from connectors_sdk.filtering.validation import Filter
from connectors_sdk.source import ConfigurableFieldValueError
from freezegun import freeze_time

from connectors.access_control import DLS_QUERY
from connectors.sources.atlassian.confluence import (
    ConfluenceClient,
    ConfluenceDataSource,
)
from connectors.sources.atlassian.confluence.client import (
    BadRequest,
    Forbidden,
    InternalServerError,
    InvalidConfluenceDataSourceTypeError,
    NotFound,
    Unauthorized,
)
from connectors.sources.atlassian.confluence.constants import (
    CONFLUENCE_CLOUD,
    CONFLUENCE_DATA_CENTER,
    CONFLUENCE_SERVER,
    CONTENT_QUERY_CLOUD,
    CONTENT_QUERY_DATA_CENTER,
    SPACE_QUERY_CLOUD,
    SPACE_QUERY_DATA_CENTER,
)
from connectors.utils import ssl_context
from tests.commons import AsyncIterator
from tests.sources.support import create_source

ADVANCED_SNIPPET = "advanced_snippet"
HOST_URL = "http://127.0.0.1:9696"
EXCEPTION_MESSAGE = "Something went wrong"
CONTENT_QUERY = "limit=1&expand=children.attachment,history.lastUpdated,body.storage"
RESPONSE_SPACE = {
    "results": [
        {
            "id": 4554779,
            "name": "DEMO",
            "_links": {
                "webui": "/spaces/DM",
            },
        }
    ],
    "start": 0,
    "limit": 1,
    "size": 1,
    "_links": {},
}

SPACE = {
    "id": 4554779,
    "name": "DEMO",
    "key": "DM",
    "_links": {
        "webui": "/spaces/DM",
    },
    "permissions": [
        {
            "id": 1,
            "subjects": {
                "group": {
                    "results": [
                        {
                            "type": "group",
                            "name": "group1",
                            "id": "group_id_1",
                        }
                    ],
                    "size": 1,
                },
            },
            "operation": {"operation": "read", "targetType": "space"},
        }
    ],
    "history": {
        "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
        "createdDate": "2023-01-03T09:24:50.633Z",
        "createdBy": {"publicName": "user1"},
    },
}

RESPONSE_PAGE = {
    "results": [
        {
            "id": 4779,
            "title": "ES-scrum",
            "type": "page",
            "history": {
                "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                "createdDate": "2023-01-03T09:24:50.633Z",
                "createdBy": {"publicName": "user1"},
            },
            "children": {"attachment": {"size": 2}},
            "body": {"storage": {"value": "This is a test page"}},
            "space": {"name": "DEMO"},
            "_links": {
                "webui": "/spaces/~1234abc/pages/4779/ES-scrum",
            },
            "ancestors": [{"title": "parent_title"}],
        }
    ],
    "start": 0,
    "limit": 1,
    "size": 1,
    "_links": {},
}

RESPONSE_PAGE_WITH_HTML = {
    "results": [
        {
            "id": 4779,
            "title": "ES-scrum",
            "type": "page",
            "history": {
                "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                "createdDate": "2023-01-03T09:24:50.633Z",
                "createdBy": {"publicName": "user1"},
            },
            "children": {"attachment": {"size": 2}},
            "body": {"storage": {"value": "<p>This is a test page</p>"}},
            "space": {"name": "DEMO"},
            "_links": {
                "webui": "/spaces/~1234abc/pages/4779/ES-scrum",
            },
            "ancestors": [{"title": "parent_title"}],
        }
    ],
    "start": 0,
    "limit": 1,
    "size": 1,
    "_links": {},
}

EXPECTED_PAGE = {
    "_id": "4779",
    "type": "page",
    "_timestamp": "2023-01-24T04:07:19.672Z",
    "title": "ES-scrum",
    "ancestors": [{"title": "parent_title"}],
    "body": "This is a test page",
    "space": "DEMO",
    "url": f"{HOST_URL}/spaces/~1234abc/pages/4779/ES-scrum",
    "labels": [None],
    "author": "user1",
    "createdDate": "2023-01-03T09:24:50.633Z",
}

EXPECTED_SPACE = {
    "_id": 4554779,
    "type": "Space",
    "title": "DEMO",
    "_timestamp": "2024-04-02T09:53:15.818621+00:00",
    "url": "http://127.0.0.1:9696/spaces/DM",
    "createdDate": "2023-01-03T09:24:50.633Z",
    "author": "user1",
    "key": "DM",
}

RESPONSE_ATTACHMENT = {
    "results": [
        {
            "id": "att3637249",
            "title": "demo.py",
            "type": "attachment",
            "version": {"when": "2023-01-03T09:24:50.633Z"},
            "history": {"createdDate": "2023-01-03T09:24:50.633Z"},
            "extensions": {"fileSize": 230},
            "_links": {
                "download": "/download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                "webui": "/pages/viewpageattachments.action?pageId=1113&preview=demo.py",
            },
        }
    ],
    "start": 0,
    "limit": 1,
    "size": 1,
    "_links": {},
}

EXPECTED_ATTACHMENT = {
    "_id": "att3637249",
    "type": "attachment",
    "_timestamp": "2023-01-03T09:24:50.633Z",
    "title": "demo.py",
    "size": 230,
    "space": "DEMO",
    "page": "ES-scrum",
    "url": f"{HOST_URL}/pages/viewpageattachments.action?pageId=1113&preview=demo.py",
    "createdDate": "2023-01-03T09:24:50.633Z",
}

RESPONSE_CONTENT = "# This is the dummy file"

RESPONSE_SPACE_KEYS = {
    "results": [
        {
            "id": 23456,
            "key": "DM",
        },
        {
            "id": 9876,
            "key": "ES",
        },
    ],
    "start": 0,
    "limit": 100,
    "size": 2,
    "_links": {},
}

EXPECTED_CONTENT = {
    "_id": "att3637249",
    "_timestamp": "2023-01-03T09:24:50.633Z",
    "_attachment": "IyBUaGlzIGlzIHRoZSBkdW1teSBmaWxl",
}

EXPECTED_CONTENT_EXTRACTED = {
    "_id": "att3637249",
    "_timestamp": "2023-01-03T09:24:50.633Z",
    "body": RESPONSE_CONTENT,
}

EXPECTED_BLOG = {
    "_id": 4779,
    "type": "blogpost",
    "_timestamp": "2023-01-24T04:07:19.672Z",
    "title": "demo-blog",
    "body": "This is a test blog",
    "space": "DEMO",
    "url": f"{HOST_URL}/spaces/~1234abc/blogposts/4779/demo-blog",
    "createdDate": "2023-01-03T09:24:50.633Z",
    "author": "user1",
}

EXPECTED_BLOG_ATTACHMENT = {
    "_id": "att3637249",
    "type": "attachment",
    "_timestamp": "2023-01-03T09:24:50.633Z",
    "title": "demo.py",
    "size": 230,
    "space": "DEMO",
    "blog": "demo-blog",
    "url": f"{HOST_URL}/pages/viewpageattachments.action?pageId=1113&preview=demo.py",
    "createdDate": "2023-01-03T09:24:50.633Z",
    "author": "user1",
}

RESPONSE_SEARCH_RESULT = {
    "results": [
        {
            "content": {
                "id": "983046",
                "type": "page",
                "space": {"name": "Software Development"},
                "body": {
                    "storage": {
                        "value": "Confluence Connector currently supports below objects for ingestion of data in ElasticSearch.\nBlogs\nAttachments\nPages\nSpaces",
                        "representation": "storage",
                        "embeddedContent": [],
                        "_expandable": {"content": "/rest/api/content/6455384"},
                    },
                },
                "history": {
                    "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                    "createdDate": "2023-01-03T09:24:50.633Z",
                    "createdBy": {"publicName": "user1", "username": "user1"},
                },
            },
            "title": "Product Details",
            "excerpt": "Confluence Connector currently supports below objects for ingestion of data in ElasticSearch.\nBlogs\nAttachments\nPages\nSpaces",
            "url": "/spaces/SD/pages/983046/Product+Details",
            "lastModified": "2022-12-19T13:06:18.000Z",
            "entityType": "content",
            "ancestors": [{"title": "page1"}],
        },
        {
            "content": {
                "id": "att4587521",
                "type": "attachment",
                "extensions": {
                    "mediaType": "application/pdf",
                    "fileSize": 1119256,
                },
                "space": {"name": "Software Development"},
                "body": {
                    "storage": {
                        "value": "",
                        "representation": "storage",
                        "embeddedContent": [],
                        "_expandable": {"content": "/rest/api/content"},
                    },
                },
                "container": {"type": "page", "title": "Product Details"},
                "_links": {"download": "/download/attachments/196717/Potential.pdf"},
                "history": {
                    "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                    "createdDate": "2023-01-03T09:24:50.633Z",
                    "createdBy": {"publicName": "user1", "username": "user1"},
                },
            },
            "title": "Potential.pdf",
            "excerpt": "Evaluation Overview",
            "url": "/pages/viewpageattachments.action?pageId=196717&preview=%2F196717%2F4587521%2FPotential.pdf",
            "lastModified": "2023-01-24T03:34:38.000Z",
            "entityType": "content",
            "ancestors": [],
        },
        {
            "space": {
                "id": "196612",
                "key": "SD",
                "type": "global",
                "history": {
                    "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                    "createdDate": "2023-01-03T09:24:50.633Z",
                    "createdBy": {"publicName": "user1", "username": "user1"},
                },
            },
            "title": "Software Development",
            "excerpt": "",
            "url": "/spaces/SD",
            "lastModified": "2022-12-13T09:49:01.000Z",
            "entityType": "space",
            "ancestors": [],
        },
    ]
}


EXPECTED_SEARCH_RESULT_FOR_FILTERING_CLOUD = [
    {
        "_id": "983046",
        "title": "Product Details",
        "_timestamp": "2022-12-19T13:06:18.000Z",
        "body": "Confluence Connector currently supports below objects for ingestion of data in ElasticSearch.\nBlogs\nAttachments\nPages\nSpaces",
        "type": "page",
        "space": "Software Development",
        "url": f"{HOST_URL}/spaces/SD/pages/983046/Product+Details",
        "createdDate": "2023-01-03T09:24:50.633Z",
        "author": "user1",
    },
    {
        "_id": "att4587521",
        "title": "Potential.pdf",
        "_timestamp": "2023-01-24T03:34:38.000Z",
        "type": "attachment",
        "url": f"{HOST_URL}/pages/viewpageattachments.action?pageId=196717&preview=%2F196717%2F4587521%2FPotential.pdf",
        "space": "Software Development",
        "size": 1119256,
        "page": "Product Details",
        "createdDate": "2023-01-03T09:24:50.633Z",
        "author": "user1",
    },
    {
        "_id": "196612",
        "title": "Software Development",
        "_timestamp": "2022-12-13T09:49:01.000Z",
        "body": None,
        "type": "space",
        "url": f"{HOST_URL}/spaces/SD",
        "createdDate": "2023-01-03T09:24:50.633Z",
        "author": "user1",
    },
]


EXPECTED_SEARCH_RESULT_FOR_FILTERING_DATA_CENTER = [
    {
        "_id": "983046",
        "title": "Product Details",
        "_timestamp": "2022-12-19T13:06:18.000Z",
        "body": "Confluence Connector currently supports below objects for ingestion of data in ElasticSearch.\nBlogs\nAttachments\nPages\nSpaces",
        "type": "page",
        "url": f"{HOST_URL}/spaces/SD/pages/983046/Product+Details",
        "space": "Software Development",
        "createdDate": "2023-01-03T09:24:50.633Z",
        "author": "user1",
    },
    {
        "_id": "att4587521",
        "title": "Potential.pdf",
        "_timestamp": "2023-01-24T03:34:38.000Z",
        "type": "attachment",
        "url": f"{HOST_URL}/pages/viewpageattachments.action?pageId=196717&preview=%2F196717%2F4587521%2FPotential.pdf",
        "space": "Software Development",
        "size": 1119256,
        "page": "Product Details",
        "createdDate": "2023-01-03T09:24:50.633Z",
        "author": "user1",
    },
    {
        "_id": "196612",
        "title": "Software Development",
        "_timestamp": "2022-12-13T09:49:01.000Z",
        "body": None,
        "type": "space",
        "url": f"{HOST_URL}/spaces/SD",
    },
]


SPACE_PERMISSION_RESPONSE = [
    {
        "id": 1,
        "subjects": {
            "group": {
                "results": [
                    {
                        "type": "group",
                        "name": "group1",
                        "id": "group_id_1",
                    }
                ],
                "size": 1,
            },
        },
        "operation": {"operation": "read", "targetType": "space"},
    },
]
PAGE_PERMISSION_RESPONSE = [
    {
        "id": 1,
        "subjects": {
            "group": {
                "results": [
                    {
                        "type": "group",
                        "name": "group2",
                        "id": "group_id_2",
                    }
                ],
                "size": 1,
            },
        },
        "operation": {"operation": "read", "targetType": "page"},
    },
]
BLOG_POST_PERMISSION_RESPONSE = [
    {
        "id": 1,
        "subjects": {
            "group": {
                "results": [
                    {
                        "type": "group",
                        "name": "group2",
                        "id": "group_id_2",
                    }
                ],
                "size": 1,
            },
        },
        "operation": {"operation": "read", "targetType": "blogpost"},
    },
]
PAGE_RESTRICTION_RESPONSE = {
    "user": {
        "results": [
            {
                "type": "known",
                "accountId": "user_id_4",
                "accountType": "atlassian",
                "displayName": "user_4",
            },
            {
                "type": "known",
                "accountId": "user_id_5",
                "accountType": "atlassian",
                "displayName": "user_5",
            },
        ],
        "size": 2,
    },
    "group": {"results": [], "size": 0},
}

EXPECTED_QUERY_RESPONSE = {
    "content": {
        "id": "983041",
        "type": "page",
    },
    "title": "page 3",
    "excerpt": "page 3 excerpt",
    "url": "/spaces/space1/pages/983041/page+3",
    "entityType": "content",
    "lastModified": "2024-04-24T08:43:17.000Z",
}


@asynccontextmanager
async def create_confluence_source(
    use_text_extraction_service=False, data_source=CONFLUENCE_SERVER
):
    async with create_source(
        ConfluenceDataSource,
        data_source=data_source,
        username="admin",
        password="changeme",
        confluence_url=HOST_URL,
        spaces="*",
        ssl_enabled=False,
        use_document_level_security=False,
        use_text_extraction_service=use_text_extraction_service,
    ) as source:
        yield source


class JSONAsyncMock(AsyncMock):
    def __init__(self, json, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._json = json

    async def json(self):
        return self._json


class StreamReaderAsyncMock(AsyncMock):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.content = StreamReader


@pytest.mark.asyncio
async def test_validate_configuration_with_invalid_concurrent_downloads():
    """Test validate configuration method of BaseDataSource class with invalid concurrent downloads"""

    # Setup
    async with create_confluence_source() as source:
        source.configuration.get_field("concurrent_downloads").value = 100000
        # Execute
        with pytest.raises(ConfigurableFieldValueError):
            await source.validate_config()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "configs",
    [
        (
            # Confluence Server with blank dependent fields
            {
                "data_source": CONFLUENCE_SERVER,
                "username": "",
                "password": "",
                "account_email": "foo@bar.me",
                "api_token": "foo",
            }
        ),
        (
            # Confluence Data Center with blank dependent fields
            {
                "data_source": CONFLUENCE_DATA_CENTER,
                "data_center_username": "",
                "data_center_password": "",
                "account_email": "foo@bar.me",
                "api_token": "foo",
            }
        ),
        (
            # Confluence Cloud with blank dependent fields
            {
                "data_source": CONFLUENCE_CLOUD,
                "username": "foo",
                "password": "bar",
                "account_email": "",
                "api_token": "",
            }
        ),
    ],
)
async def test_validate_configuration_with_invalid_dependency_fields_raises_error(
    configs,
):
    # Setup
    async with create_confluence_source() as source:
        for k, v in configs.items():
            source.configuration.get_field(k).value = v
        # Execute
        with pytest.raises(ConfigurableFieldValueError):
            await source.validate_config()


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "configs",
    [
        (
            # Confluence Server with blank non-dependent fields
            {
                "data_source": CONFLUENCE_SERVER,
                "username": "foo",
                "password": "bar",
                "account_email": "",
                "api_token": "",
            }
        ),
        (
            # Confluence Data Center with blank dependent fields
            {
                "data_source": CONFLUENCE_DATA_CENTER,
                "data_center_username": "foo",
                "data_center_password": "bar",
                "account_email": "",
                "api_token": "",
            }
        ),
        (
            # Confluence Cloud with blank non-dependent fields
            {
                "data_source": CONFLUENCE_CLOUD,
                "username": "",
                "password": "",
                "account_email": "foo@bar.me",
                "api_token": "foobar",
            }
        ),
        (
            # SSL certificate not enabled (empty ssl_ca okay)
            {
                "username": "foo",
                "password": "bar",
                "ssl_enabled": False,
                "ssl_ca": "",
            }
        ),
    ],
)
async def test_validate_config_with_valid_dependency_fields_does_not_raise_error(
    configs,
):
    async with create_confluence_source() as source:
        source.confluence_client.ping = AsyncMock()
        for k, v in configs.items():
            source.configuration.get_field(k).value = v

        await source.validate_config()


@pytest.mark.asyncio
async def test_validate_config_when_ssl_enabled_and_ssl_ca_not_empty_does_not_raise_error():
    with patch.object(ssl, "create_default_context", return_value=MockSSL()):
        async with create_confluence_source() as source:
            source.confluence_client._get_session().get = AsyncMock()
            source.configuration.get_field("username").value = "foo"
            source.configuration.get_field("password").value = "foo"
            source.configuration.get_field("ssl_enabled").value = True
            source.configuration.get_field(
                "ssl_ca"
            ).value = (
                "-----BEGIN CERTIFICATE----- Certificate -----END CERTIFICATE-----"
            )
            await source.validate_config()


@pytest.mark.asyncio
async def test_tweak_bulk_options():
    """Test tweak_bulk_options method of BaseDataSource class"""

    # Setup
    async with create_confluence_source() as source:
        source.concurrent_downloads = 10
        options = {"concurrent_downloads": 5}

        # Execute
        source.tweak_bulk_options(options)

        assert options["concurrent_downloads"] == 10


@pytest.mark.asyncio
async def test_close_with_client_session():
    """Test close method for closing the existing session"""

    # Setup
    async with create_confluence_source() as source:
        source.confluence_client._get_session()

        # Execute
        await source.close()

        assert source.confluence_client.session is None


@pytest.mark.asyncio
async def test_close_without_client_session():
    """Test close method when the session does not exist"""
    # Setup
    async with create_confluence_source() as source:
        # Execute
        await source.close()

        assert source.confluence_client.session is None


@pytest.mark.asyncio
async def test_remote_validation_when_space_keys_are_valid():
    async with create_confluence_source() as source:
        source.spaces = ["DM", "ES"]
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_SPACE_KEYS)
        )

        await source._remote_validation()


@pytest.mark.asyncio
async def test_remote_validation_when_space_keys_are_unavailable_then_raise_exception():
    async with create_confluence_source() as source:
        source.spaces = ["ES", "CS"]
        async_response = AsyncMock()
        async_response.json.return_value = RESPONSE_SPACE_KEYS

        with mock.patch(
            "connectors.sources.atlassian.confluence.ConfluenceClient.api_call",
            return_value=async_response,
        ):
            with pytest.raises(
                ConfigurableFieldValueError,
                match="Spaces 'CS' are not available. Available spaces are: 'DM, ES'",
            ):
                await source._remote_validation()


class MockSSL:
    """This class contains methods which returns dummy ssl context"""

    def load_verify_locations(self, cadata):
        """This method verify locations"""
        pass


@pytest.mark.parametrize(
    "field, data_source",
    [
        ("confluence_url", "confluence_cloud"),
        ("account_email", "confluence_cloud"),
        ("api_token", "confluence_cloud"),
        ("username", "confluence_server"),
        ("password", "confluence_server"),
        ("data_center_username", "confluence_data_center"),
        ("data_center_password", "confluence_data_center"),
    ],
)
@pytest.mark.asyncio
async def test_validate_configuration_for_empty_fields(field, data_source):
    async with create_confluence_source() as source:
        source.confluence_client.configuration.get_field(
            "data_source"
        ).value = data_source
        source.confluence_client.configuration.get_field(field).value = ""

        # Execute
        with pytest.raises(Exception):
            await source.validate_config()


@pytest.mark.asyncio
async def test_ping_with_ssl():
    """Test ping method of ConfluenceDataSource class with SSL"""

    # Execute
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock()
        source.confluence_client.ssl_enabled = True
        source.confluence_client.certificate = (
            "-----BEGIN CERTIFICATE----- Certificate -----END CERTIFICATE-----"
        )

        # Execute
        with patch.object(ssl, "create_default_context", return_value=MockSSL()):
            source.confluence_client.ssl_ctx = ssl_context(
                certificate=source.confluence_client.certificate
            )
            await source.ping()


@pytest.mark.asyncio
@patch("aiohttp.ClientSession.get")
async def test_ping_for_failed_connection_exception(mock_get):
    """Tests the ping functionality when connection can not be established to Confluence."""

    # Setup
    async with create_confluence_source() as source:
        # Execute
        with patch.object(
            ConfluenceClient, "api_call", side_effect=Exception("Something went wrong")
        ):
            with pytest.raises(Exception):
                await source.ping()


@pytest.mark.asyncio
async def test_validate_configuration_for_ssl_enabled():
    """This function tests _validate_configuration when certification is empty and ssl is enabled"""
    # Setup
    async with create_confluence_source() as source:
        source.ssl_enabled = True

        # Execute
        with pytest.raises(Exception):
            source._validate_configuration()


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_429_status():
    initial_response = ClientResponseError(None, None)
    initial_response.status = 429
    initial_response.message = "rate-limited"
    initial_response.headers = {"Retry-After": 0.1}

    payload = {"value": "Test rate limit"}
    retried_response = AsyncMock()
    retried_response.json.return_value = payload
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            side_effect=[initial_response, retried_response]
        )
        response = await source.confluence_client.api_call(
            url="http://localhost:1000/sample"
        )
        result = await response.json()

    assert result == payload


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_429_status_without_retry_after_header():
    payload = {"value": "Test rate limit"}
    async_mock_response = AsyncMock()
    async_mock_response.json.return_value = payload

    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=async_mock_response
        )
        response = await source.confluence_client.api_call(
            url="http://localhost:1000/sample"
        )
        result = await response.json()

    assert result == payload


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_400_status():
    error = ClientResponseError(None, None)
    error.status = 400

    async with create_confluence_source() as source:
        with patch(
            "aiohttp.ClientSession.get",
            side_effect=error,
        ):
            with pytest.raises(BadRequest):
                response = await source.confluence_client.api_call(
                    url="http://localhost:1000/err"
                )
                await response.json()


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_401_status():
    error = ClientResponseError(None, None)
    error.status = 401

    async with create_confluence_source() as source:
        with patch(
            "aiohttp.ClientSession.get",
            side_effect=error,
        ):
            with pytest.raises(Unauthorized):
                response = await source.confluence_client.api_call(
                    url="http://localhost:1000/err"
                )
                await response.json()


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_403_status():
    error = ClientResponseError(None, None)
    error.status = 403

    async with create_confluence_source() as source:
        with patch(
            "aiohttp.ClientSession.get",
            side_effect=error,
        ):
            with pytest.raises(Forbidden):
                response = await source.confluence_client.api_call(
                    url="http://localhost:1000/err"
                )
                await response.json()


@pytest.mark.asyncio
async def test_get_with_404_status():
    error = ClientResponseError(None, None)
    error.status = 404

    async with create_confluence_source() as source:
        with patch(
            "aiohttp.ClientSession.get",
            side_effect=error,
        ):
            with pytest.raises(NotFound):
                response = await source.confluence_client.api_call(
                    url="http://localhost:1000/err"
                )
                await response.json()


@pytest.mark.asyncio
@patch("connectors.utils.time_to_sleep_between_retries", Mock(return_value=0))
async def test_get_with_500_status():
    error = ClientResponseError(None, None)
    error.status = 500

    async with create_confluence_source() as source:
        with patch(
            "aiohttp.ClientSession.get",
            side_effect=error,
        ):
            with pytest.raises(InternalServerError):
                response = await source.confluence_client.api_call(
                    url="http://localhost:1000/err"
                )
                await response.json()


@freeze_time("2023-01-24T04:07:19")
@pytest.mark.asyncio
async def test_fetch_spaces():
    # Setup
    async with create_confluence_source() as source:
        async_response = AsyncMock()
        async_response.json.return_value = RESPONSE_SPACE

        source.confluence_client.api_call = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_SPACE)
        )
        async for response in source.confluence_client.fetch_spaces():
            assert response["id"] == EXPECTED_SPACE["_id"]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "data_source_type, expected_api_query",
    [
        (CONFLUENCE_CLOUD, SPACE_QUERY_CLOUD),
        (CONFLUENCE_DATA_CENTER, SPACE_QUERY_DATA_CENTER),
        (CONFLUENCE_SERVER, SPACE_QUERY_DATA_CENTER),
    ],
)
async def test_fetch_spaces_uses_correct_query_for_data_source_type(
    data_source_type, expected_api_query
):
    """fetch_spaces uses the DC/Server-safe query off Cloud (CONFSERVER-99908)."""
    async with create_confluence_source(data_source=data_source_type) as source:
        source.confluence_client.paginated_api_call = MagicMock(
            return_value=AsyncIterator([RESPONSE_SPACE])
        )

        async for _ in source.confluence_client.fetch_spaces():
            pass

        source.confluence_client.paginated_api_call.assert_called_once_with(
            url_name="space",
            api_query=expected_api_query,
        )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "data_source_type, expected_api_query",
    [
        (CONFLUENCE_CLOUD, SPACE_QUERY_CLOUD),
        (CONFLUENCE_DATA_CENTER, SPACE_QUERY_DATA_CENTER),
        (CONFLUENCE_SERVER, SPACE_QUERY_DATA_CENTER),
    ],
)
async def test_remote_validation_uses_correct_query_for_data_source_type(
    data_source_type, expected_api_query
):
    """_remote_validation uses the DC/Server-safe query off Cloud (CONFSERVER-99908)."""
    async with create_confluence_source(data_source=data_source_type) as source:
        source.spaces = ["DM"]
        source.confluence_client.ping = AsyncMock()
        source.confluence_client.paginated_api_call = MagicMock(
            return_value=AsyncIterator([RESPONSE_SPACE_KEYS])
        )

        await source._remote_validation()

        source.confluence_client.paginated_api_call.assert_called_once_with(
            url_name="space",
            api_query=expected_api_query,
        )


@pytest.mark.asyncio
async def test_fetch_documents():
    # Setup
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_PAGE)
        )
        source.confluence_client.index_labels = True
        source.confluence_client.data_source_type = "confluence_cloud"
        # Execute
        async for response, _, _, _, _ in source.fetch_documents(api_query=""):
            assert response == EXPECTED_PAGE


@pytest.mark.asyncio
async def test_fetch_attachments():
    # Setup
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_ATTACHMENT)
        )

        # Execute
        async for response, _ in source.fetch_attachments(
            content_id=1113,
            parent_name="ES-scrum",
            parent_space="DEMO",
            parent_type="page",
        ):
            assert response == EXPECTED_ATTACHMENT


@pytest.mark.asyncio
async def test_search_by_query():
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_SEARCH_RESULT)
        )
        source.confluence_client.data_source_type = "confluence_cloud"
        documents = []
        async for response, _ in source.search_by_query(
            query="type in ('space', 'page', 'attachment') AND space.key ='SD'"
        ):
            documents.append(response)

        if len(documents) != 0:
            assert documents == EXPECTED_SEARCH_RESULT_FOR_FILTERING_CLOUD


@pytest.mark.asyncio
async def test_search_by_query_for_datacenter():
    async with create_confluence_source() as source:
        source.confluence_client.data_source_type = "confluence_data_center"
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_SEARCH_RESULT)
        )
        documents = []
        async for response, _ in source.search_by_query(
            query="type in ('space', 'page', 'attachment') AND space.key ='SD'"
        ):
            documents.append(response)

        if len(documents) != 0:
            assert documents == EXPECTED_SEARCH_RESULT_FOR_FILTERING_DATA_CENTER


@pytest.mark.asyncio
async def test_download_attachment():
    # Setup
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(StreamReaderAsyncMock())
        )

        # Execute
        with mock.patch(
            "aiohttp.StreamReader.iter_chunked",
            return_value=AsyncIterator([bytes(RESPONSE_CONTENT, "utf-8")]),
        ):
            response = await source.download_attachment(
                url="download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                attachment=EXPECTED_ATTACHMENT,
                doit=True,
            )
            if response is not None:
                assert response == EXPECTED_CONTENT


@pytest.mark.asyncio
async def test_download_attachment_with_upper_extension():
    # Setup
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(StreamReaderAsyncMock())
        )

        # Execute
        with mock.patch(
            "aiohttp.StreamReader.iter_chunked",
            return_value=AsyncIterator([bytes(RESPONSE_CONTENT, "utf-8")]),
        ):
            attachment = copy(EXPECTED_ATTACHMENT)
            attachment["title"] = "batch.TXT"

            response = await source.download_attachment(
                url="download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                attachment=EXPECTED_ATTACHMENT,
                doit=True,
            )
            if response is not None:
                assert response == EXPECTED_CONTENT


@pytest.mark.asyncio
async def test_download_attachment_when_filesize_is_large_then_download_skips():
    """Tests the download attachments method for file size greater than max limit."""
    # Setup
    async with create_confluence_source() as source:
        async_response = AsyncMock()
        async_response.__aenter__ = AsyncMock(return_value=StreamReaderAsyncMock())

        # Execute

        with mock.patch("aiohttp.ClientSession.get", return_value=async_response):
            with mock.patch(
                "aiohttp.StreamReader.iter_chunked",
                return_value=AsyncIterator([bytes(RESPONSE_CONTENT, "utf-8")]),
            ):
                attachment = copy(EXPECTED_ATTACHMENT)
                attachment["size"] = 23000000

                response = await source.download_attachment(
                    url="download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                    attachment=attachment,
                    doit=True,
                )
                assert response is None


@pytest.mark.asyncio
async def test_download_attachment_when_unsupported_filetype_used_then_fail_download_skips():
    """Tests the download attachments method for file type is not supported"""
    # Setup
    async with create_confluence_source() as source:
        async_response = AsyncMock()
        async_response.__aenter__ = AsyncMock(return_value=StreamReaderAsyncMock())

        with mock.patch("aiohttp.ClientSession.get", return_value=async_response):
            with mock.patch(
                "aiohttp.StreamReader.iter_chunked",
                return_value=AsyncIterator([bytes(RESPONSE_CONTENT, "utf-8")]),
            ):
                attachment = copy(EXPECTED_ATTACHMENT)

                attachment["title"] = "batch.mysy"

                response = await source.download_attachment(
                    url="download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                    attachment=attachment,
                    doit=True,
                )
                assert response is None


@pytest.mark.asyncio
@patch(
    "connectors_sdk.content_extraction.ContentExtraction._check_configured",
    lambda *_: True,
)
async def test_download_attachment_with_text_extraction_enabled_adds_body():
    with (
        patch(
            "connectors_sdk.content_extraction.ContentExtraction.extract_text",
            return_value=RESPONSE_CONTENT,
        ),
        patch(
            "connectors_sdk.content_extraction.ContentExtraction.get_extraction_config",
            return_value={"host": "http://localhost:8090"},
        ),
    ):
        async with create_confluence_source(use_text_extraction_service=True) as source:
            source.confluence_client._get_session().get = AsyncMock(
                return_value=JSONAsyncMock(StreamReaderAsyncMock())
            )

            # Execute
            with mock.patch(
                "aiohttp.StreamReader.iter_chunked",
                return_value=AsyncIterator([bytes(RESPONSE_CONTENT, "utf-8")]),
            ):
                response = await source.download_attachment(
                    url="download/attachments/1113/demo.py?version=1&modificationDate=1672737890633&cacheVersion=1&api=v2",
                    attachment=EXPECTED_ATTACHMENT,
                    doit=True,
                )
                if response is not None:
                    assert response == EXPECTED_CONTENT_EXTRACTED


@pytest.mark.asyncio
@mock.patch.object(
    ConfluenceClient,
    "fetch_spaces",
    return_value=AsyncIterator([copy(SPACE)]),
)
@mock.patch.object(
    ConfluenceDataSource,
    "fetch_documents",
    side_effect=[
        (AsyncIterator([[copy(EXPECTED_PAGE), 1, "space_key", [], {}]])),
        (AsyncIterator([[copy(EXPECTED_BLOG), 1, "space_key", [], {}]])),
    ],
)
@mock.patch.object(
    ConfluenceDataSource,
    "fetch_attachments",
    side_effect=[
        (AsyncIterator([[copy(EXPECTED_ATTACHMENT), "download-url"]])),
        (AsyncIterator([[copy(EXPECTED_BLOG_ATTACHMENT), "download-url"]])),
    ],
)
@mock.patch.object(
    ConfluenceDataSource,
    "download_attachment",
    return_value=AsyncIterator([[copy(EXPECTED_CONTENT)]]),
)
@freeze_time("2024-04-02T09:53:15.818621+00:00")
async def test_get_docs(spaces_patch, pages_patch, attachment_patch, content_patch):
    """Tests the get_docs method"""

    # Setup
    async with create_confluence_source() as source:
        expected_responses = [
            EXPECTED_SPACE,
            EXPECTED_PAGE,
            EXPECTED_BLOG,
            EXPECTED_ATTACHMENT,
            EXPECTED_BLOG_ATTACHMENT,
        ]

        # Execute
        documents = []
        source.confluence_client.data_source_type = "confluence_cloud"
        async for item, _ in source.get_docs():
            documents.append(item)
        assert documents == expected_responses


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "data_source_type, expected_content_query",
    [
        (CONFLUENCE_CLOUD, CONTENT_QUERY_CLOUD),
        (CONFLUENCE_DATA_CENTER, CONTENT_QUERY_DATA_CENTER),
        (CONFLUENCE_SERVER, CONTENT_QUERY_DATA_CENTER),
    ],
)
@freeze_time("2024-04-02T09:53:15.818621+00:00")
async def test_get_docs_uses_correct_content_query_for_data_source_type(
    data_source_type, expected_content_query
):
    """get_docs drops space.permissions from the content query off Cloud (CONFSERVER-99908)."""
    async with create_confluence_source(data_source=data_source_type) as source:
        source.confluence_client.fetch_spaces = MagicMock(
            return_value=AsyncIterator([copy(SPACE)])
        )
        source.fetch_server_space_permission = AsyncMock(return_value={})
        with mock.patch.object(
            ConfluenceDataSource,
            "fetch_documents",
            side_effect=[AsyncIterator([]), AsyncIterator([])],
        ) as fetch_documents_mock:
            async for _ in source.get_docs():
                pass

        content_queries = [call.args[0] for call in fetch_documents_mock.call_args_list]
        assert len(content_queries) == 2
        for query in content_queries:
            assert query.endswith(expected_content_query)
        if data_source_type == CONFLUENCE_CLOUD:
            assert all("space.permissions" in query for query in content_queries)
        else:
            assert all("space.permissions" not in query for query in content_queries)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "data_source_type", [CONFLUENCE_CLOUD, CONFLUENCE_DATA_CENTER, CONFLUENCE_SERVER]
)
async def test_get_session(data_source_type):
    async with create_confluence_source(data_source=data_source_type) as source:
        try:
            source.confluence_client._get_session()
        except Exception as e:
            pytest.fail(
                f"Should not raise for valid data source type '{data_source_type}'. Exception: {e}"
            )


@pytest.mark.asyncio
async def test_get_session_multiple_calls_return_same_instance():
    async with create_confluence_source() as source:
        first_instance = source.confluence_client._get_session()
        second_instance = source.confluence_client._get_session()
        assert first_instance is second_instance


@pytest.mark.asyncio
async def test_get_session_raise_on_invalid_data_source_type():
    async with create_confluence_source(data_source="invalid") as source:
        with pytest.raises(InvalidConfluenceDataSourceTypeError):
            source.confluence_client._get_session()


@pytest.mark.asyncio
async def test_get_access_control_dls_disabled():
    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=False)

        acl = []
        async for access_control in source.get_access_control():
            acl.append(access_control)

        assert len(acl) == 0


@pytest.mark.asyncio
@freeze_time("2023-01-24T04:07:19")
async def test_get_access_control_dls_enabled():
    mock_users = [
        {
            # Indexable: The user is active and atlassian user.
            "self": "url1",
            "accountId": "607194d6bc3c3f006f4c35d6",
            "accountType": "atlassian",
            "displayName": "user1",
            "locale": "en-US",
            "emailAddress": "user1@dummy-domain.com",
            "active": True,
        },
        {
            # Non-Indexable: The user is no longer active.
            "self": "url2",
            "accountId": "607194d6bc3c3f006f4c35d7",
            "accountType": "atlassian",
            "displayName": "user2",
            "locale": "en-US",
            "emailAddress": "user2@dummy-domain.com",
            "active": False,
        },
        {
            # Non-Indexable: User account type is app; it must be atlassian.
            "self": "url2",
            "accountId": "607194d6bc3c3f006f4c35d7",
            "accountType": "app",
            "displayName": "user2",
            "active": False,
        },
        {
            # Non-Indexable: Personal information about user is missing.
            "accountId": "607194d6bc3c3f006f4c35d7",
            "accountType": "app",
            "displayName": "user2",
            "active": False,
        },
    ]

    mock_user1 = {
        "self": "url1",
        "accountId": "607194d6bc3c3f006f4c35d6",
        "accountType": "atlassian",
        "displayName": "user1",
        "locale": "en-US",
        "emailAddress": "user1@dummy-domain.com",
        "active": True,
        "groups": {
            "size": 1,
            "items": [
                {
                    "name": "group1",
                    "groupId": "607194d6bc3c3f006f4c35d8",
                }
            ],
        },
        "applicationRoles": {
            "size": 0,
            "items": [
                {
                    "name": "role1",
                    "key": "607194d6bc3c3f006f4c35d9",
                }
            ],
        },
    }

    expected_user_doc = {
        "_id": "607194d6bc3c3f006f4c35d6",
        "identity": {
            "account_id": "account_id:607194d6bc3c3f006f4c35d6",
            "display_name": "name:user1",
            "locale": "locale:en-US",
            "email_address": "email_address:user1@dummy-domain.com",
        },
        "created_at": "2023-01-24T04:07:19+00:00",
        "query": {
            "template": {
                "params": {
                    "access_control": [
                        "account_id:607194d6bc3c3f006f4c35d6",
                        "group_id:607194d6bc3c3f006f4c35d8",
                        "role_key:607194d6bc3c3f006f4c35d9",
                    ]
                },
                "source": DLS_QUERY,
            },
        },
    }

    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)
        source.confluence_client.data_source_type = "confluence_cloud"

        source.atlassian_access_control.fetch_all_users_for_confluence = AsyncIterator(
            [mock_users]
        )
        source.atlassian_access_control.fetch_user_for_confluence = AsyncIterator(
            [mock_user1]
        )

        user_documents = []
        async for user_doc in source.get_access_control():
            user_documents.append(user_doc)
        assert expected_user_doc in user_documents


@pytest.mark.asyncio
@freeze_time("2023-01-24T04:07:19")
async def test_get_access_control_dls_enabled_for_datacenter():
    mock_users = [
        {
            "username": "user1",
            "displayName": "user1",
            "userKey": "607194d6bc3c3f006f4c35d6",
        },
    ]

    expected_user_doc = {
        "_id": "607194d6bc3c3f006f4c35d6",
        "identity": {
            "account_id": "account_id:607194d6bc3c3f006f4c35d6",
            "display_name": "name:user1",
            "username": "name:user1",
        },
        "created_at": "2023-01-24T04:07:19+00:00",
        "query": {
            "template": {
                "params": {
                    "access_control": [
                        "account_id:607194d6bc3c3f006f4c35d6",
                        "name:user1",
                        "name:user1",
                    ]
                },
                "source": DLS_QUERY,
            },
        },
    }

    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)
        source.confluence_client.data_source_type = "confluence_data_center"

        source.confluence_client.fetch_confluence_server_users = AsyncIterator(
            [mock_users]
        )

        user_documents = []
        async for user_doc in source.get_access_control():
            user_documents.append(user_doc)

        assert expected_user_doc in user_documents


@pytest.mark.asyncio
@freeze_time("2023-01-24T04:07:19")
async def test_get_access_control_dls_enabled_for_server():
    mock_users = [
        {
            "fullName": "user1",
            "name": "user1",
            "key": "607194d6bc3c3f006f4c35d6",
            "email": "nedog73667@rentaen.com",
        },
    ]

    expected_user_doc = {
        "_id": "607194d6bc3c3f006f4c35d6",
        "identity": {
            "account_id": "account_id:607194d6bc3c3f006f4c35d6",
            "display_name": "name:user1",
            "email_address": "email_address:nedog73667@rentaen.com",
        },
        "created_at": "2023-01-24T04:07:19+00:00",
        "query": {
            "template": {
                "params": {
                    "access_control": [
                        "account_id:607194d6bc3c3f006f4c35d6",
                        "email_address:nedog73667@rentaen.com",
                        "name:user1",
                    ]
                },
                "source": DLS_QUERY,
            },
        },
    }

    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)

        source.confluence_client.fetch_confluence_server_users = AsyncIterator(
            [mock_users]
        )

        user_documents = []
        async for user_doc in source.get_access_control():
            user_documents.append(user_doc)

        assert expected_user_doc in user_documents


@pytest.mark.asyncio
async def test_fetch_confluence_server_users():
    async with create_confluence_source() as source:
        async_response = AsyncMock()
        async_response.json.return_value = {"start": 0, "users": []}
        with mock.patch(
            "connectors.sources.atlassian.confluence.ConfluenceClient.api_call",
            return_value=async_response,
        ):
            async for user in source.confluence_client.fetch_confluence_server_users():
                assert user is None


@pytest.mark.asyncio
@mock.patch.object(
    ConfluenceClient,
    "fetch_spaces",
    return_value=AsyncIterator(([copy(SPACE)])),
)
@mock.patch.object(
    ConfluenceDataSource,
    "fetch_documents",
    side_effect=[
        (
            AsyncIterator(
                [
                    [
                        copy(EXPECTED_BLOG),
                        1,
                        "space_key",
                        BLOG_POST_PERMISSION_RESPONSE,
                        {},
                    ]
                ]
            )
        ),
        (
            AsyncIterator(
                [
                    [
                        copy(EXPECTED_PAGE),
                        1,
                        "space_key",
                        PAGE_PERMISSION_RESPONSE,
                        PAGE_RESTRICTION_RESPONSE,
                    ]
                ]
            )
        ),
    ],
)
@mock.patch.object(
    ConfluenceDataSource,
    "fetch_attachments",
    side_effect=[
        (
            AsyncIterator(
                [
                    [
                        copy(EXPECTED_BLOG_ATTACHMENT),
                        "download-url",
                    ]
                ]
            )
        ),
        (
            AsyncIterator(
                [
                    [
                        copy(EXPECTED_ATTACHMENT),
                        "download-url",
                    ]
                ]
            )
        ),
    ],
)
@mock.patch.object(
    ConfluenceDataSource,
    "download_attachment",
    return_value=AsyncIterator([[copy(EXPECTED_CONTENT)]]),
)
@freeze_time("2024-04-02T09:53:15.818621+00:00")
async def test_get_docs_dls_enabled(
    spaces_patch, pages_patch, attachment_patch, content_patch
):
    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)
        source.confluence_client.data_source_type = "confluence_cloud"

        expected_responses = [
            copy(EXPECTED_SPACE) | {"_allow_access_control": ["group_id:group_id_1"]},
            copy(EXPECTED_BLOG) | {"_allow_access_control": ["group_id:group_id_2"]},
            copy(EXPECTED_PAGE)
            | {
                "_allow_access_control": [
                    "account_id:user_id_4",
                    "account_id:user_id_5",
                ]
            },
            copy(EXPECTED_ATTACHMENT)
            | {
                "_allow_access_control": [
                    "account_id:user_id_4",
                    "account_id:user_id_5",
                ]
            },
            copy(EXPECTED_BLOG_ATTACHMENT)
            | {"_allow_access_control": ["group_id:group_id_2"]},
        ]

        async for item, _ in source.get_docs():
            item.get("_allow_access_control", []).sort()
            assert item in expected_responses


@pytest.mark.parametrize(
    "filtering, expected_docs",
    [
        (
            Filter({ADVANCED_SNIPPET: {"value": [{"query": "space = DEMO"}]}}),
            [
                EXPECTED_SPACE,
                EXPECTED_ATTACHMENT,
            ],
        ),
    ],
)
@pytest.mark.asyncio
async def test_get_docs_with_advanced_rules(filtering, expected_docs):
    async with create_confluence_source() as source:
        with patch.object(
            ConfluenceDataSource,
            "search_by_query",
            side_effect=[
                (AsyncIterator([[copy(EXPECTED_ATTACHMENT), "download-url"]])),
                (AsyncIterator([[copy(EXPECTED_SPACE), None]])),
            ],
        ):
            with patch.object(
                ConfluenceDataSource,
                "download_attachment",
                return_value=AsyncIterator([[copy(EXPECTED_CONTENT)]]),
            ):
                source.get_content = mock.Mock(return_value=EXPECTED_ATTACHMENT)

                async for item, _ in source.get_docs(filtering):
                    assert item in expected_docs


@pytest.mark.asyncio
async def test_extract_identities_for_datacenter():
    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)
        response = {
            "user": {
                "results": [
                    {
                        "type": "known",
                        "username": "demo",
                        "userKey": "8ab284ca8d53fe15018d5989eb750001",
                        "displayName": "Demo",
                    }
                ],
            },
            "group": {
                "results": [
                    {
                        "type": "group",
                        "name": "confluence-users",
                    }
                ],
            },
        }
        expected_response = source._extract_identities_for_datacenter(response=response)
        assert expected_response == {"group:confluence-users", "user:demo"}


@pytest.mark.asyncio
async def test_fetch_server_space_permission():
    async with create_confluence_source() as source:
        source._dls_enabled = MagicMock(return_value=True)
        payload = {
            "permissions": {
                "VIEWSPACE": {
                    "groups": ["confluence-users"],
                    "users": [
                        "admin",
                    ],
                }
            },
        }
        async_response = AsyncMock()
        async_response.json.return_value = payload
        with mock.patch(
            "connectors.sources.atlassian.confluence.ConfluenceClient.api_call",
            return_value=async_response,
        ):
            expected_response = await source.fetch_server_space_permission(
                space_key="key"
            )
            if expected_response is not None:
                assert expected_response == {
                    "permissions": {
                        "VIEWSPACE": {
                            "groups": ["confluence-users"],
                            "users": ["admin"],
                        }
                    }
                }


@pytest.mark.asyncio
async def test_api_call_for_exception(patch_sleep):
    """This function test _api_call when credentials are incorrect"""
    async with create_confluence_source() as source:
        source.confluence_client.retry_count = 1
        with patch.object(
            aiohttp, "ClientSession", side_effect=Exception(EXCEPTION_MESSAGE)
        ):
            with pytest.raises(Exception):
                await anext(source.confluence_client.api_call(url="abc"))


@pytest.mark.asyncio
async def test_get_permission():
    async with create_confluence_source() as source:
        actual_permission = {"users": ["admin"], "groups": ["group"]}
        permisssions = source.get_permission(permission=actual_permission)
        assert permisssions == {"group:group", "user:admin"}


@mock.patch.object(
    ConfluenceDataSource,
    "fetch_documents",
    side_effect=[
        (AsyncIterator([[copy(EXPECTED_PAGE), 1, "space_key", [], {}]])),
        (AsyncIterator([[copy(EXPECTED_BLOG), 1, "space_key", [], {}]])),
    ],
)
@pytest.mark.asyncio
async def test_page_blog_coro(fetch_documents):
    async with create_confluence_source() as source:
        await source._page_blog_coro("api_query", "target")


@pytest.mark.asyncio
async def test_end_signal_is_added_to_queue_in_case_of_exception():
    END_SIGNAL = "FINISHED_TASK"
    async with create_confluence_source() as source:
        with patch.object(
            source,
            "fetch_attachments",
            side_effect=Exception("Error fetching attachments"),
        ):
            with pytest.raises(Exception):
                await source._attachment_coro(document=EXPECTED_PAGE, access_control=[])
                assert source.queue.get_nowait() == END_SIGNAL


@pytest.mark.asyncio
async def test_fetch_page_blog_documents_with_labels():
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_PAGE)
        )
        with patch.object(
            ConfluenceClient, "fetch_label", return_value=["label1", "label2"]
        ):
            source.confluence_client.index_labels = True
            async for response, _ in source.confluence_client.fetch_page_blog_documents(
                api_query="type in ('blogpost', 'page')"
            ):
                assert response == {
                    "id": 4779,
                    "title": "ES-scrum",
                    "type": "page",
                    "history": {
                        "lastUpdated": {"when": "2023-01-24T04:07:19.672Z"},
                        "createdDate": "2023-01-03T09:24:50.633Z",
                        "createdBy": {"publicName": "user1"},
                    },
                    "children": {"attachment": {"size": 2}},
                    "body": {"storage": {"value": "This is a test page"}},
                    "space": {"name": "DEMO"},
                    "_links": {
                        "webui": "/spaces/~1234abc/pages/4779/ES-scrum",
                    },
                    "ancestors": [{"title": "parent_title"}],
                    "labels": ["label1", "label2"],
                }


@pytest.mark.asyncio
async def test_fetch_documents_with_html():
    async with create_confluence_source() as source:
        source.confluence_client._get_session().get = AsyncMock(
            return_value=JSONAsyncMock(RESPONSE_PAGE_WITH_HTML)
        )
        source.confluence_client.index_labels = True
        async for response, _, _, _, _ in source.fetch_documents(api_query=""):
            assert response == EXPECTED_PAGE
