import concurrent.futures
import time
from contextlib import contextmanager
from dataclasses import dataclass

import aqt.operations.note
import pytest
import anki.collection
from _pytest.monkeypatch import MonkeyPatch  # noqa
from pytest_anki._launch import anki_running, temporary_user  # noqa
from waitress import wasyncore

from plugin import AnkiConnect, anki_version
from plugin.edit import Edit
from plugin.util import DEFAULT_CONFIG

try:
    from PyQt6 import QtTest
except ImportError:
    from PyQt5 import QtTest


ac = AnkiConnect()


# wait for n seconds, while events are being processed
def wait(seconds):
    milliseconds = int(seconds * 1000)
    QtTest.QTest.qWait(milliseconds)  # noqa


def wait_until(booleanish_function, at_most_seconds=30):
    deadline = time.time() + at_most_seconds

    while time.time() < deadline:
        if booleanish_function():
            return
        wait(0.01)

    raise Exception(f"Function {booleanish_function} never once returned "
                    f"a positive value in {at_most_seconds} seconds")


def delete_model(model_name):
    model = ac.collection().models.byName(model_name)
    ac.collection().models.remove(model["id"])


def close_all_dialogs_and_wait_for_them_to_run_closing_callbacks():
    aqt.dialogs.closeAll(onsuccess=lambda: None)
    wait_until(aqt.dialogs.allClosed)


def get_dialog_instance(name):
    return aqt.dialogs._dialogs[name][1]    # noqa


# waitress is a WSGI server that Anki starts to serve css etc to its web views.
# it seems to have a race condition issue;
# the main loop thread is trying to `select.select` the sockets
# which a worker thread is closing because of a dead connection.
# this is especially pronounced in tests,
# as we open and close windows rapidly--and so web views and their connections.
# this small patch makes waitress skip actually closing the sockets
# (unless the server is shutting down--if it is, loop exceptions are ignored).
# while the unclosed sockets might accumulate,
# this should not pose an issue in test environment.
# see https://github.com/Pylons/waitress/issues/374
@contextmanager
def waitress_patched_to_prevent_it_from_dying():
    original_close = wasyncore.dispatcher.close
    sockets_that_must_not_be_garbage_collected = []     # lists are thread-safe

    def close(self):
        if not aqt.mw.mediaServer.is_shutdown:
            sockets_that_must_not_be_garbage_collected.append(self.socket)
            self.socket = None
        original_close(self)

    with MonkeyPatch().context() as monkey:
        monkey.setattr(wasyncore.dispatcher, "close", close)
        yield


@contextmanager
def anki_patched_to_prevent_backups():
    with MonkeyPatch().context() as monkey:
        if anki_version < (2, 1, 50):
            monkey.setitem(aqt.profiles.profileConf, "numBackups", 0)
        else:
            monkey.setattr(anki.collection.Collection, "create_backup",
                           lambda *args, **kwargs: True)
        yield


@contextmanager
def empty_anki_session_started():
    with waitress_patched_to_prevent_it_from_dying():
        with anki_patched_to_prevent_backups():
            with anki_running(
                qtbot=None,  # noqa
                enable_web_debugging=False,
                profile_name="test_user",
            ) as session:
                yield session


@contextmanager
def profile_created_and_loaded(session):
    with temporary_user(session.base, "test_user", "en_US"):
        with session.profile_loaded():
            yield session


@contextmanager
def anki_connect_config_loaded(session, web_bind_port):
    with session.addon_config_created(
        package_name="plugin",
        default_config=DEFAULT_CONFIG,
        user_config={**DEFAULT_CONFIG, "webBindPort": web_bind_port}
    ):
        yield


@contextmanager
def current_decks_and_models_etc_preserved():
    deck_names_before = ac.deckNames()
    model_names_before = ac.modelNames()

    try:
        yield
    finally:
        deck_names_after = ac.deckNames()
        model_names_after = ac.modelNames()

        deck_names_to_delete = {*deck_names_after} - {*deck_names_before}
        model_names_to_delete = {*model_names_after} - {*model_names_before}

        ac.deleteDecks(decks=deck_names_to_delete, cardsToo=True)
        for model_name in model_names_to_delete:
            delete_model(model_name)

        ac.guiDeckBrowser()


@dataclass
class Setup:
    deck_id: int
    note1_id: int
    note2_id: int
    note1_card_ids: "list[int]"
    note2_card_ids: "list[int]"
    card_ids: "list[int]"


def set_up_test_deck_and_test_model_and_two_notes():
    ac.createModel(
        modelName="test_model",
        inOrderFields=["field1", "field2"],
        cardTemplates=[
            {"Front": "{{field1}}", "Back": "{{field2}}"},
            {"Front": "{{field2}}", "Back": "{{field1}}"}
        ],
        css="* {}",
    )

    deck_id = ac.createDeck("test_deck")

    note1_id = ac.addNote(dict(
        deckName="test_deck",
        modelName="test_model",
        fields={"field1": "note1 field1", "field2": "note1 field2"},
        tags={"tag1"},
    ))

    note2_id = ac.addNote(dict(
        deckName="test_deck",
        modelName="test_model",
        fields={"field1": "note2 field1", "field2": "note2 field2"},
        tags={"tag2"},
    ))

    note1_card_ids = ac.findCards(query=f"nid:{note1_id}")
    note2_card_ids = ac.findCards(query=f"nid:{note2_id}")
    card_ids = ac.findCards(query="deck:test_deck")

    return Setup(
        deck_id=deck_id,
        note1_id=note1_id,
        note2_id=note2_id,
        note1_card_ids=note1_card_ids,
        note2_card_ids=note2_card_ids,
        card_ids=card_ids,
    )


#############################################################################


def pytest_addoption(parser):
    parser.addoption("--tear-down-profile-after-each-test",
                     action="store_true",
                     default=True)
    parser.addoption("--no-tear-down-profile-after-each-test", "-T",
                     action="store_false",
                     dest="tear_down_profile_after_each_test")


def pytest_report_header(config):
    if config.option.forked:
        return "test isolation: perfect; each test is run in a separate process"
    if config.option.tear_down_profile_after_each_test:
        return "test isolation: good; user profile is torn down after each test"
    else:
        return "test isolation: poor; only newly created decks and models " \
               "are cleaned up between tests"


@pytest.fixture(autouse=True)
def run_background_tasks_on_main_thread(request, monkeypatch):  # noqa
    """
    Makes background operations such as card deletion execute on main thread
    and execute the callback immediately
    """
    def run_in_background(task, on_done=None, kwargs=None):
        future = concurrent.futures.Future()

        try:
            future.set_result(task(**kwargs if kwargs is not None else {}))
        except BaseException as e:
            future.set_exception(e)

        if on_done is not None:
            on_done(future)

    monkeypatch.setattr(aqt.mw.taskman, "run_in_background", run_in_background)


# don't use run_background_tasks_on_main_thread for tests that don't run Anki
def pytest_generate_tests(metafunc):
    if (
        run_background_tasks_on_main_thread.__name__ in metafunc.fixturenames
        and session_scope_empty_session.__name__ not in metafunc.fixturenames
    ):
        metafunc.fixturenames.remove(run_background_tasks_on_main_thread.__name__)


@pytest.fixture(scope="session")
def session_scope_empty_session():
    with empty_anki_session_started() as session:
        yield session


@pytest.fixture(scope="session")
def session_scope_session_with_profile_loaded(session_scope_empty_session):
    with profile_created_and_loaded(session_scope_empty_session):
        yield session_scope_empty_session


@pytest.fixture
def session_with_profile_loaded(session_scope_empty_session, request):
    """
    Like anki_session fixture from pytest-anki, but:
      * Default profile is loaded
      * It's relying on session-wide app instance so that
        it can be used without forking every test;
        this can be useful to speed up tests and also
        to examine Anki's stdout/stderr, which is not visible with forking.
      * If command line option --no-tear-down-profile-after-each-test is passed,
        only the newly created decks and models are deleted.
        Otherwise, the profile is completely torn down after each test.
        Tearing down the profile is significantly slower.
    """
    if request.config.option.tear_down_profile_after_each_test:
        with profile_created_and_loaded(session_scope_empty_session):
            yield session_scope_empty_session
    else:
        session = request.getfixturevalue(
            session_scope_session_with_profile_loaded.__name__
        )
        with current_decks_and_models_etc_preserved():
            yield session


@pytest.fixture
def setup(session_with_profile_loaded):
    """
    Like session_with_profile_loaded, but also:
      * Added are:
        * A deck `test_deck`
        * A model `test_model` with fields `filed1` and `field2`
          and two cards per note
        * Two notes with two valid cards each using the above deck and model
      * Edit dialog is registered with dialog manager
      * Any dialogs, if open, are safely closed on exit
    """
    Edit.register_with_anki()
    yield set_up_test_deck_and_test_model_and_two_notes()
    close_all_dialogs_and_wait_for_them_to_run_closing_callbacks()