# coding: utf-8
"""Tests for the IPython tab-completion machinery."""
# -----------------------------------------------------------------------------
# Module imports
# -----------------------------------------------------------------------------

# stdlib
import io
import gc
import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path

from tempfile import TemporaryDirectory

# our own packages
from traitlets.config.loader import Config

from IPython.core.history import HistoryAccessor, HistoryManager, extract_hist_ranges

import pytest


def test_proper_default_encoding():
    assert sys.getdefaultencoding() == "utf-8"


def hmmax_instance_maker(N: int):
    if os.name == "nt":

        @pytest.fixture()
        def inner():
            pass

    else:

        @pytest.fixture()
        def inner():
            assert HistoryManager._max_inst == 1
            HistoryManager._max_inst = N
            lh = len(HistoryManager._instances)
            try:
                yield
                gc.collect()
                assert len(HistoryManager._instances) == lh
            finally:
                HistoryManager._max_inst = 1

    return inner


hmmax2 = hmmax_instance_maker(2)
hmmax3 = hmmax_instance_maker(3)


def test_history(hmmax2):
    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        tmp_path = Path(tmpdir)
        hist_manager_ori = ip.history_manager
        hist_file = tmp_path / "history_test_history1.sqlite"
        try:
            ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file)
            hist = ["a=1", "def f():\n    test = 1\n    return test", "b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)

            ip.history_manager.db_log_output = True
            # Doesn't match the input, but we'll just check it's stored.
            ip.history_manager.output_hist_reprs[3] = "spam"
            ip.history_manager.store_output(3)

            assert ip.history_manager.input_hist_raw == [""] + hist

            # Detailed tests for _get_range_session
            grs = ip.history_manager._get_range_session
            assert list(grs(start=2, stop=-1)) == list(zip([0], [2], hist[1:-1]))
            assert list(grs(start=-2)) == list(zip([0, 0], [2, 3], hist[-2:]))
            assert list(grs(output=True)) == list(
                zip([0, 0, 0], [1, 2, 3], zip(hist, [None, None, "spam"]))
            )

            # Check whether specifying a range beyond the end of the current
            # session results in an error (gh-804)
            ip.run_line_magic("hist", "2-500")

            # Check that we can write non-ascii characters to a file
            ip.run_line_magic("hist", "-f %s" % (tmp_path / "test1"))
            ip.run_line_magic("hist", "-pf %s" % (tmp_path / "test2"))
            ip.run_line_magic("hist", "-nf %s" % (tmp_path / "test3"))
            ip.run_line_magic("save", "%s 1-10" % (tmp_path / "test4"))

            # New session
            ip.history_manager.reset()
            newcmds = ["z=5", "class X(object):\n    pass", "k='p'", "z=5"]
            for i, cmd in enumerate(newcmds, start=1):
                ip.history_manager.store_inputs(i, cmd)
            gothist = ip.history_manager.get_range(start=1, stop=4)
            assert list(gothist) == list(zip([0, 0, 0], [1, 2, 3], newcmds))
            # Previous session:
            gothist = ip.history_manager.get_range(-1, 1, 4)
            assert list(gothist) == list(zip([1, 1, 1], [1, 2, 3], hist))

            newhist = [(2, i, c) for (i, c) in enumerate(newcmds, 1)]

            # Check get_hist_tail
            gothist = ip.history_manager.get_tail(5, output=True, include_latest=True)
            expected = [(1, 3, (hist[-1], "spam"))] + [
                (s, n, (c, None)) for (s, n, c) in newhist
            ]
            assert list(gothist) == expected

            gothist = ip.history_manager.get_tail(2)
            expected = newhist[-3:-1]
            assert list(gothist) == expected

            # Check get_hist_search

            gothist = ip.history_manager.search("*test*")
            assert list(gothist) == [(1, 2, hist[1])]

            gothist = ip.history_manager.search("*=*")
            assert list(gothist) == [
                (1, 1, hist[0]),
                (1, 2, hist[1]),
                (1, 3, hist[2]),
                newhist[0],
                newhist[2],
                newhist[3],
            ]

            gothist = ip.history_manager.search("*=*", n=4)
            assert list(gothist) == [
                (1, 3, hist[2]),
                newhist[0],
                newhist[2],
                newhist[3],
            ]

            gothist = ip.history_manager.search("*=*", unique=True)
            assert list(gothist) == [
                (1, 1, hist[0]),
                (1, 2, hist[1]),
                (1, 3, hist[2]),
                newhist[2],
                newhist[3],
            ]

            gothist = ip.history_manager.search("*=*", unique=True, n=3)
            assert list(gothist) == [(1, 3, hist[2]), newhist[2], newhist[3]]

            gothist = ip.history_manager.search("b*", output=True)
            assert list(gothist) == [(1, 3, (hist[2], "spam"))]

            # Cross testing: check that magic %save can get previous session.
            testfilename = (tmp_path / "test.py").resolve()
            ip.run_line_magic("save", str(testfilename) + " ~1/1-3")
            with io.open(testfilename, encoding="utf-8") as testfile:
                assert testfile.read() == "# coding: utf-8\n" + "\n".join(hist) + "\n"

            # Duplicate line numbers - check that it doesn't crash, and
            # gets a new session
            ip.history_manager.store_inputs(1, "rogue")
            ip.history_manager.writeout_cache()
            assert ip.history_manager.session_number == 3

            # Check that session and line values are not just max values
            sessid, lineno, entry = newhist[-1]
            assert lineno > 1
            ip.history_manager.reset()
            lineno = 1
            ip.history_manager.store_inputs(lineno, entry)
            gothist = ip.history_manager.search("*=*", unique=True)
            hist = list(gothist)[-1]
            assert sessid < hist[0]
            assert hist[1:] == (lineno, entry)
        finally:
            # Ensure saving thread is shut down before we try to clean up the files
            ip.history_manager.end_session()
            # Forcibly close database rather than relying on garbage collection
            ip.history_manager.save_thread.stop()
            ip.history_manager.db.close()
            # swap back
            ip.history_manager = hist_manager_ori


def test_extract_hist_ranges():
    instr = "1 2/3 ~4/5-6 ~4/7-~4/9 ~9/2-~7/5 ~10/"
    expected = [
        (0, 1, 2),  # 0 == current session
        (2, 3, 4),
        (-4, 5, 7),
        (-4, 7, 10),
        (-9, 2, None),  # None == to end
        (-8, 1, None),
        (-7, 1, 6),
        (-10, 1, None),
    ]
    actual = list(extract_hist_ranges(instr))
    assert actual == expected


def test_extract_hist_ranges_empty_str():
    instr = ""
    expected = [(0, 1, None)]  # 0 == current session, None == to end
    actual = list(extract_hist_ranges(instr))
    assert actual == expected


@pytest.mark.parametrize(
    "instr,expected,description",
    [
        ("21-23", [(0, 21, 24)], "Just line numbers"),
        ("25-", [(0, 25, None)], "Line start, no end"),
        ("~4/", [(-4, 1, None)], "with trailing slash"),
        # ("~4", [(-4, 1, None)], "without trailing slash"),
        ("~4/1-5", [(-4, 1, 6)], "with line ranges and trailing slash"),
        (
            "~4/1- ~5/ ~6/1-3",
            [(-4, 1, None), (-5, 1, None), (-6, 1, 4)],
            "multiple sessions with mixed syntax",
        ),
        ("~10/ ~20/", [(-10, 1, None), (-20, 1, None)], "larger session numbers"),
        # ("~1", [(-1, 1, None)], "single digit session without slash"),
        ("~1/", [(-1, 1, None)], "single digit session with slash"),
        # ("~2", [(-2, 1, None)], "backward compatibility without slash"),
        ("~2/", [(-2, 1, None)], "backward compatibility with slash"),
        ("4-", [(0, 4, None)], "from line 4 onward in current session"),
        ("~4/4-", [(-4, 4, None)], "from line 4 onward in session 4"),
        ("2/4-", [(2, 4, None)], "from line 4 onward in session 2"),
        ("~5/10-", [(-5, 10, None)], "from line 10 onward in session 5"),
    ],
)
def test_misc_extract_hist_ranges(instr, expected, description):
    actual = list(extract_hist_ranges(instr))
    assert (
        actual == expected
    ), f"Failed for '{instr}' ({description}): expected {expected}, got {actual}"


def test_magic_rerun():
    """Simple test for %rerun (no args -> rerun last line)"""
    ip = get_ipython()
    ip.run_cell("a = 10", store_history=True)
    ip.run_cell("a += 1", store_history=True)
    assert ip.user_ns["a"] == 11
    ip.run_cell("%rerun", store_history=True)
    assert ip.user_ns["a"] == 12


def test_timestamp_type():
    ip = get_ipython()
    info = ip.history_manager.get_session_info()
    assert isinstance(info[1], datetime)


def test_hist_file_config(hmmax3):
    cfg = Config()
    tfile = tempfile.NamedTemporaryFile(delete=False)
    cfg.HistoryManager.hist_file = Path(tfile.name)
    try:
        hm = HistoryManager(shell=get_ipython(), config=cfg)
        assert hm.hist_file == cfg.HistoryManager.hist_file
    finally:
        try:
            Path(tfile.name).unlink()
        except OSError:
            # same catch as in testing.tools.TempFileMixin
            # On Windows, even though we close the file, we still can't
            # delete it.  I have no clue why
            pass
            HistoryManager.__max_inst = 1


def test_histmanager_disabled(hmmax2):
    """Ensure that disabling the history manager doesn't create a database."""
    cfg = Config()
    cfg.HistoryAccessor.enabled = False

    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        hist_manager_ori = ip.history_manager
        hist_file = Path(tmpdir) / "history.sqlite"
        cfg.HistoryManager.hist_file = hist_file
        try:
            ip.history_manager = HistoryManager(shell=ip, config=cfg)
            hist = ["a=1", "def f():\n    test = 1\n    return test", "b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)
            assert ip.history_manager.input_hist_raw == [""] + hist
            ip.history_manager.reset()
            ip.history_manager.end_session()
        finally:
            ip.history_manager = hist_manager_ori

    # hist_file should not be created
    assert hist_file.exists() is False


def test_get_tail_session_awareness(hmmax3):
    """Test .get_tail() is:
        - session specific in HistoryManager
        - session agnostic in HistoryAccessor
    same for .get_last_session_id()
    """
    with TemporaryDirectory() as tmpdir:
        ip = get_ipython()
        tmp_path = Path(tmpdir)
        hist_file = tmp_path / "history.sqlite"
        get_source = lambda x: x[2]
        hm1 = None
        hm2 = None
        ha = None
        try:
            # hm1 creates a new session and adds history entries,
            # ha catches up
            hm1 = HistoryManager(shell=ip, hist_file=hist_file)
            hm1_last_sid = hm1.get_last_session_id
            ha = HistoryAccessor(hist_file=hist_file)
            ha_last_sid = ha.get_last_session_id

            hist1 = ["a=1", "b=1", "c=1"]
            for i, h in enumerate(hist1 + [""], start=1):
                hm1.store_inputs(i, h)
            assert list(map(get_source, hm1.get_tail())) == hist1
            assert list(map(get_source, ha.get_tail())) == hist1
            sid1 = hm1_last_sid()
            assert sid1 is not None
            assert ha_last_sid() == sid1

            # hm2 creates a new session and adds entries,
            # ha catches up
            hm2 = HistoryManager(shell=ip, hist_file=hist_file)
            hm2_last_sid = hm2.get_last_session_id

            hist2 = ["a=2", "b=2", "c=2"]
            for i, h in enumerate(hist2 + [""], start=1):
                hm2.store_inputs(i, h)
            tail = hm2.get_tail(n=3)
            assert list(map(get_source, tail)) == hist2
            tail = ha.get_tail(n=3)
            assert list(map(get_source, tail)) == hist2
            sid2 = hm2_last_sid()
            assert sid2 is not None
            assert ha_last_sid() == sid2
            assert sid2 != sid1

            # but hm1 still maintains its point of reference
            # and adding more entries to it doesn't change others
            # immediate perspective
            assert hm1_last_sid() == sid1
            tail = hm1.get_tail(n=3)
            assert list(map(get_source, tail)) == hist1

            hist3 = ["a=3", "b=3", "c=3"]
            for i, h in enumerate(hist3 + [""], start=5):
                hm1.store_inputs(i, h)
            tail = hm1.get_tail(n=7)
            assert list(map(get_source, tail)) == hist1 + [""] + hist3
            tail = hm2.get_tail(n=3)
            assert list(map(get_source, tail)) == hist2
            tail = ha.get_tail(n=3)
            assert list(map(get_source, tail)) == hist2
            assert hm1_last_sid() == sid1
            assert hm2_last_sid() == sid2
            assert ha_last_sid() == sid2
        finally:
            if hm1:
                hm1.save_thread.stop()
                hm1.db.close()
            if hm2:
                hm2.save_thread.stop()
                hm2.db.close()
            if ha:
                ha.db.close()


def test_calling_run_cell(hmmax2):
    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        tmp_path = Path(tmpdir)
        hist_manager_ori = ip.history_manager
        hist_file = tmp_path / "history_test_history1.sqlite"
        try:
            ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file)
            import time
    
            session_number = ip.history_manager.session_number
            ip.run_cell(raw_cell="get_ipython().run_cell(raw_cell='1', store_history=True)", store_history=True)
            while ip.history_manager.db_input_cache:
                time.sleep(0)
            new_session_number = ip.history_manager.session_number
        finally:
            # Ensure saving thread is shut down before we try to clean up the files
            ip.history_manager.end_session()
            # Forcibly close database rather than relying on garbage collection
            ip.history_manager.save_thread.stop()
            ip.history_manager.db.close()

            ip.history_manager = hist_manager_ori
    assert session_number == new_session_number, ValueError(f"{session_number} != {new_session_number}")