import asyncio
import os
import time
from concurrent.futures.thread import ThreadPoolExecutor
from pathlib import Path
from typing import Sequence

import pytest
import wait_for2

from clarabot_nano.server.fstx.filesystem.operation.base import AbstractIOQueueOperation
from clarabot_nano.server.fstx.filesystem.operation.copy import CopyOperation
from clarabot_nano.server.fstx.filesystem.operation.delete import DeleteOperation
from clarabot_nano.server.fstx.filesystem.operation.mkdir import MkdirOperation
from clarabot_nano.server.fstx.filesystem.operation.rename import RenameOperation
from clarabot_nano.server.fstx.filesystem.operation.replace import ReplaceOperation
from clarabot_nano.server.fstx.filesystem.operation.write import WriteTruncateOperation, SeekWriteOperation
from clarabot_nano.server.fstx.filesystem.tx_queue import FilesystemTxQueue
from clarabot_nano.utility.path.validate import normalize_absolute_path
from clarabot_nano_tests.testing.file import temp_path


class _TestingFilesystemTxQueue(FilesystemTxQueue):
    def empty(self):
        return (
            not self._locks
            and not self._ops_by_account
            and not self._tracker._affects_upstream
            and not self._tracker._affects_downstream
        )


class _TestingOperation(AbstractIOQueueOperation):
    OPERATION_KEY = "_TEST"

    def __init__(self, account_id: int, path: str, cycles: int):
        AbstractIOQueueOperation.__init__(self, account_id)
        self.path = path
        self.cycles = cycles

    def paths(self) -> Sequence[str]:
        return (self.path,)

    def execute(self):
        for _ in range(self.cycles):
            time.sleep(1)


@pytest.fixture
async def fstx_queue():
    with ThreadPoolExecutor(thread_name_prefix="TestP-") as pool:
        tx_queue = _TestingFilesystemTxQueue(pool)
        try:
            yield tx_queue
        finally:
            await tx_queue.close()


@pytest.mark.asyncio
async def test_server_fstx_queue_ops(tmpdir, fstx_queue: _TestingFilesystemTxQueue):
    r = Path(normalize_absolute_path(temp_path(tmpdir)))
    r.mkdir()

    # mkdir
    dir_path0 = r / "dir0"
    assert not dir_path0.exists()
    assert fstx_queue.empty()
    async with fstx_queue.wait_ready(MkdirOperation(0, dir_path0.as_posix())):
        assert not fstx_queue.empty()
        assert not dir_path0.exists()  # test not submitting
    assert not fstx_queue.empty()
    await asyncio.sleep(0.05)
    assert fstx_queue.empty()
    assert not dir_path0.exists()
    async with fstx_queue.wait_ready(MkdirOperation(0, dir_path0.as_posix())) as submitter:
        assert not fstx_queue.empty()
        assert not dir_path0.exists()
        await submitter()
        assert fstx_queue.empty()
        assert dir_path0.is_dir()
    assert fstx_queue.empty()
    assert dir_path0.is_dir()
    async with fstx_queue.wait_ready(MkdirOperation(0, dir_path0.as_posix())) as submitter:
        assert not fstx_queue.empty()
        assert dir_path0.is_dir()
        with pytest.raises(FileExistsError):
            await submitter()
        assert fstx_queue.empty()
        assert dir_path0.is_dir()
    assert fstx_queue.empty()
    assert dir_path0.is_dir()

    assert fstx_queue.empty()
    t_path0 = r / "temp0"
    f_path0 = r / "file0"
    content0 = os.urandom(17)
    content1 = os.urandom(15)
    content2 = os.urandom(13)
    # write
    assert not t_path0.exists()
    assert not f_path0.exists()
    async with fstx_queue.wait_ready(
        WriteTruncateOperation(0, t_path0.as_posix(), f_path0.as_posix(), content0)
    ) as submitter:
        await submitter()
    assert t_path0.is_file()
    assert not f_path0.exists()
    assert t_path0.read_bytes() == content0
    async with fstx_queue.wait_ready(
        SeekWriteOperation(0, t_path0.as_posix(), f_path0.as_posix(), content1, len(content0))
    ) as submitter:
        await submitter()
    assert t_path0.is_file()
    assert not f_path0.exists()
    assert t_path0.read_bytes() == content0 + content1
    async with fstx_queue.wait_ready(
        SeekWriteOperation(0, t_path0.as_posix(), f_path0.as_posix(), content2, len(content0))
    ) as submitter:
        await submitter()
    assert t_path0.is_file()
    assert not f_path0.exists()
    assert t_path0.read_bytes() == content0 + content2 + content1[13:]
    async with fstx_queue.wait_ready(
        WriteTruncateOperation(0, t_path0.as_posix(), f_path0.as_posix(), content0 + content1)
    ) as submitter:
        await submitter()
    assert t_path0.is_file()
    assert not f_path0.exists()
    assert t_path0.read_bytes() == content0 + content1

    assert fstx_queue.empty()
    # replace
    async with fstx_queue.wait_ready(ReplaceOperation(0, t_path0.as_posix(), f_path0.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert f_path0.is_file()
    assert f_path0.read_bytes() == content0 + content1

    assert fstx_queue.empty()
    # rename
    async with fstx_queue.wait_ready(RenameOperation(0, f_path0.as_posix(), dir_path0.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert not f_path0.exists()
    dirf_path0 = dir_path0 / f_path0.name
    assert dirf_path0.is_file()
    dirf_path1 = dir_path0 / "asd0"
    async with fstx_queue.wait_ready(RenameOperation(0, dirf_path0.as_posix(), dirf_path1.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert not f_path0.exists()
    assert not dirf_path0.exists()
    assert dirf_path1.is_file()
    dir_path1 = r / "dir1"
    async with fstx_queue.wait_ready(RenameOperation(0, dir_path0.as_posix(), dir_path1.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert not f_path0.exists()
    assert not dirf_path0.exists()
    assert not dirf_path1.exists()
    dirf_path2 = dir_path1 / dirf_path1.name
    assert dirf_path2.is_file()

    assert fstx_queue.empty()
    # copy
    async with fstx_queue.wait_ready(CopyOperation(0, dirf_path2.as_posix(), f_path0.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert f_path0.is_file()
    assert not dirf_path0.exists()
    assert not dirf_path1.exists()
    assert dirf_path2.is_file()
    async with fstx_queue.wait_ready(CopyOperation(0, dir_path1.as_posix(), dir_path0.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert f_path0.is_file()
    assert not dirf_path0.exists()
    assert dirf_path1.is_file()
    assert dirf_path2.is_file()

    assert fstx_queue.empty()
    # delete
    async with fstx_queue.wait_ready(DeleteOperation(0, dir_path1.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert f_path0.is_file()
    assert not dirf_path0.exists()
    assert not dir_path1.exists()
    assert dirf_path1.is_file()
    assert not dirf_path2.exists()
    async with fstx_queue.wait_ready(DeleteOperation(0, f_path0.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert not f_path0.exists()
    assert not dirf_path0.exists()
    assert not dir_path1.exists()
    assert dirf_path1.is_file()
    assert not dirf_path2.exists()
    async with fstx_queue.wait_ready(DeleteOperation(0, dirf_path1.as_posix())) as submitter:
        await submitter()
    assert not t_path0.exists()
    assert not f_path0.exists()
    assert not dirf_path0.exists()
    assert not dir_path1.exists()
    assert not dirf_path1.exists()
    assert not dirf_path2.exists()
    assert dir_path0.is_dir()

    assert fstx_queue.empty()


class _TestingSeekWriteOperation(SeekWriteOperation):
    def execute(self):
        SeekWriteOperation.execute(self)
        with open(self.path, "rb") as f:
            return f.read()


@pytest.mark.asyncio
async def test_server_fstx_queue_locking(tmpdir, fstx_queue: _TestingFilesystemTxQueue):
    r = Path(normalize_absolute_path(temp_path(tmpdir)))
    r.mkdir()
    t_path = r / "tmp"
    f_path = r / "fin"

    test_init_future = asyncio.get_running_loop().create_future()
    t_path.touch()  # file must exist for parallel writes

    parts = [os.urandom(19) for _ in range(7)]
    reads = []

    async def parallel_writes(account, t_path_, f_path_, offset_, content):
        async with fstx_queue.wait_ready(
            _TestingSeekWriteOperation(account, t_path_.as_posix(), f_path_.as_posix(), content, offset_)
        ) as submitter:
            await test_init_future  # make all wait and set lock-wait
            reads.append(await submitter())

    tasks = []
    offset = 0
    for i, part in enumerate(parts):
        tasks.append(asyncio.create_task(parallel_writes(i, t_path, f_path, offset, part)))
        offset += len(part)

    test_init_future.set_result(True)
    await asyncio.gather(*tasks)
    assert fstx_queue.empty()

    assert t_path.read_bytes() == b"".join(parts)
    assert len(reads) == len(parts)
    assert reads[0] == parts[0]
    assert reads[-1] == b"".join(parts)
    assert len(set(reads)) == len(parts)


@pytest.mark.asyncio
async def test_server_fstx_queue_timeout_cancel(fstx_queue: _TestingFilesystemTxQueue):
    async with fstx_queue.wait_ready(_TestingOperation(0, "asdf", 4)) as submitter:
        try:
            await wait_for2.wait_for(submitter(), 1.5)
        except asyncio.TimeoutError:
            pass
    assert not fstx_queue.empty()


@pytest.mark.asyncio
async def test_server_fstx_queue_errors(tmpdir, fstx_queue: _TestingFilesystemTxQueue):
    r = Path(normalize_absolute_path(temp_path(tmpdir)))
    r.mkdir()
    path = r / "tmp"
    path2 = r / "tmp2"
    path.mkdir()

    op = DeleteOperation(0, path.as_posix())
    op2 = DeleteOperation(0, path2.as_posix())

    try:
        async with fstx_queue.wait_ready(op, paths=op.paths()):
            pass
    except RuntimeError as e:
        assert e.args[0] == "Do not provide operation and paths both."
    else:
        assert False
    try:
        async with fstx_queue.wait_ready():
            pass
    except AttributeError as e:
        assert e.args[0] == "'NoneType' object has no attribute 'paths'"
    else:
        assert False

    async with fstx_queue.wait_ready(paths=op2.paths()) as submitter:
        with pytest.raises(RuntimeError, match=r"Provide an operation at submit if the preparation has none\."):
            submitter()
        with pytest.raises(RuntimeError, match=r"Fs operation paths do not match prepared\."):
            submitter(op)

    async with fstx_queue.wait_ready(op) as submitter:
        with pytest.raises(RuntimeError, match=r"Do not provide an operation if the preparation has one\."):
            submitter(op)
        await submitter()
        with pytest.raises(RuntimeError, match=r"Fs operation already submitted\."):
            submitter()

    assert fstx_queue.empty()
