import json
import pytest
import os

from stat import S_IREAD
from backup.config import Config, Setting
from backup.ha import AddonStopper
from backup.exceptions import SupervisorFileSystemError
from .faketime import FakeTime
from dev.simulated_supervisor import SimulatedSupervisor, URL_MATCH_START_ADDON, URL_MATCH_STOP_ADDON, URL_MATCH_ADDON_INFO
from dev.request_interceptor import RequestInterceptor
from .helpers import skipForRoot


def getSaved(config: Config):
    with open(config.get(Setting.STOP_ADDON_STATE_PATH)) as f:
        data = json.load(f)
        return set(data["start"]), set(data["watchdog"])


def save(config: Config, to_start, to_watchdog_enable):
    with open(config.get(Setting.STOP_ADDON_STATE_PATH), "w") as f:
        json.dump({"start": list(to_start), "watchdog": list(to_watchdog_enable)}, f)

@pytest.mark.asyncio
async def test_no_stop_config(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug = "test_slug_1"
    supervisor.installAddon(slug, "Test decription")
    addon_stopper.allowRun()
    addon_stopper.isBackingUp(False)
    assert supervisor.addon(slug)["state"] == "started"
    await addon_stopper.stopAddons("ignore")
    assert supervisor.addon(slug)["state"] == "started"
    await addon_stopper.check()
    await addon_stopper.startAddons()
    assert supervisor.addon(slug)["state"] == "started"


@pytest.mark.asyncio
async def test_load_addons_on_boot(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    slug2 = "test_slug_2"
    supervisor.installAddon(slug2, "Test decription")
    slug3 = "test_slug_3"
    supervisor.installAddon(slug3, "Test decription")

    config.override(Setting.STOP_ADDONS, slug1)

    save(config, {slug3}, {slug2})

    await addon_stopper.start(False)
    assert addon_stopper.must_start == {slug3}
    assert addon_stopper.must_enable_watchdog == {slug2}

    addon_stopper.allowRun()
    assert addon_stopper.must_start == {slug1, slug3}
    assert addon_stopper.must_enable_watchdog == {slug2}


@pytest.mark.asyncio
async def test_do_nothing_while_backing_up(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    slug2 = "test_slug_2"
    supervisor.installAddon(slug2, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1, slug2]))

    await addon_stopper.start(False)
    addon_stopper.allowRun()
    addon_stopper.isBackingUp(True)
    assert addon_stopper.must_start == {slug1, slug2}

    await addon_stopper.check()

    assert not interceptor.urlWasCalled(URL_MATCH_START_ADDON)
    assert not interceptor.urlWasCalled(URL_MATCH_STOP_ADDON)


@pytest.mark.asyncio
async def test_start_and_stop(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"

    await addon_stopper.stopAddons("ignore")

    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.check()
    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.startAddons()
    assert supervisor.addon(slug1)["state"] == "started"
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_start_and_stop_error(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"

    await addon_stopper.stopAddons("ignore")

    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.check()
    assert supervisor.addon(slug1)["state"] == "stopped"
    supervisor.addon(slug1)["state"] = "error"
    assert supervisor.addon(slug1)["state"] == "error"
    await addon_stopper.startAddons()
    assert supervisor.addon(slug1)["state"] == "started"
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_stop_failure(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, slug1)
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"
    interceptor.setError(URL_MATCH_STOP_ADDON, 400)

    await addon_stopper.stopAddons("ignore")
    assert interceptor.urlWasCalled(URL_MATCH_STOP_ADDON)
    assert getSaved(config) == (set(), set())
    assert supervisor.addon(slug1)["state"] == "started"
    await addon_stopper.check()
    await addon_stopper.startAddons()
    assert supervisor.addon(slug1)["state"] == "started"
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_start_failure(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor, time: FakeTime) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"

    await addon_stopper.stopAddons("ignore")

    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())
    assert supervisor.addon(slug1)["state"] == "stopped"
    interceptor.setError(URL_MATCH_START_ADDON, 400)
    await addon_stopper.startAddons()
    assert getSaved(config) == (set(), set())
    assert interceptor.urlWasCalled(URL_MATCH_START_ADDON)
    assert supervisor.addon(slug1)["state"] == "stopped"


@pytest.mark.asyncio
async def test_delayed_start(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor, time: FakeTime) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"
    await addon_stopper.stopAddons("ignore")
    assert supervisor.addon(slug1)["state"] == "stopped"
    assert getSaved(config) == ({slug1}, set())

    # start the addon again, which simluates the supervisor's tendency to report an addon as started right after stopping it.
    supervisor.addon(slug1)["state"] = "started"
    await addon_stopper.check()
    await addon_stopper.startAddons()
    assert getSaved(config) == ({slug1}, set())

    time.advance(seconds=30)
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())

    time.advance(seconds=30)
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())

    time.advance(seconds=30)
    supervisor.addon(slug1)["state"] = "stopped"
    await addon_stopper.check()
    assert supervisor.addon(slug1)["state"] == "started"
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_delayed_start_give_up(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor, time: FakeTime) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"
    await addon_stopper.stopAddons("ignore")
    assert supervisor.addon(slug1)["state"] == "stopped"
    assert getSaved(config) == ({slug1}, set())

    # start the addon again, which simluates the supervisor's tendency to report an addon as started right after stopping it.
    supervisor.addon(slug1)["state"] = "started"
    await addon_stopper.check()
    await addon_stopper.startAddons()
    assert getSaved(config) == ({slug1}, set())

    time.advance(seconds=30)
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())

    time.advance(seconds=30)
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())

    # Should clear saved state after this, since it stops checking after 2 minutes.
    time.advance(seconds=100)
    await addon_stopper.check()
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_disable_watchdog(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    supervisor.addon(slug1)["watchdog"] = True

    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"

    await addon_stopper.stopAddons("ignore")

    assert supervisor.addon(slug1)["state"] == "stopped"
    assert supervisor.addon(slug1)["watchdog"] is False
    await addon_stopper.check()
    assert supervisor.addon(slug1)["state"] == "stopped"
    assert supervisor.addon(slug1)["watchdog"] is False
    await addon_stopper.startAddons()
    assert supervisor.addon(slug1)["state"] == "started"
    assert supervisor.addon(slug1)["watchdog"] is True
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_enable_watchdog_on_reboot(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, time: FakeTime) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    supervisor.addon(slug1)["watchdog"] = False
    save(config, set(), {slug1})

    await addon_stopper.start(False)
    addon_stopper.allowRun()
    assert addon_stopper.must_enable_watchdog == {slug1}

    time.advance(minutes=5)
    await addon_stopper.check()
    assert supervisor.addon(slug1)["watchdog"] is True
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_enable_watchdog_waits_for_start(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    supervisor.addon(slug1)["watchdog"] = False
    save(config, {slug1}, {slug1})

    await addon_stopper.start(False)
    addon_stopper.allowRun()
    assert addon_stopper.must_enable_watchdog == {slug1}

    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, {slug1})

    supervisor.addon(slug1)["state"] = "stopped"
    await addon_stopper.check()
    assert supervisor.addon(slug1)["state"] == "started"
    assert supervisor.addon(slug1)["watchdog"] is True
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_get_info_failure_on_stop(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, slug1)
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"
    interceptor.setError(URL_MATCH_ADDON_INFO, 400)

    await addon_stopper.stopAddons("ignore")
    assert interceptor.urlWasCalled(URL_MATCH_ADDON_INFO)
    assert getSaved(config) == (set(), set())
    assert supervisor.addon(slug1)["state"] == "started"
    await addon_stopper.check()
    await addon_stopper.startAddons()
    assert supervisor.addon(slug1)["state"] == "started"
    assert getSaved(config) == (set(), set())


@pytest.mark.asyncio
async def test_get_info_failure_on_start(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor) -> None:
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"

    await addon_stopper.stopAddons("ignore")

    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())
    assert supervisor.addon(slug1)["state"] == "stopped"
    interceptor.setError(URL_MATCH_ADDON_INFO, 400)
    await addon_stopper.startAddons()
    assert getSaved(config) == (set(), set())
    assert interceptor.urlWasCalled(URL_MATCH_ADDON_INFO)
    assert supervisor.addon(slug1)["state"] == "stopped"


@pytest.mark.asyncio
async def test_read_only_fs(supervisor: SimulatedSupervisor, addon_stopper: AddonStopper, config: Config, interceptor: RequestInterceptor) -> None:
    # This test can't be run as the root user, since no file is read-only to root.
    skipForRoot()

    # Stop an addon
    slug1 = "test_slug_1"
    supervisor.installAddon(slug1, "Test decription")
    config.override(Setting.STOP_ADDONS, ",".join([slug1]))
    addon_stopper.allowRun()
    addon_stopper.must_start = set()
    assert supervisor.addon(slug1)["state"] == "started"
    await addon_stopper.stopAddons("ignore")
    assert supervisor.addon(slug1)["state"] == "stopped"
    await addon_stopper.check()
    assert getSaved(config) == ({slug1}, set())

    # make the state file unmodifiable
    os.chmod(config.get(Setting.STOP_ADDON_STATE_PATH), S_IREAD)

    # verify we raise a known error when trying to save.
    with pytest.raises(SupervisorFileSystemError):
        await addon_stopper.startAddons()