from __future__ import annotations

import base64
import hashlib
import re

# noinspection PyProtectedMember
from importlib.metadata import FileHash, PackagePath
from pathlib import Path
from typing import cast
from unittest.mock import Mock, call

import pytest

import streamlink.plugins
import tests.plugin
from streamlink.options import Argument, Arguments

# noinspection PyProtectedMember
from streamlink.plugin.plugin import (
    HIGH_PRIORITY,
    LOW_PRIORITY,
    NO_PRIORITY,
    NORMAL_PRIORITY,
    Matcher,
    Matchers,
    Plugin,
    pluginargument,
    pluginmatcher,
)
from streamlink.session import Streamlink
from streamlink.session.plugins import StreamlinkPlugins
from streamlink.utils.args import boolean, comma_list_filter
from tests.plugin.testplugin import TestPlugin as _TestPlugin


PATH_BUILTINPLUGINS = Path(streamlink.plugins.__path__[0])
PATH_TESTPLUGINS = Path(tests.plugin.__path__[0])
PATH_TESTPLUGINS_OVERRIDE = PATH_TESTPLUGINS / "override"


class _Plugin(Plugin):
    def _get_streams(self):  # pragma: no cover
        pass


@pytest.fixture(autouse=True)
def caplog(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
    caplog.set_level(1, "streamlink")
    return caplog


@pytest.fixture(scope="module")
def fake_plugin():
    @pluginmatcher(re.compile(r"fake"))
    @pluginargument("foo")
    @pluginargument("bar")
    class FakePlugin(_Plugin):
        __module__ = "streamlink.plugins.fake"

    return FakePlugin


def test_empty(caplog: pytest.LogCaptureFixture, session: Streamlink):
    assert session.plugins.get_names() == []
    assert session.plugins.get_loaded() == {}
    assert caplog.record_tuples == []


def test_set_get_del(session: Streamlink, fake_plugin: type[Plugin]):
    assert "fake" not in session.plugins

    session.plugins["fake"] = fake_plugin
    assert "fake" in session.plugins
    assert session.plugins["fake"] is fake_plugin
    assert session.plugins.get_names() == ["fake"]
    assert session.plugins.get_loaded() == {"fake": fake_plugin}
    assert session.plugins.get_loaded() is not session.plugins.get_loaded()

    del session.plugins["fake"]
    assert "fake" not in session.plugins
    assert session.plugins.get_names() == []
    assert session.plugins.get_loaded() == {}


def test_update_clear(session: Streamlink, fake_plugin: type[Plugin]):
    assert "fake" not in session.plugins

    session.plugins.update({"fake": fake_plugin})
    assert "fake" in session.plugins
    assert session.plugins["fake"] is fake_plugin
    assert session.plugins.get_names() == ["fake"]
    assert session.plugins.get_loaded() == {"fake": fake_plugin}

    session.plugins.clear()
    assert "fake" not in session.plugins
    assert session.plugins.get_names() == []
    assert session.plugins.get_loaded() == {}


def test_iter_arguments(session: Streamlink, fake_plugin: type[Plugin]):
    session.plugins.update({"fake": fake_plugin})
    assert [(name, [arg.argument_name(name) for arg in args]) for name, args in session.plugins.iter_arguments()] == [
        ("fake", ["--fake-foo", "--fake-bar"]),
    ]


class TestLoad:
    def test_load_builtin(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, fake_plugin: type[Plugin]):
        mock = Mock(return_value={"fake": fake_plugin})
        monkeypatch.setattr(StreamlinkPlugins, "_load_plugins_from_path", mock)
        session = Streamlink(plugins_builtin=True, plugins_lazy=False)

        assert mock.call_args_list == [call(PATH_BUILTINPLUGINS)]
        assert "fake" in session.plugins
        assert session.plugins.get_names() == ["fake"]
        assert session.plugins.get_loaded() == {"fake": fake_plugin}
        assert session.plugins["fake"].__module__ == "streamlink.plugins.fake"
        assert caplog.record_tuples == []

    def test_load_path_empty(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, session: Streamlink):
        assert not session.plugins.load_path(tmp_path)
        assert session.plugins.get_names() == []
        assert session.plugins.get_loaded() == {}
        assert caplog.record_tuples == []

    def test_load_path_testplugins(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
        assert session.plugins.load_path(PATH_TESTPLUGINS)
        assert "testplugin" in session.plugins
        assert "testplugin_invalid" not in session.plugins
        assert "testplugin_missing" not in session.plugins
        assert session.plugins.get_names() == ["testplugin"]
        assert session.plugins["testplugin"].__name__ == "TestPlugin"
        assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin"
        assert caplog.record_tuples == []

    def test_load_path_testplugins_override(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
        assert session.plugins.load_path(PATH_TESTPLUGINS)
        assert "testplugin" in session.plugins
        assert caplog.record_tuples == []

        assert session.plugins.load_path(PATH_TESTPLUGINS_OVERRIDE)
        assert "testplugin" in session.plugins
        assert session.plugins.get_names() == ["testplugin"]
        assert session.plugins["testplugin"].__name__ == "TestPluginOverride"
        assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin"
        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
            (
                "streamlink.session",
                "info",
                f"Plugin testplugin is being overridden by {PATH_TESTPLUGINS_OVERRIDE / 'testplugin.py'}"
                + " (sha256:47d9c5ec4167565db13aae76f2fbedcf961f5fca386e60c4384ee5e92714b5f3)",
            ),
        ]

    def test_load_path_testplugins_override_matchers(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
        assert _TestPlugin.matchers
        session.plugins._matchers.update({"testplugin": _TestPlugin.matchers})

        assert "testplugin" not in session.plugins
        assert session.plugins.get_names() == ["testplugin"]
        assert caplog.record_tuples == []

        assert session.plugins.load_path(PATH_TESTPLUGINS)
        assert session.plugins.get_names() == ["testplugin"]
        assert session.plugins["testplugin"].__name__ == "TestPlugin"
        assert session.plugins["testplugin"].__module__ == "streamlink.plugins.testplugin"
        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
            (
                "streamlink.session",
                "info",
                f"Plugin testplugin is being overridden by {PATH_TESTPLUGINS / 'testplugin.py'}"
                + " (sha256:088c9f6ddbe5ff046c0ea1ce0cacb7baff46189d153cf3f149b2e023ddf66f6c)",
            ),
        ]

    def test_importerror(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink):
        monkeypatch.setattr("importlib.machinery.FileFinder.find_spec", Mock(return_value=None))
        assert not session.plugins.load_path(PATH_TESTPLUGINS)
        assert "testplugin" not in session.plugins
        assert session.plugins.get_names() == []
        assert [(record.name, record.levelname, record.message, bool(record.exc_info)) for record in caplog.records] == [
            (
                "streamlink.session",
                "error",
                f"Failed to load plugin testplugin from {PATH_TESTPLUGINS}\n",
                True,
            ),
            (
                "streamlink.session",
                "error",
                f"Failed to load plugin testplugin_invalid from {PATH_TESTPLUGINS}\n",
                True,
            ),
            (
                "streamlink.session",
                "error",
                f"Failed to load plugin testplugin_missing from {PATH_TESTPLUGINS}\n",
                True,
            ),
        ]

    def test_syntaxerror(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink):
        monkeypatch.setattr("importlib.machinery.SourceFileLoader.exec_module", Mock(side_effect=SyntaxError))
        with pytest.raises(SyntaxError):
            session.plugins.load_path(PATH_TESTPLUGINS)
        assert session.plugins.get_names() == []
        assert caplog.record_tuples == []


class TestLoadPluginsData:
    @pytest.fixture()
    def session(self, monkeypatch: pytest.MonkeyPatch, fake_plugin: type[Plugin], metadata_files: Mock):
        class MockStreamlinkPlugins(StreamlinkPlugins):
            def load_builtin(self):
                self._plugins.update({"fake": fake_plugin})
                return True

        monkeypatch.setattr("streamlink.session.session.StreamlinkPlugins", MockStreamlinkPlugins)

        return Streamlink(plugins_builtin=True, plugins_lazy=True)

    @pytest.fixture(autouse=True)
    def metadata_files(self, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch, pluginsdata: str | None):
        options = getattr(request, "param", {})
        package_record = options.get("package-record", True)
        mode = options.get("package-record-hash-mode", "sha256")
        filehash = options.get("package-record-hash-value", None)

        mock = Mock(return_value=[])
        monkeypatch.setattr("importlib.metadata.files", mock)

        if not package_record or pluginsdata is None:
            yield mock
            return

        if filehash is None:
            # https://packaging.python.org/en/latest/specifications/recording-installed-packages/#the-record-file
            digest = hashlib.sha256(pluginsdata.encode("utf-8")).digest()
            filehash = base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=")

        packagepath = PackagePath("streamlink", "plugins", "_plugins.json")
        packagepath.hash = FileHash(f"{mode}={filehash}")

        mock.return_value = [packagepath]
        yield mock

        assert mock.call_args_list == [call("streamlink")]

    @pytest.fixture()
    def pluginsdata(self, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch):
        data = getattr(request, "param", "{}")

        mock = Mock()
        monkeypatch.setattr("streamlink.session.plugins._PLUGINSDATA_PATH", mock)

        if data is None:
            mock.read_bytes.side_effect = FileNotFoundError
        else:
            mock.read_bytes.return_value = data.encode("utf-8")

        return data

    # noinspection PyTestParametrized
    @pytest.mark.parametrize(
        ("metadata_files", "pluginsdata", "logs"),
        [
            pytest.param(
                {},
                None,
                [],
                id="empty-json-file",
            ),
            pytest.param(
                {"package-record": False},
                """{}""",
                [],
                id="no-package-record",
            ),
            pytest.param(
                {"package-record-hash-mode": "unknown"},
                """{}""",
                [("streamlink.session", "error", "Unknown plugins data hash mode, falling back to loading all plugins")],
                id="invalid-package-record-hash-mode",
            ),
            pytest.param(
                {"package-record-hash-value": "invalid"},
                """{}""",
                [("streamlink.session", "error", "Plugins data checksum mismatch, falling back to loading all plugins")],
                id="invalid-package-record-hash-value",
            ),
        ],
        indirect=["metadata_files", "pluginsdata"],
    )
    def test_fallback_load_builtin(self, caplog: pytest.LogCaptureFixture, session: Streamlink, logs: list):
        assert session.plugins.get_names() == ["fake"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == logs

    # noinspection JsonStandardCompliance
    @pytest.mark.parametrize(
        "pluginsdata",
        [
            pytest.param(
                # language=json
                """
                    // foo
                    // bar
                    {
                        "testplugin": {
                            "matchers": [
                                {"pattern": "foo"}
                            ],
                            "arguments": []
                        }
                    }
                """,
                id="json-comments",
            ),
        ],
        indirect=True,
    )
    def test_strip_json_comment(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
        assert "fake" not in session.plugins
        assert "testplugin" not in session.plugins
        assert session.plugins.get_names() == ["testplugin"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == []

    @pytest.mark.parametrize(
        "pluginsdata",
        [
            pytest.param(
                # language=json
                """
                    {
                        "testpluginA": {
                            "matchers": [
                                {"pattern": "foo"},
                                {"pattern": "bar", "flags": 64, "priority": 10, "name": "bar"}
                            ],
                            "arguments": []
                        },
                        "testpluginB": {
                            "matchers": [
                                {"pattern": "baz"}
                            ],
                            "arguments": []
                        }
                    }
                """,
                id="matchers",
            ),
        ],
        indirect=True,
    )
    def test_matchers(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
        assert "fake" not in session.plugins
        assert "testpluginA" not in session.plugins
        assert "testpluginB" not in session.plugins
        assert session.plugins.get_names() == ["testpluginA", "testpluginB"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == []

        matchers_a = Matchers()
        matchers_a.add(Matcher(pattern=re.compile(r"foo"), priority=NORMAL_PRIORITY, name=None))
        matchers_a.add(Matcher(pattern=re.compile(r"bar", re.VERBOSE), priority=10, name="bar"))
        matchers_b = Matchers()
        matchers_b.add(Matcher(pattern=re.compile(r"baz"), priority=NORMAL_PRIORITY, name=None))
        assert list(session.plugins.iter_matchers()) == [("testpluginA", matchers_a), ("testpluginB", matchers_b)]

    @pytest.mark.parametrize(
        "pluginsdata",
        [
            pytest.param(
                # language=json
                """
                    {
                        "success": {
                            "matchers": [{"pattern": "foo"}],
                            "arguments": []
                        },
                        "fail": {
                            "matchers": [{"pattern": {"invalid": "type"}}],
                            "arguments": []
                        }
                    }
                """,
                id="matchers",
            ),
        ],
        indirect=True,
    )
    def test_matchers_failure(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
        assert "fake" in session.plugins
        assert "success" not in session.plugins
        assert "fail" not in session.plugins
        assert session.plugins.get_names() == ["fake"]
        assert [name for name, matchers in session.plugins.iter_matchers()] == ["fake"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == [
            ("streamlink.session", "error", "Error while loading pluginmatcher data from JSON"),
        ]

    @pytest.mark.parametrize(
        "pluginsdata",
        [
            pytest.param(
                # language=json
                """
                    {
                        "empty": {
                            "matchers": [{"pattern": "foo"}]
                        },
                        "testpluginA": {
                            "matchers": [{"pattern": "bar"}],
                            "arguments": [
                                {
                                    "name": "foo",
                                    "action": "store",
                                    "nargs": 1,
                                    "default": "foo",
                                    "choices": ["foo", "bar"],
                                    "required": true,
                                    "help": "foo",
                                    "metavar": "FOO",
                                    "dest": "oof",
                                    "argument_name": "oof"
                                },
                                {
                                    "name": "bar",
                                    "const": "bar"
                                }
                            ]
                        },
                        "testpluginB": {
                            "matchers": [{"pattern": "baz"}],
                            "arguments": [
                                {
                                    "name": "invalid",
                                    "type": "type_which_does_not_exist"
                                },
                                {
                                    "name": "bool",
                                    "type": "bool"
                                },
                                {
                                    "name": "cmf",
                                    "type": "comma_list_filter",
                                    "type_args": [["1", "2", "3"]],
                                    "type_kwargs": {"unique": true}
                                }
                            ]
                        }
                    }
                """,
                id="arguments",
            ),
        ],
        indirect=True,
    )
    def test_arguments(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
        assert "fake" not in session.plugins
        assert "testpluginA" not in session.plugins
        assert "testpluginB" not in session.plugins
        assert session.plugins.get_names() == ["empty", "testpluginA", "testpluginB"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == []

        # arguments are added in reverse order:
        # `Arguments` does this because of the reverse order of the @pluginargument decorator
        arguments_a = Arguments()
        arguments_a.add(
            Argument(
                name="bar",
                const="bar",
            ),
        )
        arguments_a.add(
            Argument(
                name="foo",
                action="store",
                nargs=1,
                default="foo",
                choices=["foo", "bar"],
                required=True,
                help="foo",
                metavar="FOO",
                dest="oof",
                argument_name="oof",
            ),
        )
        arguments_b = Arguments()
        arguments_b.add(
            Argument(
                name="cmf",
                type=comma_list_filter(["1", "2", "3"], unique=True),
            ),
        )
        arguments_b.add(
            Argument(
                name="bool",
                type=boolean,
            ),
        )
        assert list(session.plugins.iter_arguments()) == [("testpluginA", arguments_a), ("testpluginB", arguments_b)]

    @pytest.mark.parametrize(
        "pluginsdata",
        [
            pytest.param(
                # language=json
                """
                    {
                        "success": {
                            "matchers": [{"pattern": "foo"}],
                            "arguments": [{"name": "foo"}]
                        },
                        "fail": {
                            "matchers": [{"pattern": "bar"}],
                            "arguments": [{"name": {"invalid": "type"}}]
                        }
                    }
                """,
                id="arguments",
            ),
        ],
        indirect=True,
    )
    def test_arguments_failure(self, caplog: pytest.LogCaptureFixture, session: Streamlink, pluginsdata: str):
        assert "fake" in session.plugins
        assert "success" not in session.plugins
        assert "fail" not in session.plugins
        assert session.plugins.get_names() == ["fake"]
        assert [name for name, arguments in session.plugins.iter_arguments()] == ["fake"]
        assert [(record.name, record.levelname, record.message) for record in caplog.get_records(when="setup")] == [
            ("streamlink.session", "error", "Error while loading pluginargument data from JSON"),
        ]


class TestMatchURL:
    def test_priority(self, session: Streamlink):
        @pluginmatcher(priority=HIGH_PRIORITY, pattern=re.compile(r"^(high|normal|low|no)$"))
        class HighPriority(_Plugin):
            pass

        @pluginmatcher(priority=NORMAL_PRIORITY, pattern=re.compile(r"^(normal|low|no)$"))
        class NormalPriority(_Plugin):
            pass

        @pluginmatcher(priority=LOW_PRIORITY, pattern=re.compile(r"^(low|no)$"))
        class LowPriority(_Plugin):
            pass

        @pluginmatcher(priority=NO_PRIORITY, pattern=re.compile(r"^no$"))
        class NoPriority(_Plugin):
            pass

        session.plugins.update({
            "high": HighPriority,
            "normal": NormalPriority,
            "low": LowPriority,
            "no": NoPriority,
        })

        assert session.plugins.match_url("no") == ("high", HighPriority)
        assert session.plugins.match_url("low") == ("high", HighPriority)
        assert session.plugins.match_url("normal") == ("high", HighPriority)
        assert session.plugins.match_url("high") == ("high", HighPriority)

        del session.plugins["high"]

        assert session.plugins.match_url("no") == ("normal", NormalPriority)
        assert session.plugins.match_url("low") == ("normal", NormalPriority)
        assert session.plugins.match_url("normal") == ("normal", NormalPriority)
        assert session.plugins.match_url("high") is None

    def test_no_priority(self, session: Streamlink):
        @pluginmatcher(priority=NO_PRIORITY, pattern=re.compile(r"^no$"))
        class NoPriority(_Plugin):
            pass

        session.plugins.update({
            "no": NoPriority,
        })
        assert session.plugins.match_url("no") is None


class TestMatchURLLoadLazy:
    @pytest.fixture(autouse=True)
    def _pluginsdir(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setattr("streamlink.session.plugins._PLUGINS_PATH", PATH_TESTPLUGINS)

    @pytest.fixture()
    def _loaded_plugins(self, session: Streamlink):
        # loaded built-in or custom plugin
        session.plugins.update({"testplugin": _TestPlugin})
        assert session.plugins.get_loaded() == {"testplugin": _TestPlugin}

    @pytest.fixture()
    def _loaded_matchers(self, session: Streamlink):
        matchers = cast("Matchers", _TestPlugin.matchers)
        session.plugins._matchers.update({"testplugin": matchers})
        assert session.plugins.get_loaded() == {}
        assert session.plugins.get_names() == ["testplugin"]

    @pytest.mark.usefixtures("_loaded_plugins")
    def test_loaded(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
        assert session.plugins.match_url("http://test.se") == ("testplugin", _TestPlugin)
        assert [(record.name, record.levelname, record.message) for record in caplog.records] == []

    @pytest.mark.usefixtures("_loaded_matchers")
    def test_load(self, caplog: pytest.LogCaptureFixture, session: Streamlink):
        lookup = session.plugins.match_url("http://test.se")
        assert lookup is not None
        name, plugin = lookup
        assert name == "testplugin"
        # can't compare plugin classes here due to exec_module(), so just compare module name, matchers and arguments
        assert plugin.__module__ == "streamlink.plugins.testplugin"
        assert plugin.matchers == _TestPlugin.matchers
        assert plugin.arguments == _TestPlugin.arguments
        assert "testplugin" in session.plugins.get_loaded()
        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
            ("streamlink.session", "debug", "Loading plugin: testplugin"),
        ]

    @pytest.mark.usefixtures("_loaded_matchers")
    def test_fail_builtin(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, session: Streamlink):
        path = str(PATH_TESTPLUGINS / "testplugin.py")
        mock = Mock(side_effect=ImportError("", path=path))
        monkeypatch.setattr("streamlink.session.plugins.exec_module", mock)

        assert session.plugins.match_url("http://test.se") is None
        assert session.plugins.get_loaded() == {}
        assert [(record.name, record.levelname, record.message) for record in caplog.records] == [
            ("streamlink.session", "debug", "Loading plugin: testplugin"),
            ("streamlink.session", "error", f"Failed to load plugin testplugin from {path}\n"),
        ]