import asyncio
import json
import os
import re
import socket
import subprocess
import sys
import time
from http import HTTPStatus
from pathlib import Path

import psutil
import pytest
import requests
from conftest import cleanup_eos_eosdash, is_owned_pid
from loguru import logger
from pydantic import ValidationError

from akkudoktoreos.config.configabc import is_home_assistant_addon
from akkudoktoreos.core.version import __version__
from akkudoktoreos.server.server import (
    ServerCommonSettings,
    get_default_host,
    get_default_port,
    get_host_ip,
    wait_for_port_free,
)


@pytest.mark.skipif(os.name == "nt", reason="Directory symlinks require privileges on Windows")
def test_temp_directory_alias_is_contained(tmp_path, is_in_test_dir):
    """Resolved path containment accepts aliases but rejects sibling directories."""
    target = tmp_path / "server"
    target.mkdir()
    alias = tmp_path / "alias"
    alias.symlink_to(target, target_is_directory=True)

    assert is_in_test_dir(alias / "config.json", str(target))
    assert is_in_test_dir(target / "config.json", str(alias))
    assert not is_in_test_dir(tmp_path / "server-other" / "config.json", str(target))


class TestServer:
    def test_server_setup_for_class(self, server_setup_for_class, is_in_test_dir):
        """Ensure server is started."""
        server = server_setup_for_class["server"]
        eos_dir = server_setup_for_class["eos_dir"]

        # Assure server is running
        result = requests.get(f"{server}/v1/health", timeout=2)
        assert result.status_code == HTTPStatus.OK
        health = result.json()
        assert health["status"] == "alive"
        assert health["version"] == __version__

        result = requests.get(f"{server}/v1/config", timeout=2)
        assert result.status_code == HTTPStatus.OK

        # Get testing config
        config_json = result.json()
        config_folder_path = Path(config_json["general"]["config_folder_path"])
        config_file_path = Path(config_json["general"]["config_file_path"])
        data_folder_path = Path(config_json["general"]["data_folder_path"])
        data_ouput_path = Path(config_json["general"]["data_output_path"])
        # Assure we are working in test environment
        assert is_in_test_dir(config_folder_path, eos_dir)
        assert is_in_test_dir(config_file_path, eos_dir)
        assert is_in_test_dir(data_folder_path, eos_dir)
        assert is_in_test_dir(data_ouput_path, eos_dir)


class TestServerSettingsValidation:
    """Test the port restrictions in ServerCommonSettings when running as HA addon."""

    def test_ha_addon_default_ports_ok(self, config_eos, monkeypatch):
        """Default ports are accepted in HA addon mode."""
        monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
        assert config_eos.server.port == get_default_port()  # 8503
        assert config_eos.server.eosdash_port == get_default_port() + 1  # 8504

    def test_server_port_restriction_in_ha_addon(self, config_eos, monkeypatch):
        """Server port must be the default (8503) in HA addon mode."""
        monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
        with pytest.raises(ValidationError) as excinfo:
            config_eos.server.port = 9000
        assert "Server port number `8503` for Home Assistant add-on can not be changed" in str(
            excinfo.value
        )

    def test_eosdash_port_restriction_in_ha_addon(self, config_eos, monkeypatch):
        """EOSdash port must be the default (8504) in HA addon mode."""
        monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
        with pytest.raises(ValidationError) as excinfo:
            config_eos.server.eosdash_port = 9001
        assert "EOSdash port number `8504` for Home Assistant add-on can not be changed" in str(
            excinfo.value
        )

    def test_ports_allowed_when_not_ha_addon(self, config_eos):
        """Custom ports are allowed when not in HA addon mode."""
        # is_home_assistant_addon() returns False by default in this test environment
        config_eos.server.port = 9000
        config_eos.server.eosdash_port = 9001
        assert config_eos.server.port == 9000
        assert config_eos.server.eosdash_port == 9001


class TestServerStartStop:
    @pytest.mark.asyncio
    async def test_forward_stream_truncates_very_long_line(self, monkeypatch, tmp_path):
        """Test logging from EOSdash can also handle very long lines."""

        eos_dir = tmp_path
        monkeypatch.setenv("EOS_DIR", str(eos_dir))
        monkeypatch.setenv("EOS_CONFIG_DIR", str(eos_dir))

        # Import after env vars are set
        from akkudoktoreos.server.rest.starteosdash import (
            EOSDASH_LOG_MAX_LINE_BYTES,
            _eosdash_log_worker,
            eosdash_log_queue,
            forward_stream,
        )

        # ---- Ensure queue + worker are initialized ----
        if eosdash_log_queue is None:
            from akkudoktoreos.server.rest import starteosdash

            starteosdash.eosdash_log_queue = asyncio.Queue(maxsize=10)
            worker_task = asyncio.create_task(_eosdash_log_worker())
        else:
            worker_task = None

        long_message = "X" * (EOSDASH_LOG_MAX_LINE_BYTES + 10_000)
        raw_line = f"INFO some.module:123 some_func - {long_message}\n"
        raw_bytes = raw_line.encode()

        reader = asyncio.StreamReader()
        reader.feed_data(raw_bytes)
        reader.feed_eof()

        # ---- Capture Loguru output ----
        records = []

        def sink(message):
            records.append(message.record)

        logger_id = logger.add(sink, level="INFO")

        try:
            await forward_stream(reader)

            # Allow log worker to flush queue
            await asyncio.sleep(0)

        finally:
            logger.remove(logger_id)

            # Clean shutdown of worker (important for pytest)
            if worker_task:
                from akkudoktoreos.server.rest import starteosdash

                if starteosdash.eosdash_log_queue:
                    starteosdash.eosdash_log_queue.put_nowait(None)
                await worker_task

        # ---- Assert ----
        assert len(records) == 1, "Expected exactly one log record"

        record = records[0]
        msg = record["message"]

        assert msg.endswith("[TRUNCATED]"), "Expected truncation marker"
        assert len(msg) <= EOSDASH_LOG_MAX_LINE_BYTES + 20

    @pytest.mark.asyncio
    async def test_server_start_eosdash(self, config_eos, monkeypatch, tmp_path):
        """Test the EOSdash server startup from EOS.

        Do not use any fixture as this will make pytest the owner of the EOSdash port.

        Tests that:
        1. EOSdash starts via the supervisor
        2. The /eosdash/health endpoint returns OK
        3. EOSdash reports correct status and version
        4. EOSdash can be terminated cleanly
        """
        eos_dir = tmp_path
        monkeypatch.setenv("EOS_DIR", str(eos_dir))
        monkeypatch.setenv("EOS_CONFIG_DIR", str(eos_dir))

        # Import with environment vars set to prevent creation of EOS.config.json in wrong dir.
        from akkudoktoreos.server.rest import starteosdash

        config_eos.server.host = get_default_host()
        config_eos.server.port = 8503
        config_eos.server.eosdash_host = config_eos.server.host
        config_eos.server.eosdash_port = 8504
        timeout = 120

        eosdash_server = f"http://{config_eos.server.eosdash_host}:{config_eos.server.eosdash_port}"

        # Port may be blocked
        assert wait_for_port_free(
            config_eos.server.eosdash_port, timeout=120, waiting_app_name="EOSdash"
        )

        owned_processes: list[psutil.Process] = []
        try:
            await starteosdash.supervise_eosdash()
            process = starteosdash.eosdash_proc
            assert process is not None, "EOSdash supervisor did not start a process"
            owned_processes.append(psutil.Process(process.pid))

            startup = False
            error = ""
            for _ in range(int(timeout / 3)):
                try:
                    resp = requests.get(f"{eosdash_server}/eosdash/health", timeout=2)
                    if resp.status_code == HTTPStatus.OK:
                        startup = True
                        break
                    error = f"{resp.status_code}, {str(resp.content)}"
                except requests.RequestException as ex:
                    error = str(ex)
                await asyncio.sleep(3)

            assert startup, f"Connection to {eosdash_server}/eosdash/health failed: {error}"
            health = resp.json()
            assert health.get("status") == "alive"
            assert health.get("version") == __version__
            assert is_owned_pid(owned_processes[0], health.get("pid"))

            # Terminate the process started by this test, then reap it via asyncio.
            process.terminate()
            await asyncio.wait_for(process.wait(), timeout=timeout)
            try:
                resp = requests.get(f"{eosdash_server}/eosdash/health", timeout=2)
            except requests.RequestException:
                pass
            else:
                assert resp.status_code != HTTPStatus.OK
        finally:
            cleanup_eos_eosdash(
                host=config_eos.server.host,
                port=config_eos.server.port,
                eosdash_host=config_eos.server.eosdash_host,
                eosdash_port=config_eos.server.eosdash_port,
                server_timeout=timeout,
                owned_processes=owned_processes,
                config_dir=str(config_eos.general.config_folder_path),
            )
            if starteosdash.eosdash_proc is not None:
                await asyncio.wait_for(starteosdash.eosdash_proc.wait(), timeout=timeout)

    @pytest.mark.skipif(os.name == "nt", reason="Server restart not supported on Windows")
    def test_server_restart(self, server_setup_for_function, is_system_test, is_in_test_dir):
        """Test server restart."""
        server = server_setup_for_function["server"]
        eos_dir = server_setup_for_function["eos_dir"]
        timeout = server_setup_for_function["timeout"]

        result = requests.get(f"{server}/v1/config")
        assert result.status_code == HTTPStatus.OK

        # Get testing config
        config_json = result.json()
        config_folder_path = Path(config_json["general"]["config_folder_path"])
        config_file_path = Path(config_json["general"]["config_file_path"])
        data_folder_path = Path(config_json["general"]["data_folder_path"])
        data_ouput_path = Path(config_json["general"]["data_output_path"])
        cache_file_path = data_folder_path.joinpath(config_json["cache"]["subpath"]).joinpath(
            "cachefilestore.json"
        )
        # Assure we are working in test environment
        assert is_in_test_dir(config_folder_path, eos_dir)
        assert is_in_test_dir(config_file_path, eos_dir)
        assert is_in_test_dir(data_folder_path, eos_dir)
        assert is_in_test_dir(data_ouput_path, eos_dir)

        if is_system_test:
            # Prepare cache entry and get cached data
            result = requests.put(f"{server}/v1/config/weather/provider", json="BrightSky")
            assert result.status_code == HTTPStatus.OK

            result = requests.post(f"{server}/v1/prediction/update/BrightSky")
            assert result.status_code == HTTPStatus.OK

            result = requests.get(f"{server}/v1/prediction/series?key=weather_temp_air")
            assert result.status_code == HTTPStatus.OK
            data = result.json()
            assert data["data"] != {}

            result = requests.put(f"{server}/v1/config/file")
            assert result.status_code == HTTPStatus.OK

        # Save cache
        result = requests.post(f"{server}/v1/admin/cache/save")
        assert result.status_code == HTTPStatus.OK
        cache = result.json()

        assert cache_file_path.exists()

        result = requests.get(f"{server}/v1/admin/cache")
        assert result.status_code == HTTPStatus.OK
        cache = result.json()

        result = requests.get(f"{server}/v1/health")
        assert result.status_code == HTTPStatus.OK
        pid = result.json()["pid"]

        result = requests.post(f"{server}/v1/admin/server/restart")
        assert result.status_code == HTTPStatus.OK
        assert "Restarting EOS.." in result.json()["message"]
        new_pid = result.json()["pid"]

        # Wait for server to shut down
        for retries in range(10):
            try:
                result = requests.get(f"{server}/v1/health", timeout=2)
                if result.status_code == HTTPStatus.OK:
                    pid = result.json()["pid"]
                    if pid == new_pid:
                        # Already started
                        break
                else:
                    break
            except Exception as ex:
                break
            time.sleep(3)

        # Assure EOS is up again
        startup = False
        error = ""
        for retries in range(int(timeout / 5)):
            try:
                result = requests.get(f"{server}/v1/health", timeout=2)
                if result.status_code == HTTPStatus.OK:
                    startup = True
                    break
                error = f"{result.status_code}, {str(result.content)}"
            except Exception as ex:
                error = str(ex)
            time.sleep(5)

        assert startup, f"Connection to {server}/v1/health failed: {error}"
        assert result.json()["status"] == "alive"
        pid = result.json()["pid"]
        assert pid == new_pid

        result = requests.get(f"{server}/v1/admin/cache")
        assert result.status_code == HTTPStatus.OK
        new_cache = result.json()

        assert cache.items() <= new_cache.items()

        if is_system_test:
            result = requests.get(f"{server}/v1/config")
            assert result.status_code == HTTPStatus.OK
            assert result.json()["weather"]["provider"] == "BrightSky"

            # Wait for initialisation task to have finished
            time.sleep(5)

            result = requests.get(f"{server}/v1/prediction/series?key=weather_temp_air")
            assert result.status_code == HTTPStatus.OK
            assert result.json() == data

        # Shutdown the newly created server
        result = requests.post(f"{server}/v1/admin/server/shutdown")
        assert result.status_code == HTTPStatus.OK
        assert "Stopping EOS.." in result.json()["message"]
        new_pid = result.json()["pid"]


class TestServerWithEnv:
    eos_env = {
        "EOS_SERVER__EOSDASH_PORT": "8555",
    }

    def test_server_setup_for_class(self, server_setup_for_class):
        """Ensure server is started with environment passed to configuration."""
        server = server_setup_for_class["server"]

        assert server_setup_for_class["eosdash_port"] == int(
            self.eos_env["EOS_SERVER__EOSDASH_PORT"]
        )

        result = requests.get(f"{server}/v1/config")
        assert result.status_code == HTTPStatus.OK

        # Get testing config
        config_json = result.json()

        # Assure config got configuration from environment
        assert config_json["server"]["eosdash_port"] == int(
            self.eos_env["EOS_SERVER__EOSDASH_PORT"]
        )


class TestServerReload:
    """EOS with uvicorn's auto-reload (``server.reload``, used by ``make run-dev``)."""

    STARTUP_TIMEOUT_S = 90

    @staticmethod
    def free_port() -> int:
        """Return a free TCP port on the default host that EOS accepts (1024..49151).

        Starts above the ports of the other server tests (8503, 8504, 8555).
        """
        for port in range(8513, 8600):
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                try:
                    sock.bind((get_default_host(), port))
                except OSError:
                    continue
                return port
        pytest.fail("No free port for the EOS reload test in 8513..8599")

    @staticmethod
    def stop_process_tree(process: subprocess.Popen) -> None:
        """Stop a process and all of its descendants (reloader, worker)."""
        try:
            parent = psutil.Process(process.pid)
            processes = [*parent.children(recursive=True), parent]
        except psutil.NoSuchProcess:
            return
        for proc in processes:
            try:
                proc.terminate()
            except psutil.NoSuchProcess:
                pass
        _, alive = psutil.wait_procs(processes, timeout=15)
        for proc in alive:
            try:
                proc.kill()
            except psutil.NoSuchProcess:
                pass
        process.wait(timeout=15)

    def test_server_starts_with_reload(self, tmp_path: Path) -> None:
        """The reload worker initializes the singletons that run_eos() set up in the parent.

        The worker process imports the app afresh. Without initialization in the
        app's lifespan, the startup fails with 'access before init'.
        """
        eos_dir = tmp_path / "eos"
        eos_dir.mkdir()
        # An empty config file, so EOS does not fall back to the user's config file.
        (eos_dir / "EOS.config.json").write_text("{}", encoding="utf-8")
        host = get_default_host()
        port = self.free_port()
        env = os.environ.copy()
        env.update(
            {
                "EOS_DIR": str(eos_dir),
                "EOS_CONFIG_DIR": str(eos_dir),
                "PYTHONIOENCODING": "utf-8",
            }
        )
        log_path = tmp_path / "eos.log"
        with log_path.open("w", encoding="utf-8") as log_file:
            # cwd: the reloader watches the working directory, not the repository.
            process = subprocess.Popen(
                [
                    sys.executable,
                    "-m",
                    "akkudoktoreos.server.eos",
                    "--host",
                    host,
                    "--port",
                    str(port),
                    "--reload",
                    "true",
                    "--startup_eosdash",
                    "false",
                ],
                cwd=tmp_path,
                env=env,
                stdout=log_file,
                stderr=subprocess.STDOUT,
            )
            try:
                health: dict = {}
                deadline = time.monotonic() + self.STARTUP_TIMEOUT_S
                while time.monotonic() < deadline and process.poll() is None:
                    try:
                        response = requests.get(f"http://{host}:{port}/v1/health", timeout=2)
                        if response.status_code == HTTPStatus.OK:
                            health = response.json()
                            break
                    except requests.RequestException:
                        pass
                    time.sleep(1)
                worker_pids = {
                    child.pid for child in psutil.Process(process.pid).children(recursive=True)
                } if process.poll() is None else set()
            finally:
                self.stop_process_tree(process)

        log = log_path.read_text(encoding="utf-8", errors="replace")
        assert health.get("status") == "alive", f"EOS with --reload did not start:\n{log}"
        assert "before init" not in log, log
        # The app runs in the reload worker, a child of the started process.
        assert health.get("pid") in worker_pids


class TestEosdashRedirect:
    """Redirects to EOSdash must target an address the client can reach.

    See https://github.com/Akkudoktor-EOS/EOS/issues/1320: the redirect was built from
    the EOSdash bind address, so remote clients were sent to their own localhost.
    """

    @pytest.fixture
    def client(self, config_eos):
        from fastapi.testclient import TestClient

        from akkudoktoreos.server.eos import app

        config_eos.server.eosdash_host = "127.0.0.1"
        config_eos.server.eosdash_port = 8504
        return TestClient(app, follow_redirects=False)

    @pytest.mark.parametrize("host", ["localhost:8503", "127.0.0.1:8503"])
    def test_root_redirect_uses_request_host(self, client, host):
        """The root redirect points to the host the client used, not to the bind address."""
        response = client.get("/", headers={"Host": host})
        assert response.status_code == HTTPStatus.SEE_OTHER
        assert response.headers["location"] == f"http://{host.split(':')[0]}:8504/"

    def test_root_redirect_uses_host_ip_of_the_eos_machine(self, client):
        """Access by the IP address of the EOS machine redirects to that address."""
        host_ip = get_host_ip()
        response = client.get("/", headers={"Host": f"{host_ip}:8503"})
        assert response.status_code == HTTPStatus.SEE_OTHER
        assert response.headers["location"] == f"http://{host_ip}:8504/"

    def test_root_redirect_ignores_untrusted_forwarded_headers(self, client):
        """Raw forwarding headers cannot override the public dashboard address."""
        response = client.get(
            "/",
            headers={
                "Host": "localhost",
                "X-Forwarded-Host": "eos.example.com",
                "X-Forwarded-Proto": "https",
            },
        )
        assert response.status_code == HTTPStatus.SEE_OTHER
        assert response.headers["location"] == "http://localhost:8504/"

    def test_root_redirect_keeps_local_host(self, client):
        """Local access still redirects to the local EOSdash."""
        response = client.get("/", headers={"Host": "127.0.0.1:8503"})
        assert response.status_code == HTTPStatus.SEE_OTHER
        assert response.headers["location"] == "http://127.0.0.1:8504/"

    def test_untrusted_request_host_is_not_reflected(self, client):
        """An unknown Host header must not become the redirect target."""
        response = client.get("/", headers={"Host": "attacker.example"})
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "attacker.example:8504" not in response.text
        assert "eosdash_public_url" in response.text

    def test_untrusted_request_host_on_unknown_path(self, client):
        """The 404 page offers no link for an unknown Host header."""
        response = client.get("/no-such-page", headers={"Host": "attacker.example"})
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "attacker.example:8504" not in response.text

    def test_eosdash_path_redirect_keeps_path(self, client):
        """The path is preserved when redirecting to EOSdash."""
        response = client.get("/eosdash/health", headers={"Host": "localhost:8503"})
        assert response.status_code == HTTPStatus.SEE_OTHER
        assert response.headers["location"] == "http://localhost:8504/eosdash/health"

    @pytest.mark.parametrize(
        "path",
        [
            "/eosdash:payload",
            "/eosdash+custom:payload",
            "/eosdash:%09payload",
            "/eosdash:%0Apayload",
        ],
    )
    def test_scheme_bearing_path_is_rejected(self, client, path):
        """User paths must not replace the configured dashboard URL with a scheme."""
        response = client.get(path, headers={"Host": "localhost:8503"})
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "location" not in response.headers

    @pytest.mark.parametrize(
        "path",
        ["eosdash:payload", "./eosdash:payload", "eosdash:\tpayload", "eosdash:\npayload"],
    )
    def test_sanitizer_rejects_schemes_after_normalization(self, path):
        """Normalization must not turn an accepted path into a scheme-bearing URL."""
        from akkudoktoreos.server.eos import _sanitize_redirect_path

        assert _sanitize_redirect_path(path) is None

    def test_unknown_path_error_page_links_to_request_host(self, client):
        """The 404 page links to EOSdash on the host the client used."""
        response = client.get("/no-such-page", headers={"Host": "localhost:8503"})
        assert response.status_code == HTTPStatus.NOT_FOUND
        # The link must be real HTML, the error page escapes the message it is given.
        assert "&lt;a href" not in response.text
        # Compare the whole link target, a substring check would also accept a foreign host.
        hrefs = [href for href in re.findall(r'href="([^"]*)"', response.text) if href != "/docs"]
        assert hrefs == ["http://localhost:8504/"]

    def test_error_page_escapes_request_url(self, client):
        """A crafted URL is shown as text, never as markup."""
        response = client.get(
            "/%3Cscript%3Ealert(1)%3C/script%3E", headers={"Host": "localhost:8503"}
        )
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "<script>alert(1)</script>" not in response.text

    @pytest.mark.parametrize("host", ["[::1]", "[::1]:8503"])
    def test_direct_ipv6_preserves_address(self, client, host):
        """An IPv6 address keeps its brackets in the redirect."""
        response = client.get("/", headers={"Host": host})
        assert response.headers["location"] == "http://[::1]:8504/"

    def test_untrusted_ipv6_host_is_not_reflected(self, client):
        """An IPv6 address that the configuration does not know is not reflected."""
        response = client.get("/", headers={"Host": "[2001:db8::1234]:8503"})
        assert response.status_code == HTTPStatus.NOT_FOUND
        assert "2001:db8::1234" not in response.headers.get("location", "")

    @pytest.mark.parametrize(
        "public_url",
        [
            "https://energy.example.com",
            "https://energy.example.com:443",
            "https://energy.example.com:9443/dashboard",
            "https://[2001:db8::1234]/dashboard",
        ],
    )
    def test_public_url_preserves_proxy_port_and_prefix(self, client, config_eos, public_url):
        config_eos.server.eosdash_public_url = public_url + "/"
        headers = {"Host": "internal:8503", "X-Forwarded-Host": "wrong.example"}
        response = client.get("/", headers=headers)
        assert response.headers["location"] == public_url + "/"
        response = client.get("/eosdash/health", headers=headers)
        assert response.headers["location"] == public_url + "/eosdash/health"
        response = client.get("/missing", headers=headers)
        hrefs = [href for href in re.findall(r'href="([^"]*)"', response.text) if href != "/docs"]
        assert hrefs == [public_url + "/"]

    @pytest.mark.parametrize(
        "value",
        [
            "",
            "/dashboard",
            "//example.com",
            "ftp://example.com",
            "https://user:password@example.com",
            "https://example.com?token=a",
            "https://example.com#fragment",
            "https://example.com:99999",
            "https://example.com:0",
            "https://example.com\\evil",
            "https://example.com/\nheader",
            "https://example.com/a b",
        ],
    )
    def test_invalid_public_url_rejected(self, config_eos, value):
        with pytest.raises(ValueError):
            config_eos.server.eosdash_public_url = value