# coding=utf-8
# Copyright 2026 HuggingFace Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import importlib
import os
import warnings

import pytest

from diffusers import __version__
from diffusers.utils import deprecate

from ..testing_utils import Expectations, str_to_bool


# Used to test the hub
USER = "__DUMMY_TRANSFORMERS_USER__"
ENDPOINT_STAGING = "https://hub-ci.huggingface.co"

# Not critical, only usable on the sandboxed CI instance.
TOKEN = "hf_94wBhPGp6KrrTH3KDchhKpRxZwd6dmHWLL"


class TestDeprecate:
    higher_version = ".".join([str(int(__version__.split(".")[0]) + 1)] + __version__.split(".")[1:])
    lower_version = "0.0.1"

    def test_deprecate_function_arg(self):
        kwargs = {"deprecated_arg": 4}

        with pytest.warns(FutureWarning) as warning:
            output = deprecate("deprecated_arg", self.higher_version, "message", take_from=kwargs)

        assert output == 4
        assert (
            str(warning[0].message)
            == f"The `deprecated_arg` argument is deprecated and will be removed in version {self.higher_version}."
            " message"
        )

    def test_deprecate_function_arg_tuple(self):
        kwargs = {"deprecated_arg": 4}

        with pytest.warns(FutureWarning) as warning:
            output = deprecate(("deprecated_arg", self.higher_version, "message"), take_from=kwargs)

        assert output == 4
        assert (
            str(warning[0].message)
            == f"The `deprecated_arg` argument is deprecated and will be removed in version {self.higher_version}."
            " message"
        )

    def test_deprecate_function_args(self):
        kwargs = {"deprecated_arg_1": 4, "deprecated_arg_2": 8}
        with pytest.warns(FutureWarning) as warning:
            output_1, output_2 = deprecate(
                ("deprecated_arg_1", self.higher_version, "Hey"),
                ("deprecated_arg_2", self.higher_version, "Hey"),
                take_from=kwargs,
            )
        assert output_1 == 4
        assert output_2 == 8
        assert (
            str(warning[0].message) == "The `deprecated_arg_1` argument is deprecated and will be removed in version"
            f" {self.higher_version}. Hey"
        )
        assert (
            str(warning[1].message) == "The `deprecated_arg_2` argument is deprecated and will be removed in version"
            f" {self.higher_version}. Hey"
        )

    def test_deprecate_function_incorrect_arg(self):
        kwargs = {"deprecated_arg": 4}

        with pytest.raises(TypeError) as error:
            deprecate(("wrong_arg", self.higher_version, "message"), take_from=kwargs)

        assert "test_deprecate_function_incorrect_arg in" in str(error.value)
        assert "line" in str(error.value)
        assert "got an unexpected keyword argument `deprecated_arg`" in str(error.value)

    def test_deprecate_arg_no_kwarg(self):
        with pytest.warns(FutureWarning) as warning:
            deprecate(("deprecated_arg", self.higher_version, "message"))

        assert (
            str(warning[0].message)
            == f"`deprecated_arg` is deprecated and will be removed in version {self.higher_version}. message"
        )

    def test_deprecate_args_no_kwarg(self):
        with pytest.warns(FutureWarning) as warning:
            deprecate(
                ("deprecated_arg_1", self.higher_version, "Hey"),
                ("deprecated_arg_2", self.higher_version, "Hey"),
            )
        assert (
            str(warning[0].message)
            == f"`deprecated_arg_1` is deprecated and will be removed in version {self.higher_version}. Hey"
        )
        assert (
            str(warning[1].message)
            == f"`deprecated_arg_2` is deprecated and will be removed in version {self.higher_version}. Hey"
        )

    def test_deprecate_class_obj(self):
        class Args:
            arg = 5

        with pytest.warns(FutureWarning) as warning:
            arg = deprecate(("arg", self.higher_version, "message"), take_from=Args())

        assert arg == 5
        assert (
            str(warning[0].message)
            == f"The `arg` attribute is deprecated and will be removed in version {self.higher_version}. message"
        )

    def test_deprecate_class_objs(self):
        class Args:
            arg = 5
            foo = 7

        with pytest.warns(FutureWarning) as warning:
            arg_1, arg_2 = deprecate(
                ("arg", self.higher_version, "message"),
                ("foo", self.higher_version, "message"),
                ("does not exist", self.higher_version, "message"),
                take_from=Args(),
            )

        assert arg_1 == 5
        assert arg_2 == 7
        assert (
            str(warning[0].message)
            == f"The `arg` attribute is deprecated and will be removed in version {self.higher_version}. message"
        )
        assert (
            str(warning[1].message)
            == f"The `foo` attribute is deprecated and will be removed in version {self.higher_version}. message"
        )

    def test_deprecate_incorrect_version(self):
        kwargs = {"deprecated_arg": 4}

        with pytest.raises(ValueError) as error:
            deprecate(("wrong_arg", self.lower_version, "message"), take_from=kwargs)

        assert (
            str(error.value)
            == "The deprecation tuple ('wrong_arg', '0.0.1', 'message') should be removed since diffusers' version"
            f" {__version__} is >= {self.lower_version}"
        )

    def test_deprecate_incorrect_no_standard_warn(self):
        with pytest.warns(FutureWarning) as warning:
            deprecate(("deprecated_arg", self.higher_version, "This message is better!!!"), standard_warn=False)

        assert str(warning[0].message) == "This message is better!!!"

    def test_deprecate_stacklevel(self):
        with pytest.warns(FutureWarning) as warning:
            deprecate(("deprecated_arg", self.higher_version, "This message is better!!!"), standard_warn=False)
        assert str(warning[0].message) == "This message is better!!!"
        assert "diffusers/tests/others/test_utils.py" in warning[0].filename

    def test_deprecate_testing_utils_module(self):
        import diffusers.utils.testing_utils

        with warnings.catch_warnings(record=True) as caught_warnings:
            warnings.simplefilter("always")
            importlib.reload(diffusers.utils.testing_utils)

        deprecation_warnings = [w for w in caught_warnings if issubclass(w.category, FutureWarning)]
        assert len(deprecation_warnings) >= 1, "Expected at least one FutureWarning from diffusers.utils.testing_utils"

        messages = [str(w.message) for w in deprecation_warnings]
        assert any("diffusers.utils.testing_utils" in msg for msg in messages), (
            f"Expected a deprecation warning mentioning 'diffusers.utils.testing_utils', got: {messages}"
        )
        assert any(
            "diffusers.utils.testing_utils is deprecated and will be removed in a future version." in msg
            for msg in messages
        ), f"Expected deprecation message substring not found, got: {messages}"


class TestFourierFilter:
    """Tests for :func:`diffusers.utils.torch_utils.fourier_filter` (FreeU helper)."""

    def _run_without_complexhalf_warning(self, dtype):
        import torch

        from diffusers.utils.torch_utils import fourier_filter

        x = torch.randn(1, 4, 32, 32, dtype=dtype)
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            out = fourier_filter(x, threshold=1, scale=0.5)

        messages = [str(w.message) for w in caught]
        assert not any("ComplexHalf" in m for m in messages), (
            f"Unexpected ComplexHalf warning emitted by fourier_filter: {messages}"
        )
        return out

    def test_fourier_filter_float16_no_complexhalf_warning(self):
        import torch

        out = self._run_without_complexhalf_warning(torch.float16)
        assert out.dtype == torch.float16

    def test_fourier_filter_bfloat16_no_complexhalf_warning(self):
        import torch

        out = self._run_without_complexhalf_warning(torch.bfloat16)
        assert out.dtype == torch.bfloat16

    def test_fourier_filter_preserves_dtype_and_shape(self):
        import torch

        from diffusers.utils.torch_utils import fourier_filter

        for dtype in (torch.float32, torch.float16, torch.bfloat16):
            x = torch.randn(2, 3, 16, 16, dtype=dtype)
            out = fourier_filter(x, threshold=1, scale=0.5)
            assert out.dtype == dtype
            assert out.shape == x.shape


class TestRandnTensor:
    """Tests for :func:`diffusers.utils.torch_utils.randn_tensor`."""

    def test_mps_suppresses_cpu_generator_info_log(self):
        import torch

        from diffusers.utils import logging as diffusers_logging
        from diffusers.utils import torch_utils

        from ..testing_utils import CaptureLogger

        gen = torch.Generator(device="cpu")
        diffusers_logging.set_verbosity_info()

        def _capture(target_device):
            with CaptureLogger(torch_utils.logger) as cl:
                try:
                    torch_utils.randn_tensor((1, 2), generator=gen, device=target_device, dtype=torch.float32)
                except (AssertionError, RuntimeError):
                    pass
            return cl.out

        mps_out = _capture("mps")
        assert "moved to" not in mps_out, f"MPS target should not emit the CPU-fallback info log, got: {mps_out}"

        cuda_out = _capture("cuda")
        assert "moved to" in cuda_out, f"Non-MPS target should still emit the CPU-fallback info log, got: {cuda_out}"


# Copied from https://github.com/huggingface/transformers/blob/main/tests/utils/test_expectations.py
class TestExpectations:
    def test_expectations(self):
        expectations = Expectations(
            {
                (None, None): 1,
                ("cuda", 8): 2,
                ("cuda", 7): 3,
                ("rocm", 8): 4,
                ("rocm", None): 5,
                ("cpu", None): 6,
                ("xpu", 3): 7,
            }
        )

        def check(value, key):
            assert expectations.find_expectation(key) == value

        # npu has no matches so should find default expectation
        check(1, ("npu", None))
        check(7, ("xpu", 3))
        check(2, ("cuda", 8))
        check(3, ("cuda", 7))
        check(4, ("rocm", 9))
        check(4, ("rocm", None))
        check(2, ("cuda", 2))

        expectations = Expectations({("cuda", 8): 1})
        with pytest.raises(ValueError):
            expectations.find_expectation(("xpu", None))


def parse_flag_from_env(key, default=False):
    try:
        value = os.environ[key]
    except KeyError:
        # KEY isn't set, default to `default`.
        _value = default
    else:
        # KEY is set, convert it to True or False.
        try:
            _value = str_to_bool(value)
        except ValueError:
            # More values are supported, but let's keep the message simple.
            raise ValueError(f"If set, {key} must be yes or no.")
    return _value


_run_staging = parse_flag_from_env("HUGGINGFACE_CO_STAGING", default=False)


def is_staging_test(test_case):
    """
    Decorator marking a test as a staging test.

    Those tests will run using the staging environment of huggingface.co instead of the real model hub.
    """
    if not _run_staging:
        return pytest.mark.skip("test is staging test")(test_case)
    else:
        return pytest.mark.is_staging_test()(test_case)