"""Test The generic (IP Camera) config flow."""

from __future__ import annotations

from collections.abc import Generator
import contextlib
from copy import deepcopy
import errno
from http import HTTPStatus
import os.path
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, PropertyMock, _patch, patch

import httpx
import pytest
import respx

from homeassistant.components.camera import async_get_image
from homeassistant.components.generic.config_flow import slug
from homeassistant.components.generic.const import (
    CONF_CONFIRMED_OK,
    CONF_CONTENT_TYPE,
    CONF_FRAMERATE,
    CONF_LIMIT_REFETCH_TO_URL_CHANGE,
    CONF_STILL_IMAGE_URL,
    CONF_STREAM_SOURCE,
    DOMAIN,
    SECTION_ADVANCED,
)
from homeassistant.components.stream import (
    CONF_RTSP_TRANSPORT,
    CONF_USE_WALLCLOCK_AS_TIMESTAMPS,
)
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import (
    CONF_AUTHENTICATION,
    CONF_PASSWORD,
    CONF_USERNAME,
    CONF_VERIFY_SSL,
    HTTP_BASIC_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er

from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator, WebSocketGenerator

TESTDATA = {
    CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
    CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
    CONF_USERNAME: "fred_flintstone",
    CONF_PASSWORD: "bambam",
    SECTION_ADVANCED: {
        CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
        CONF_FRAMERATE: 5,
        CONF_VERIFY_SSL: False,
    },
}

TESTDATA_ONLYSTILL = TESTDATA.copy()
TESTDATA_ONLYSTILL.pop(CONF_STREAM_SOURCE)

TESTDATA_ONLYSTREAM = TESTDATA.copy()
TESTDATA_ONLYSTREAM.pop(CONF_STILL_IMAGE_URL)


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form(
    hass: HomeAssistant,
    fakeimgbytes_png: bytes,
    hass_client: ClientSessionGenerator,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
    mock_setup_entry: _patch[MagicMock],
    hass_ws_client: WebSocketGenerator,
) -> None:
    """Test the form with a normal set of settings."""

    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"

    # HA should now be serving a WS connection for a preview stream.
    ws_client = await hass_ws_client()
    flow_id = user_flow["flow_id"]
    await ws_client.send_json_auto_id(
        {
            "type": "generic_camera/start_preview",
            "flow_id": flow_id,
        },
    )
    json = await ws_client.receive_json()

    # Check stream_url is absolute (required by HLS player for child playlist URLs)
    stream_preview_url = json["event"]["attributes"]["stream_url"]
    assert stream_preview_url.startswith("http")

    client = await hass_client()
    still_preview_url = json["event"]["attributes"]["still_url"]
    # Check the preview image works.
    resp = await client.get(still_preview_url)
    assert resp.status == HTTPStatus.OK
    assert await resp.read() == fakeimgbytes_png

    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    assert result2["type"] is FlowResultType.CREATE_ENTRY
    assert result2["title"] == "127_0_0_1"
    assert result2["options"] == {
        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
        CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
        CONF_USERNAME: "fred_flintstone",
        CONF_PASSWORD: "bambam",
        CONF_CONTENT_TYPE: "image/png",
        SECTION_ADVANCED: {
            CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
            CONF_FRAMERATE: 5.0,
            CONF_VERIFY_SSL: False,
        },
    }

    # Check that the preview image is disabled after.
    resp = await client.get(still_preview_url)
    assert resp.status == HTTPStatus.NOT_FOUND


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_only_stillimage(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test we complete ok if the user wants still images only."""
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA_ONLYSTILL,
    )
    await hass.async_block_till_done()
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    assert result2["type"] is FlowResultType.CREATE_ENTRY
    assert result2["title"] == "127_0_0_1"
    assert result2["options"] == {
        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
        CONF_USERNAME: "fred_flintstone",
        CONF_PASSWORD: "bambam",
        CONF_CONTENT_TYPE: "image/png",
        SECTION_ADVANCED: {
            CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
            CONF_FRAMERATE: 5.0,
            CONF_VERIFY_SSL: False,
        },
    }

    assert respx.calls.call_count == 1


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_reject_preview(
    hass: HomeAssistant,
    mock_create_stream: _patch[MagicMock],
    user_flow: ConfigFlowResult,
) -> None:
    """Test we go back to the config screen if the user rejects the preview."""
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: False},
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["step_id"] == "user"


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_still_preview_cam_off(
    hass: HomeAssistant,
    mock_create_stream: _patch[MagicMock],
    user_flow: ConfigFlowResult,
    hass_client: ClientSessionGenerator,
    hass_ws_client: WebSocketGenerator,
) -> None:
    """Test camera errors are triggered during preview."""
    with (
        patch(
            "homeassistant.components.generic.camera.GenericCamera.is_on",
            new_callable=PropertyMock(return_value=False),
        ),
    ):
        result1 = await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA,
        )
        assert result1["type"] is FlowResultType.FORM
        assert result1["step_id"] == "user_confirm"

        # HA should now be serving a WS connection for a preview stream.
        ws_client = await hass_ws_client()
        flow_id = user_flow["flow_id"]
        await ws_client.send_json_auto_id(
            {
                "type": "generic_camera/start_preview",
                "flow_id": flow_id,
            },
        )
        json = await ws_client.receive_json()

        client = await hass_client()
        still_preview_url = json["event"]["attributes"]["still_url"]
        # Try to view the image, should be unavailable.
        client = await hass_client()
        resp = await client.get(still_preview_url)
    assert resp.status == HTTPStatus.SERVICE_UNAVAILABLE


@respx.mock
@pytest.mark.usefixtures("fakeimg_gif")
async def test_form_only_stillimage_gif(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test we complete ok if the user wants a gif."""
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA_ONLYSTILL,
    )
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    await hass.async_block_till_done()
    assert result2["type"] is FlowResultType.CREATE_ENTRY
    assert result2["options"][CONF_CONTENT_TYPE] == "image/gif"


@respx.mock
async def test_form_only_svg_whitespace(
    hass: HomeAssistant,
    fakeimgbytes_svg: bytes,
    user_flow: ConfigFlowResult,
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test we complete ok if svg starts with whitespace, issue #68889."""
    fakeimgbytes_wspace_svg = bytes("  \n ", encoding="utf-8") + fakeimgbytes_svg
    respx.get("http://127.0.0.1/testurl/1").respond(stream=fakeimgbytes_wspace_svg)
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA_ONLYSTILL,
    )
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    await hass.async_block_till_done()
    assert result2["type"] is FlowResultType.CREATE_ENTRY


@respx.mock
@pytest.mark.parametrize(
    "image_file",
    [
        ("sample1_animate.png"),
        ("sample2_jpeg_odd_header.jpg"),
        ("sample3_jpeg_odd_header.jpg"),
        ("sample4_K5-60mileAnim-320x240.gif"),
        ("sample5_webp.webp"),
    ],
)
async def test_form_only_still_sample(
    hass: HomeAssistant, user_flow: ConfigFlowResult, image_file, mock_setup_entry
) -> None:
    """Test various sample images #69037."""
    image_path = os.path.join(os.path.dirname(__file__), image_file)
    image_bytes = await hass.async_add_executor_job(Path(image_path).read_bytes)
    respx.get("http://127.0.0.1/testurl/1").respond(stream=image_bytes)
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA_ONLYSTILL,
    )
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    await hass.async_block_till_done()
    assert result2["type"] is FlowResultType.CREATE_ENTRY


@respx.mock
@pytest.mark.parametrize(
    ("template", "url", "expected_result", "expected_errors"),
    [
        # Test we can handle templates in strange parts of the url, #70961.
        (
            "http://localhost:812{{3}}/static/icons/favicon-apple-180x180.png",
            "http://localhost:8123/static/icons/favicon-apple-180x180.png",
            "user_confirm",
            None,
        ),
        (
            "{% if 1 %}https://bla{% else %}https://yo{% endif %}",
            "https://bla/",
            "user_confirm",
            None,
        ),
        (
            "http://{{example.org",
            "http://example.org",
            "user",
            {"still_image_url": "template_error"},
        ),
        (
            "invalid1://invalid:4\\1",
            "invalid1://invalid:4%5c1",
            "user",
            {"still_image_url": "malformed_url"},
        ),
        (
            "relative/urls/are/not/allowed.jpg",
            "relative/urls/are/not/allowed.jpg",
            "user",
            {"still_image_url": "relative_url"},
        ),
    ],
)
async def test_form_still_template(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    fakeimgbytes_png: bytes,
    mock_setup_entry: Generator[AsyncMock],
    template,
    url,
    expected_result,
    expected_errors,
) -> None:
    """Test we can handle various templates."""
    with contextlib.suppress(httpx.InvalidURL):
        # There is no need to mock the request if its an
        # invalid url because we will never make the request
        respx.get(url).respond(stream=fakeimgbytes_png)
    data = TESTDATA_ONLYSTILL.copy()
    data[CONF_STILL_IMAGE_URL] = template
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        data,
    )
    await hass.async_block_till_done()
    assert result2["step_id"] == expected_result
    assert result2.get("errors") == expected_errors


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_rtsp_mode(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test we complete ok if the user enters a stream url."""
    data = deepcopy(TESTDATA)
    data[SECTION_ADVANCED][CONF_RTSP_TRANSPORT] = "tcp"
    data[CONF_STREAM_SOURCE] = "rtsp://127.0.0.1/testurl/2"
    result1 = await hass.config_entries.flow.async_configure(user_flow["flow_id"], data)
    assert result1["type"] is FlowResultType.FORM
    assert result1["step_id"] == "user_confirm"
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    assert result2["type"] is FlowResultType.CREATE_ENTRY
    assert result2["title"] == "127_0_0_1"
    assert result2["options"] == {
        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
        CONF_STREAM_SOURCE: "rtsp://127.0.0.1/testurl/2",
        CONF_USERNAME: "fred_flintstone",
        CONF_PASSWORD: "bambam",
        CONF_CONTENT_TYPE: "image/png",
        SECTION_ADVANCED: {
            CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
            CONF_FRAMERATE: 5.0,
            CONF_VERIFY_SSL: False,
            CONF_RTSP_TRANSPORT: "tcp",
        },
    }


async def test_form_only_stream(
    hass: HomeAssistant,
    fakeimgbytes_jpg: bytes,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we complete ok if the user wants stream only."""
    data = TESTDATA_ONLYSTREAM.copy()
    data[CONF_STREAM_SOURCE] = "rtsp://user:pass@127.0.0.1/testurl/2"
    result1 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        data,
    )

    assert result1["type"] is FlowResultType.FORM
    result2 = await hass.config_entries.flow.async_configure(
        result1["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )

    assert result2["title"] == "127_0_0_1"
    assert result2["options"] == {
        CONF_STREAM_SOURCE: "rtsp://user:pass@127.0.0.1/testurl/2",
        CONF_USERNAME: "fred_flintstone",
        CONF_PASSWORD: "bambam",
        CONF_CONTENT_TYPE: "image/jpeg",
        SECTION_ADVANCED: {
            CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
            CONF_FRAMERATE: 5.0,
            CONF_VERIFY_SSL: False,
        },
    }

    with patch(
        "homeassistant.components.camera._async_get_stream_image",
        return_value=fakeimgbytes_jpg,
    ):
        image_obj = await async_get_image(hass, "camera.127_0_0_1")
        assert image_obj.content == fakeimgbytes_jpg


async def test_form_still_and_stream_not_provided(
    hass: HomeAssistant, user_flow: ConfigFlowResult
) -> None:
    """Test we show a suitable error if neither still or stream URL are provided."""
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        {
            SECTION_ADVANCED: {
                CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
                CONF_FRAMERATE: 5,
                CONF_VERIFY_SSL: False,
            },
        },
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"base": "no_still_image_or_stream_url"}


@respx.mock
@pytest.mark.parametrize(
    ("side_effect", "expected_message"),
    [
        (httpx.TimeoutException, {"still_image_url": "unable_still_load"}),
        (
            httpx.HTTPStatusError("", request=None, response=httpx.Response(401)),
            {"still_image_url": "unable_still_load_auth"},
        ),
        (
            httpx.HTTPStatusError("", request=None, response=httpx.Response(403)),
            {"still_image_url": "unable_still_load_auth"},
        ),
        (
            httpx.HTTPStatusError("", request=None, response=httpx.Response(404)),
            {"still_image_url": "unable_still_load_not_found"},
        ),
        (
            httpx.HTTPStatusError("", request=None, response=httpx.Response(500)),
            {"still_image_url": "unable_still_load_server_error"},
        ),
        (
            httpx.HTTPStatusError("", request=None, response=httpx.Response(503)),
            {"still_image_url": "unable_still_load_server_error"},
        ),
        (  # Errors without specific handler should show the general message.
            httpx.HTTPStatusError("", request=None, response=httpx.Response(507)),
            {"still_image_url": "unable_still_load"},
        ),
    ],
)
async def test_form_image_http_exceptions(
    side_effect,
    expected_message,
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we handle image http exceptions."""
    respx.get("http://127.0.0.1/testurl/1").side_effect = [side_effect]
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == expected_message


@respx.mock
async def test_form_image_http_302(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
    fakeimgbytes_png: bytes,
) -> None:
    """Test we handle image http 302 (temporary redirect)."""
    respx.get("http://127.0.0.1/testurl/1").side_effect = [
        httpx.Response(
            status_code=302, headers={"Location": "http://127.0.0.1/testurl2/1"}
        )
    ]
    respx.get("http://127.0.0.1/testurl2/1", name="fake_img2").respond(
        stream=fakeimgbytes_png
    )
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["step_id"] == "user_confirm"


@respx.mock
async def test_form_stream_invalidimage(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we handle invalid image when a stream is specified."""
    respx.get("http://127.0.0.1/testurl/1").respond(stream=b"invalid")
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"still_image_url": "invalid_still_image"}


@respx.mock
async def test_form_stream_invalidimage2(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we handle invalid image when a stream is specified."""
    respx.get("http://127.0.0.1/testurl/1").respond(content=None)
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"still_image_url": "unable_still_load_no_image"}


@respx.mock
async def test_form_stream_invalidimage3(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we handle invalid image when a stream is specified."""
    respx.get("http://127.0.0.1/testurl/1").respond(content=bytes([0xFF]))
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )

    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"still_image_url": "invalid_still_image"}


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_stream_timeout(
    hass: HomeAssistant,
    user_flow: ConfigFlowResult,
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test we handle invalid auth."""
    mock_create_stream.return_value.start = AsyncMock()
    mock_create_stream.return_value.stop = AsyncMock()
    mock_create_stream.return_value.hass = hass
    mock_create_stream.return_value.add_provider.return_value.part_recv = AsyncMock()
    mock_create_stream.return_value.add_provider.return_value.part_recv.return_value = (
        False
    )
    result2 = await hass.config_entries.flow.async_configure(
        user_flow["flow_id"],
        TESTDATA,
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"stream_source": "timeout"}


@respx.mock
async def test_form_stream_not_set_up(hass: HomeAssistant, user_flow) -> None:
    """Test we handle if stream has not been set up."""

    with patch(
        "homeassistant.components.generic.config_flow.create_stream",
        side_effect=HomeAssistantError("Stream integration is not set up."),
    ):
        result1 = await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA_ONLYSTREAM,
        )
    await hass.async_block_till_done()

    assert result1["type"] is FlowResultType.FORM
    assert result1["errors"] == {"stream_source": "stream_not_set_up"}


@respx.mock
async def test_form_stream_other_error(hass: HomeAssistant, user_flow) -> None:
    """Test the unknown error for streams."""

    with (
        patch(
            "homeassistant.components.generic.config_flow.create_stream",
            side_effect=HomeAssistantError("Some other error."),
        ),
        pytest.raises(HomeAssistantError),
    ):
        await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA_ONLYSTREAM,
        )
    await hass.async_block_till_done()


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_stream_permission_error(
    hass: HomeAssistant, user_flow: ConfigFlowResult
) -> None:
    """Test we handle permission error."""
    with patch(
        "homeassistant.components.generic.config_flow.create_stream",
        side_effect=PermissionError(),
    ):
        result2 = await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA,
        )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"stream_source": "stream_not_permitted"}


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_no_route_to_host(
    hass: HomeAssistant, user_flow: ConfigFlowResult
) -> None:
    """Test we handle no route to host."""
    with patch(
        "homeassistant.components.generic.config_flow.create_stream",
        side_effect=OSError(errno.EHOSTUNREACH, "No route to host"),
    ):
        result2 = await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA,
        )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"stream_source": "stream_no_route_to_host"}


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_stream_io_error(
    hass: HomeAssistant, user_flow: ConfigFlowResult
) -> None:
    """Test we handle an io error when setting up stream."""
    with patch(
        "homeassistant.components.generic.config_flow.create_stream",
        side_effect=OSError(errno.EIO, "Input/output error"),
    ):
        result2 = await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA,
        )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"stream_source": "stream_io_error"}


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_form_oserror(hass: HomeAssistant, user_flow: ConfigFlowResult) -> None:
    """Test we handle OS error when setting up stream."""
    with (
        patch(
            "homeassistant.components.generic.config_flow.create_stream",
            side_effect=OSError("Some other OSError"),
        ),
        pytest.raises(OSError),
    ):
        await hass.config_entries.flow.async_configure(
            user_flow["flow_id"],
            TESTDATA,
        )


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_options_template_error(
    hass: HomeAssistant,
    mock_create_stream: _patch[MagicMock],
    config_entry: MockConfigEntry,
) -> None:
    """Test the options flow with a template error."""
    result = await hass.config_entries.options.async_init(config_entry.entry_id)
    assert result["type"] is FlowResultType.FORM
    assert result["step_id"] == "init"

    # try updating the still image url
    data = TESTDATA.copy()
    data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/2"
    result2 = await hass.config_entries.options.async_configure(
        result["flow_id"],
        user_input=data,
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["step_id"] == "user_confirm"

    result2a = await hass.config_entries.options.async_configure(
        result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
    )
    assert result2a["type"] is FlowResultType.CREATE_ENTRY

    result3 = await hass.config_entries.options.async_init(config_entry.entry_id)
    assert result3["type"] is FlowResultType.FORM
    assert result3["step_id"] == "init"

    # verify that an invalid template reports the correct UI error.
    data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/{{1/0}}"
    result4 = await hass.config_entries.options.async_configure(
        result3["flow_id"],
        user_input=data,
    )
    assert result4.get("type") is FlowResultType.FORM
    assert result4["errors"] == {"still_image_url": "template_error"}

    # verify that an invalid template reports the correct UI error.
    data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
    data[CONF_STREAM_SOURCE] = "http://127.0.0.2/testurl/{{1/0}}"
    result5 = await hass.config_entries.options.async_configure(
        result4["flow_id"],
        user_input=data,
    )

    assert result5.get("type") is FlowResultType.FORM
    assert result5["errors"] == {"stream_source": "template_error"}

    # verify that an relative stream url is rejected.
    data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
    data[CONF_STREAM_SOURCE] = "relative/stream.mjpeg"
    result6 = await hass.config_entries.options.async_configure(
        result5["flow_id"],
        user_input=data,
    )
    assert result6.get("type") is FlowResultType.FORM
    assert result6["errors"] == {"stream_source": "relative_url"}

    # verify that an malformed stream url is rejected.
    data[CONF_STILL_IMAGE_URL] = "http://127.0.0.1/testurl/1"
    data[CONF_STREAM_SOURCE] = "http://example.com:45:56"
    result7 = await hass.config_entries.options.async_configure(
        result6["flow_id"],
        user_input=data,
    )
    assert result7.get("type") is FlowResultType.FORM
    assert result7["errors"] == {"stream_source": "malformed_url"}


async def test_slug(hass: HomeAssistant, caplog: pytest.LogCaptureFixture) -> None:
    """Test that the slug function generates an error in case of invalid template.

    Other paths in the slug function are already tested by other tests.
    """
    result = slug(hass, "http://127.0.0.2/testurl/{{1/0}}")
    assert result is None
    assert "Syntax error in" in caplog.text

    result = slug(hass, "http://example.com:999999999999/stream")
    assert result is None
    assert "Syntax error in" in caplog.text


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_options_only_stream(
    hass: HomeAssistant,
    mock_setup_entry: _patch[MagicMock],
    mock_create_stream: _patch[MagicMock],
) -> None:
    """Test the options flow without a still_image_url."""

    mock_entry = MockConfigEntry(
        title="Test Camera",
        domain=DOMAIN,
        data={},
        options=TESTDATA_ONLYSTREAM,
    )
    mock_entry.add_to_hass(hass)
    await hass.config_entries.async_setup(mock_entry.entry_id)

    result = await hass.config_entries.options.async_init(mock_entry.entry_id)
    assert result["type"] is FlowResultType.FORM
    assert result["step_id"] == "init"

    # try updating the config options
    result2 = await hass.config_entries.options.async_configure(
        result["flow_id"],
        user_input=TESTDATA_ONLYSTREAM,
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["step_id"] == "user_confirm"

    result3 = await hass.config_entries.options.async_configure(
        result2["flow_id"], user_input={CONF_CONFIRMED_OK: True}
    )
    assert result3["type"] is FlowResultType.CREATE_ENTRY
    assert result3["data"][CONF_CONTENT_TYPE] == "image/jpeg"


async def test_options_still_and_stream_not_provided(
    hass: HomeAssistant,
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test we show a suitable error if neither still or stream URL are provided."""
    data = TESTDATA_ONLYSTILL.copy()
    data.pop(CONF_STILL_IMAGE_URL)

    mock_entry = MockConfigEntry(
        title="Test Camera",
        domain=DOMAIN,
        data={},
        options=data,
    )
    mock_entry.add_to_hass(hass)
    await hass.config_entries.async_setup(mock_entry.entry_id)

    result = await hass.config_entries.options.async_init(mock_entry.entry_id)
    assert result["type"] is FlowResultType.FORM
    assert result["step_id"] == "init"

    result2 = await hass.config_entries.options.async_configure(
        result["flow_id"],
        user_input=data,
    )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"base": "no_still_image_or_stream_url"}


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_options_permission_error(
    hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
    """Test we handle a PermissionError and pass the message through."""

    result = await hass.config_entries.options.async_init(config_entry.entry_id)
    with patch(
        "homeassistant.components.generic.config_flow.create_stream",
        side_effect=PermissionError("Some message"),
    ):
        result2 = await hass.config_entries.options.async_configure(
            result["flow_id"],
            TESTDATA,
        )
    assert result2["type"] is FlowResultType.FORM
    assert result2["errors"] == {"stream_source": "stream_not_permitted"}


async def test_migrate_existing_ids(
    hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
    """Test that existing ids are migrated for issue #70568."""
    test_data = {
        CONF_STILL_IMAGE_URL: "http://127.0.0.1/testurl/1",
        CONF_STREAM_SOURCE: "http://127.0.0.1/testurl/2",
        CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION,
        CONF_USERNAME: "fred_flintstone",
        CONF_PASSWORD: "bambam",
        CONF_FRAMERATE: 5,
        CONF_LIMIT_REFETCH_TO_URL_CHANGE: False,
        CONF_VERIFY_SSL: False,
    }

    test_data[CONF_CONTENT_TYPE] = "image/png"
    old_unique_id = "54321"
    entity_id = "camera.sample_camera"

    mock_entry = MockConfigEntry(
        domain=DOMAIN, unique_id=old_unique_id, options=test_data, title="My Title"
    )
    new_unique_id = mock_entry.entry_id
    mock_entry.add_to_hass(hass)

    entity_entry = entity_registry.async_get_or_create(
        "camera",
        DOMAIN,
        old_unique_id,
        suggested_object_id="sample camera",
        config_entry=mock_entry,
    )
    assert entity_entry.entity_id == entity_id
    assert entity_entry.unique_id == old_unique_id

    await hass.config_entries.async_setup(mock_entry.entry_id)
    await hass.async_block_till_done()

    entity_entry = entity_registry.async_get(entity_id)
    assert entity_entry.unique_id == new_unique_id


@respx.mock
@pytest.mark.usefixtures("fakeimg_png")
async def test_options_use_wallclock_as_timestamps(
    hass: HomeAssistant,
    mock_create_stream: _patch[MagicMock],
    hass_client: ClientSessionGenerator,
    hass_ws_client: WebSocketGenerator,
    fakeimgbytes_png: bytes,
    config_entry: MockConfigEntry,
    mock_setup_entry: _patch[MagicMock],
) -> None:
    """Test the use_wallclock_as_timestamps option flow."""

    result = await hass.config_entries.options.async_init(
        config_entry.entry_id, context={"show_advanced_options": True}
    )
    assert result["type"] is FlowResultType.FORM
    assert result["step_id"] == "init"

    data = deepcopy(TESTDATA)
    data[SECTION_ADVANCED][CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
    result2 = await hass.config_entries.options.async_configure(
        result["flow_id"],
        user_input=data,
    )
    assert result2["type"] is FlowResultType.FORM

    ws_client = await hass_ws_client()
    flow_id = result2["flow_id"]
    await ws_client.send_json_auto_id(
        {
            "type": "generic_camera/start_preview",
            "flow_id": flow_id,
            "flow_type": "options_flow",
        },
    )
    json = await ws_client.receive_json()

    client = await hass_client()
    still_preview_url = json["event"]["attributes"]["still_url"]
    # Check the preview image works.
    resp = await client.get(still_preview_url)
    assert resp.status == HTTPStatus.OK
    assert await resp.read() == fakeimgbytes_png

    # Test what happens if user rejects the preview
    result3 = await hass.config_entries.options.async_configure(
        result2["flow_id"], user_input={CONF_CONFIRMED_OK: False}
    )
    assert result3["type"] is FlowResultType.FORM
    assert result3["step_id"] == "init"
    result4 = await hass.config_entries.options.async_configure(
        result3["flow_id"],
        user_input=data,
    )
    assert result4["type"] is FlowResultType.FORM
    assert result4["step_id"] == "user_confirm"
    result5 = await hass.config_entries.options.async_configure(
        result4["flow_id"],
        user_input={CONF_CONFIRMED_OK: True},
    )
    assert result5["type"] is FlowResultType.CREATE_ENTRY