"""Tests for async util methods from Python source."""

import contextlib
import glob
import importlib
import os
from pathlib import Path, PurePosixPath
import time
from typing import Any
from unittest.mock import Mock, patch

import pytest

from homeassistant import block_async_io
from homeassistant.core import HomeAssistant

from .common import extract_stack_to_frame


@pytest.fixture(autouse=True)
def disable_block_async_io(disable_block_async_io):
    """Disable the loop protection from block_async_io after each test."""


async def test_protect_loop_debugger_sleep(caplog: pytest.LogCaptureFixture) -> None:
    """Test time.sleep injected by the debugger is not reported."""
    block_async_io.enable()
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/.venv/blah/pydevd.py",
                lineno="23",
                line="do_something()",
            ),
        ]
    )
    with (
        patch(
            "homeassistant.block_async_io.get_current_frame",
            return_value=frames,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        time.sleep(0)
    assert "Detected blocking call inside the event loop" not in caplog.text


async def test_protect_loop_sleep() -> None:
    """Test time.sleep not injected by the debugger raises."""
    block_async_io.enable()
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/no_dev.py",
                lineno="23",
                line="do_something()",
            ),
        ]
    )
    with (
        pytest.raises(RuntimeError, match="Caught blocking call to sleep with args"),
        patch(
            "homeassistant.block_async_io.get_current_frame",
            return_value=frames,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        time.sleep(0)


async def test_protect_loop_sleep_get_current_frame_raises() -> None:
    """Test time.sleep when get_current_frame raises ValueError."""
    block_async_io.enable()
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/no_dev.py",
                lineno="23",
                line="do_something()",
            ),
        ]
    )
    with (
        pytest.raises(RuntimeError, match="Caught blocking call to sleep with args"),
        patch(
            "homeassistant.block_async_io.get_current_frame",
            side_effect=ValueError,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        time.sleep(0)


async def test_protect_loop_importlib_import_module_non_integration(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test import_module in the loop for non-loaded module."""
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/no_dev.py",
                lineno="23",
                line="do_something()",
            ),
        ]
    )
    with (
        patch.object(block_async_io, "_IN_TESTS", False),
        patch(
            "homeassistant.block_async_io.get_current_frame",
            return_value=frames,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        block_async_io.enable()
        with pytest.raises(ImportError):
            importlib.import_module("not_loaded_module")

    assert "Detected blocking call to import_module" in caplog.text


async def test_protect_loop_importlib_import_loaded_module_non_integration(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test import_module in the loop for a loaded module."""
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/no_dev.py",
                lineno="23",
                line="do_something()",
            ),
        ]
    )
    with (
        patch.object(block_async_io, "_IN_TESTS", False),
        patch(
            "homeassistant.block_async_io.get_current_frame",
            return_value=frames,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        block_async_io.enable()
        importlib.import_module("sys")

    assert "Detected blocking call to import_module" not in caplog.text


async def test_protect_loop_importlib_import_module_in_integration(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test import_module in the loop for non-loaded module in an integration."""
    frames = extract_stack_to_frame(
        [
            Mock(
                filename="/home/paulus/homeassistant/core.py",
                lineno="23",
                line="do_something()",
            ),
            Mock(
                filename="/home/paulus/homeassistant/components/hue/light.py",
                lineno="23",
                line="self.light.is_on",
            ),
            Mock(
                filename="/home/paulus/aiohue/lights.py",
                lineno="2",
                line="something()",
            ),
        ]
    )
    with (
        patch.object(block_async_io, "_IN_TESTS", False),
        patch(
            "homeassistant.block_async_io.get_current_frame",
            return_value=frames,
        ),
        patch(
            "homeassistant.helpers.frame.get_current_frame",
            return_value=frames,
        ),
    ):
        block_async_io.enable()
        with pytest.raises(ImportError):
            importlib.import_module("not_loaded_module")

    assert (
        "Detected blocking call to import_module with args ('not_loaded_module',) "
        "inside the event loop by "
        "integration 'hue' at homeassistant/components/hue/light.py, line 23"
    ) in caplog.text


async def test_protect_loop_open(caplog: pytest.LogCaptureFixture) -> None:
    """Test open of a file in /proc is not reported."""
    block_async_io.enable()
    with (
        contextlib.suppress(FileNotFoundError),
        open("/proc/does_not_exist", encoding="utf8"),
    ):
        pass
    assert "Detected blocking call to open with args" not in caplog.text


async def test_protect_open(caplog: pytest.LogCaptureFixture) -> None:
    """Test opening a file in the event loop logs."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    with (
        contextlib.suppress(FileNotFoundError),
        open("/config/data_not_exist", encoding="utf8"),
    ):
        pass

    assert "Detected blocking call to open with args" in caplog.text


async def test_enable_multiple_times(caplog: pytest.LogCaptureFixture) -> None:
    """Test trying to enable multiple times."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()

    with pytest.raises(
        RuntimeError, match="Blocking call detection is already enabled"
    ):
        block_async_io.enable()


@pytest.mark.parametrize(
    "path",
    [
        "/config/data_not_exist",
        Path("/config/data_not_exist"),
        PurePosixPath("/config/data_not_exist"),
    ],
)
async def test_protect_open_path(path: Any, caplog: pytest.LogCaptureFixture) -> None:
    """Test opening a file by path in the event loop logs."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    with contextlib.suppress(FileNotFoundError), open(path, encoding="utf8"):
        pass

    assert "Detected blocking call to open with args" in caplog.text


async def test_protect_loop_glob(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test glob calls in the loop are logged."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    glob.glob("/dev/null")
    assert "Detected blocking call to glob with args" in caplog.text
    caplog.clear()
    await hass.async_add_executor_job(glob.glob, "/dev/null")
    assert "Detected blocking call to glob with args" not in caplog.text


async def test_protect_loop_iglob(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test iglob calls in the loop are logged."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    glob.iglob("/dev/null")
    assert "Detected blocking call to iglob with args" in caplog.text
    caplog.clear()
    await hass.async_add_executor_job(glob.iglob, "/dev/null")
    assert "Detected blocking call to iglob with args" not in caplog.text


async def test_protect_loop_scandir(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test glob calls in the loop are logged."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    with contextlib.suppress(FileNotFoundError):
        os.scandir("/path/that/does/not/exists")
    assert "Detected blocking call to scandir with args" in caplog.text
    caplog.clear()
    with contextlib.suppress(FileNotFoundError):
        await hass.async_add_executor_job(os.scandir, "/path/that/does/not/exists")
    assert "Detected blocking call to scandir with args" not in caplog.text


async def test_protect_loop_listdir(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test listdir calls in the loop are logged."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    with contextlib.suppress(FileNotFoundError):
        os.listdir("/path/that/does/not/exists")
    assert "Detected blocking call to listdir with args" in caplog.text
    caplog.clear()
    with contextlib.suppress(FileNotFoundError):
        await hass.async_add_executor_job(os.listdir, "/path/that/does/not/exists")
    assert "Detected blocking call to listdir with args" not in caplog.text


async def test_protect_loop_walk(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test os.walk calls in the loop are logged."""
    with patch.object(block_async_io, "_IN_TESTS", False):
        block_async_io.enable()
    with contextlib.suppress(FileNotFoundError):
        os.walk("/path/that/does/not/exists")
    assert "Detected blocking call to walk with args" in caplog.text
    caplog.clear()
    with contextlib.suppress(FileNotFoundError):
        await hass.async_add_executor_job(os.walk, "/path/that/does/not/exists")
    assert "Detected blocking call to walk with args" not in caplog.text


async def test_open_calls_ignored_in_tests(caplog: pytest.LogCaptureFixture) -> None:
    """Test opening a file in tests is ignored."""
    assert block_async_io._IN_TESTS
    block_async_io.enable()
    with (
        contextlib.suppress(FileNotFoundError),
        open("/config/data_not_exist", encoding="utf8"),
    ):
        pass

    assert "Detected blocking call to open with args" not in caplog.text