# Copied and adapted from tendo
from __future__ import annotations

import logging
import os
import sys
import tempfile
import unittest
from multiprocessing import Process

from angr.utils.mp import Initializer

logger = logging.getLogger(name=__name__)


if sys.platform != "win32":
    import fcntl


class SingleInstanceException(BaseException):
    pass


class SingleInstance:
    """Class that can be instantiated only once per machine.

    If you want to prevent your script from running in parallel just instantiate SingleInstance() class. If is there
    another instance already running it will throw a `SingleInstanceException`.

    >>> me = SingleInstance()

    This option is very useful if you have scripts executed by crontab at small amounts of time.

    Remember that this works by creating a lock file with a filename based on the full path to the script file.

    Providing a flavor_id will augment the filename with the provided flavor_id, allowing you to create multiple
    singleton instances from the same file. This is particularly useful if you want specific functions to have their
    own singleton instances.
    """

    def __init__(self, flavor_id: str = "", lockfile: str = "") -> None:
        self.initialized = False
        if lockfile:
            self.lockfile = lockfile
        else:
            basename = (
                os.path.splitext(os.path.abspath(__file__))[0].replace("/", "-").replace(":", "").replace("\\", "-")
                + f"-{flavor_id}"
                + ".lock"
            )
            self.lockfile = os.path.normpath(tempfile.gettempdir() + "/" + basename)

        logger.debug("SingleInstance lockfile: %s", self.lockfile)
        if sys.platform == "win32":
            try:
                # file already exists, we try to remove (in case previous
                # execution was interrupted)
                if os.path.exists(self.lockfile):
                    os.unlink(self.lockfile)
                self.fd = os.open(self.lockfile, os.O_CREAT | os.O_EXCL | os.O_RDWR)
            except OSError as ex:
                type, e, tb = sys.exc_info()
                if e.errno == 13:
                    logger.debug("Another instance is already running, quitting.")
                    raise SingleInstanceException from ex
                print(e.errno)
                raise
        else:  # non Windows
            self.fp = open(self.lockfile, "w")  # noqa: SIM115
            self.fp.flush()
            try:
                fcntl.lockf(self.fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except OSError as ex:
                logger.debug("Another instance is already running, quitting.")
                raise SingleInstanceException from ex
        self.initialized = True

    def __del__(self) -> None:
        if not self.initialized:
            return
        try:
            if sys.platform == "win32":
                if hasattr(self, "fd"):
                    os.close(self.fd)
                    os.unlink(self.lockfile)
            else:
                fcntl.lockf(self.fp, fcntl.LOCK_UN)
                # os.close(self.fp)
                if os.path.isfile(self.lockfile):
                    os.unlink(self.lockfile)
        except Exception as e:
            if logger:
                logger.debug(e)
            else:
                print(f"Unloggable error: {e}")
            sys.exit(-1)


def f(initializer: Initializer, name: str) -> None:
    initializer.initialize()
    tmp = logger.level
    logger.setLevel(logging.CRITICAL)  # we do not want to see the warning
    try:
        me2 = SingleInstance(flavor_id=name)  # noqa
    except SingleInstanceException:
        sys.exit(-1)
    logger.setLevel(tmp)


class testSingleton(unittest.TestCase):
    def test_1(self) -> None:
        me = SingleInstance(flavor_id="test-1")
        del me  # now the lock should be removed
        assert True

    def test_2(self) -> None:
        p = Process(
            target=f,
            args=(
                Initializer.get(),
                "test-2",
            ),
        )
        p.start()
        p.join()
        # the called function should succeed
        assert p.exitcode == 0, f"{p.exitcode} != 0"

    def test_3(self) -> None:
        me = SingleInstance(flavor_id="test-3")  # noqa -- me should still kept
        p = Process(
            target=f,
            args=(
                Initializer.get(),
                "test-3",
            ),
        )
        p.start()
        p.join()
        # the called function should fail because we already have another
        # instance running
        assert p.exitcode != 0, f"{p.exitcode} != 0 (2nd execution)"
        # note, we return -1 but this translates to 255 meanwhile we'll
        # consider that anything different from 0 is good
        p = Process(
            target=f,
            args=(
                Initializer.get(),
                "test-3",
            ),
        )
        p.start()
        p.join()
        # the called function should fail because we already have another
        # instance running
        assert p.exitcode != 0, f"{p.exitcode} != 0 (3rd execution)"

    def test_4(self) -> None:
        lockfile = "/tmp/foo.lock"
        me = SingleInstance(lockfile=lockfile)
        assert me.lockfile == lockfile


if __name__ == "__main__":
    logger.addHandler(logging.StreamHandler())
    logger.setLevel(logging.DEBUG)
    unittest.main()
