# pragma pylint: disable=missing-docstring,C0103

from copy import deepcopy
from datetime import UTC, datetime
from fractions import Fraction
from io import StringIO
from pathlib import Path
from unittest.mock import MagicMock

import numpy as np
import pandas as pd
import pytest

from freqtrade.misc import (
    dataframe_to_json,
    deep_merge_dicts,
    dump_json_to_file,
    file_dump_json,
    file_load_json,
    is_file_in_dir,
    json_to_dataframe,
    pair_to_filename,
    parse_db_uri_for_logging,
    plural,
    safe_value_fallback,
    safe_value_fallback2,
    safe_value_nested,
)


def test_dump_json_to_file_non_finite_values() -> None:
    buffer = StringIO()
    data = {
        "finite": 1.23,
        "numpy_finite": [np.float32(1.5), np.int64(2)],
        "nested": [float("nan"), (np.float32("inf"), np.float64("-inf"))],
    }

    dump_json_to_file(buffer, data)

    assert buffer.getvalue() == (
        '{"finite":1.23,"numpy_finite":[1.5,2],"nested":[null,[null,null]]}'
    )


def test_dump_json_to_file_large_fraction() -> None:
    buffer = StringIO()
    fraction = Fraction(10**1000, 3)

    dump_json_to_file(buffer, {"fraction": fraction})

    assert buffer.getvalue() == f'{{"fraction":"{fraction}"}}'


def test_dump_json_to_file_datetime_format() -> None:
    buffer = StringIO()

    dump_json_to_file(buffer, {"date": datetime(2026, 7, 11, 1, 2, 3, tzinfo=UTC)})

    assert buffer.getvalue() == '{"date":"2026-07-11T01:02:03+00:00"}'


def test_file_dump_json(mocker) -> None:
    file_open = mocker.patch("freqtrade.misc.Path.open", MagicMock())
    json_dump = mocker.patch("orjson.dumps", MagicMock(return_value=b"[1,2,3]"))
    file_dump_json(Path("somefile"), [1, 2, 3])
    assert file_open.call_count == 1
    assert json_dump.call_count == 1
    file_open = mocker.patch("freqtrade.misc.gzip.open", MagicMock())
    json_dump = mocker.patch("orjson.dumps", MagicMock(return_value=b"[1,2,3]"))
    file_dump_json(Path("somefile"), [1, 2, 3], True)
    assert file_open.call_count == 1
    assert json_dump.call_count == 1


def test_file_load_json(mocker, testdatadir) -> None:
    # 7m .json does not exist
    ret = file_load_json(testdatadir / "UNITTEST_BTC-7m.json")
    assert not ret
    # 1m json exists (but no .gz exists)
    ret = file_load_json(testdatadir / "UNITTEST_BTC-1m.json")
    assert ret
    # 8 .json is empty and will fail if it's loaded. .json.gz is a copy of 1.json
    ret = file_load_json(testdatadir / "UNITTEST_BTC-8m.json")
    assert ret


def test_is_file_in_dir(tmp_path):
    # Create a temporary directory and file
    dir_path = tmp_path / "subdir"
    dir_path.mkdir()
    file_path = dir_path / "test.txt"
    file_path.touch()

    # Test that the function returns True when the file is in the directory
    assert is_file_in_dir(file_path, dir_path) is True

    # Test that the function returns False when the file is not in the directory
    assert is_file_in_dir(file_path, tmp_path) is False

    file_path2 = tmp_path / "../../test2.txt"
    assert is_file_in_dir(file_path2, tmp_path) is False


@pytest.mark.parametrize(
    "pair,expected_result",
    [
        ("ETH/BTC", "ETH_BTC"),
        ("ETH/USDT", "ETH_USDT"),
        ("ETH/USDT:USDT", "ETH_USDT_USDT"),  # swap with USDT as settlement currency
        ("ETH/USD:USD", "ETH_USD_USD"),  # swap with USD as settlement currency
        ("AAVE/USD:USD", "AAVE_USD_USD"),  # swap with USDT as settlement currency
        ("ETH/USDT:USDT-210625", "ETH_USDT_USDT-210625"),  # expiring futures
        ("Fabric Token/ETH", "Fabric_Token_ETH"),
        ("ETHH20", "ETHH20"),
        (".XBTBON2H", "_XBTBON2H"),
        ("ETHUSD.d", "ETHUSD_d"),
        ("ADA-0327", "ADA-0327"),
        ("BTC-USD-200110", "BTC-USD-200110"),
        ("BTC-PERP:USDT", "BTC-PERP_USDT"),
        ("F-AKRO/USDT", "F-AKRO_USDT"),
        ("LC+/ETH", "LC__ETH"),
        ("CMT@18/ETH", "CMT_18_ETH"),
        ("LBTC:1022/SAI", "LBTC_1022_SAI"),
        ("$PAC/BTC", "_PAC_BTC"),
        ("ACC_OLD/BTC", "ACC_OLD_BTC"),
    ],
)
def test_pair_to_filename(pair, expected_result):
    pair_s = pair_to_filename(pair)
    assert pair_s == expected_result


def test_safe_value_nested():
    dict1 = {
        "first": {"rows": {"pass": "dog", "number": "1"}},
        "second": "notadict",
        "third": None,
        "fourth": 15,
    }
    assert safe_value_nested(dict1, "first.rows.pass") == "dog"
    assert safe_value_nested(dict1, "first.rows.fail", default_value="cat") == "cat"
    assert safe_value_nested(dict1, "first") == dict1["first"]
    assert safe_value_nested(dict1, "second.rows.pass") is None
    assert safe_value_nested(dict1, "second.rows.pass", default_value="fallback") == "fallback"
    assert safe_value_nested(dict1, "third.rows.pass", default_value="fallback") == "fallback"
    assert safe_value_nested(dict1, "third.rows.pass") is None
    assert safe_value_nested(dict1, "fourth.rows.pass") is None
    assert safe_value_nested(dict1, "fourth") == 15
    assert safe_value_nested(dict1, "fourth", default_value="fallback") == 15
    assert safe_value_nested(dict1, "fourth.rows.pass", default_value="fallback") == "fallback"


def test_safe_value_fallback():
    dict1 = {"keya": None, "keyb": 2, "keyc": 5, "keyd": None}
    assert safe_value_fallback(dict1, "keya", "keyb") == 2
    assert safe_value_fallback(dict1, "keyb", "keya") == 2

    assert safe_value_fallback(dict1, "keyb", "keyc") == 2
    assert safe_value_fallback(dict1, "keya", "keyc") == 5

    assert safe_value_fallback(dict1, "keyc", "keyb") == 5

    assert safe_value_fallback(dict1, "keya", "keyd") is None

    assert safe_value_fallback(dict1, "keyNo", "keyNo") is None
    assert safe_value_fallback(dict1, "keyNo", "keyNo", 55) == 55
    assert safe_value_fallback(dict1, "keyNo", default_value=55) == 55
    assert safe_value_fallback(dict1, "keyNo", None, default_value=55) == 55


def test_safe_value_fallback2():
    dict1 = {"keya": None, "keyb": 2, "keyc": 5, "keyd": None}
    dict2 = {"keya": 20, "keyb": None, "keyc": 6, "keyd": None}
    assert safe_value_fallback2(dict1, dict2, "keya", "keya") == 20
    assert safe_value_fallback2(dict2, dict1, "keya", "keya") == 20

    assert safe_value_fallback2(dict1, dict2, "keyb", "keyb") == 2
    assert safe_value_fallback2(dict2, dict1, "keyb", "keyb") == 2

    assert safe_value_fallback2(dict1, dict2, "keyc", "keyc") == 5
    assert safe_value_fallback2(dict2, dict1, "keyc", "keyc") == 6

    assert safe_value_fallback2(dict1, dict2, "keyd", "keyd") is None
    assert safe_value_fallback2(dict2, dict1, "keyd", "keyd") is None
    assert safe_value_fallback2(dict2, dict1, "keyd", "keyd", 1234) == 1234

    assert safe_value_fallback2(dict1, dict2, "keyNo", "keyNo") is None
    assert safe_value_fallback2(dict2, dict1, "keyNo", "keyNo") is None
    assert safe_value_fallback2(dict2, dict1, "keyNo", "keyNo", 1234) == 1234


def test_plural() -> None:
    assert plural(0, "page") == "pages"
    assert plural(0.0, "page") == "pages"
    assert plural(1, "page") == "page"
    assert plural(1.0, "page") == "page"
    assert plural(2, "page") == "pages"
    assert plural(2.0, "page") == "pages"
    assert plural(-1, "page") == "page"
    assert plural(-1.0, "page") == "page"
    assert plural(-2, "page") == "pages"
    assert plural(-2.0, "page") == "pages"
    assert plural(0.5, "page") == "pages"
    assert plural(1.5, "page") == "pages"
    assert plural(-0.5, "page") == "pages"
    assert plural(-1.5, "page") == "pages"

    assert plural(0, "ox", "oxen") == "oxen"
    assert plural(0.0, "ox", "oxen") == "oxen"
    assert plural(1, "ox", "oxen") == "ox"
    assert plural(1.0, "ox", "oxen") == "ox"
    assert plural(2, "ox", "oxen") == "oxen"
    assert plural(2.0, "ox", "oxen") == "oxen"
    assert plural(-1, "ox", "oxen") == "ox"
    assert plural(-1.0, "ox", "oxen") == "ox"
    assert plural(-2, "ox", "oxen") == "oxen"
    assert plural(-2.0, "ox", "oxen") == "oxen"
    assert plural(0.5, "ox", "oxen") == "oxen"
    assert plural(1.5, "ox", "oxen") == "oxen"
    assert plural(-0.5, "ox", "oxen") == "oxen"
    assert plural(-1.5, "ox", "oxen") == "oxen"


@pytest.mark.parametrize(
    "conn_url,expected",
    [
        (
            "postgresql+psycopg://scott123:scott123@host:1245/dbname",
            "postgresql+psycopg://scott123:*****@host:1245/dbname",
        ),
        (
            "postgresql+psycopg://scott123:scott123@host.name.com/dbname",
            "postgresql+psycopg://scott123:*****@host.name.com/dbname",
        ),
        (
            "mariadb+mariadbconnector://app_user:Password123!@127.0.0.1:3306/company",
            "mariadb+mariadbconnector://app_user:*****@127.0.0.1:3306/company",
        ),
        (
            "mysql+pymysql://user:pass@some_mariadb/dbname?charset=utf8mb4",
            "mysql+pymysql://user:*****@some_mariadb/dbname?charset=utf8mb4",
        ),
        (
            "postgresql+psycopg://scott:p%40ss@host/dbname",
            "postgresql+psycopg://scott:*****@host/dbname",
        ),
        (
            "postgresql+psycopg://scott:pa:ss@host/dbname",
            "postgresql+psycopg://scott:*****@host/dbname",
        ),
        (
            "postgresql+psycopg://scott:scott@[::1]:5432/dbname",
            "postgresql+psycopg://scott:*****@[::1]:5432/dbname",
        ),
        (
            "postgresql+psycopg://scott@host/dbname",
            "postgresql+psycopg://scott@host/dbname",
        ),
        (
            "postgresql+psycopg://host/dbname",
            "postgresql+psycopg://host/dbname",
        ),
        (
            "sqlite:////freqtrade/user_data/tradesv3.sqlite",
            "sqlite:////freqtrade/user_data/tradesv3.sqlite",
        ),
    ],
)
def test_parse_db_uri_for_logging(conn_url, expected) -> None:
    assert parse_db_uri_for_logging(conn_url) == expected


def test_deep_merge_dicts():
    a = {"first": {"rows": {"pass": "dog", "number": "1", "test": None}}}
    b = {"first": {"rows": {"fail": "cat", "number": "5", "test": "asdf"}}}
    res = {"first": {"rows": {"pass": "dog", "fail": "cat", "number": "5", "test": "asdf"}}}
    res2 = {"first": {"rows": {"pass": "dog", "fail": "cat", "number": "1", "test": None}}}
    assert deep_merge_dicts(b, deepcopy(a)) == res

    assert deep_merge_dicts(a, deepcopy(b)) == res2

    res2["first"]["rows"]["test"] = "asdf"
    assert deep_merge_dicts(a, deepcopy(b), allow_null_overrides=False) == res2


def test_dataframe_json(ohlcv_history):
    from pandas.testing import assert_frame_equal

    json = dataframe_to_json(ohlcv_history.copy())
    dataframe = json_to_dataframe(json)

    assert list(ohlcv_history.columns) == list(dataframe.columns)
    assert len(ohlcv_history) == len(dataframe)

    assert_frame_equal(ohlcv_history, dataframe)
    ohlcv_history.at[1, "date"] = pd.NaT
    json = dataframe_to_json(ohlcv_history.copy())

    dataframe = json_to_dataframe(json)