import os
import platform
import sys
from unittest.mock import Mock, patch

import pytest

from celery.utils.imports import (NotAPackage, cwd_in_path, find_module, gen_task_name, module_file, qualname,
                                  reload_from_cwd)


def test_find_module():
    def imp_side_effect(module):
        if module == 'foo':
            return None
        else:
            raise ImportError(module)

    assert find_module('celery')
    imp = Mock()
    imp.side_effect = imp_side_effect
    with pytest.raises(NotAPackage) as exc_info:
        find_module('foo.bar.baz', imp=imp)
    assert exc_info.value.args[0] == 'foo'
    assert find_module('celery.worker.request')


def test_find_module_legacy_namespace_package(tmp_path, monkeypatch):
    monkeypatch.chdir(str(tmp_path))
    (tmp_path / 'pkg' / 'foo').mkdir(parents=True)
    (tmp_path / 'pkg' / '__init__.py').write_text(
        'from pkgutil import extend_path\n'
        '__path__ = extend_path(__path__, __name__)\n')
    (tmp_path / 'pkg' / 'foo' / '__init__.py').write_text('')
    (tmp_path / 'pkg' / 'foo' / 'bar.py').write_text('')
    with patch.dict(sys.modules):
        for modname in list(sys.modules):
            if modname == 'pkg' or modname.startswith('pkg.'):
                del sys.modules[modname]
        with pytest.raises(ImportError):
            find_module('pkg.missing')
        with pytest.raises(ImportError):
            find_module('pkg.foo.missing')
        assert find_module('pkg.foo.bar')
        with pytest.raises(NotAPackage) as exc_info:
            find_module('pkg.foo.bar.missing')
        assert exc_info.value.args[0] == 'pkg.foo.bar'


def test_find_module_pep420_namespace_package(tmp_path, monkeypatch):
    monkeypatch.chdir(str(tmp_path))
    (tmp_path / 'pkg' / 'foo').mkdir(parents=True)
    (tmp_path / 'pkg' / 'foo' / '__init__.py').write_text('')
    (tmp_path / 'pkg' / 'foo' / 'bar.py').write_text('')
    with patch.dict(sys.modules):
        for modname in list(sys.modules):
            if modname == 'pkg' or modname.startswith('pkg.'):
                del sys.modules[modname]
        with pytest.raises(ImportError):
            find_module('pkg.missing')
        with pytest.raises(ImportError):
            find_module('pkg.foo.missing')
        assert find_module('pkg.foo.bar')
        with pytest.raises(NotAPackage) as exc_info:
            find_module('pkg.foo.bar.missing')
        assert exc_info.value.args[0] == 'pkg.foo.bar'


def test_qualname():
    Class = type('Fox', (object,), {
        '__module__': 'quick.brown',
    })
    assert qualname(Class) == 'quick.brown.Fox'
    assert qualname(Class()) == 'quick.brown.Fox'


def test_reload_from_cwd(patching):
    reload = patching('celery.utils.imports.reload')
    reload_from_cwd('foo')
    reload.assert_called()


def test_reload_from_cwd_custom_reloader():
    reload = Mock()
    reload_from_cwd('foo', reload)
    reload.assert_called()


def test_module_file():
    m1 = Mock()
    m1.__file__ = '/opt/foo/xyz.pyc'
    assert module_file(m1) == '/opt/foo/xyz.py'
    m2 = Mock()
    m2.__file__ = '/opt/foo/xyz.py'
    assert module_file(m1) == '/opt/foo/xyz.py'


def test_cwd_in_path(tmp_path, monkeypatch):
    now_cwd = os.getcwd()
    t = str(tmp_path) + "/foo"
    os.mkdir(t)
    os.chdir(t)
    with cwd_in_path():
        assert os.path.exists(t) is True

    if sys.platform == "win32" or "Windows" in platform.platform():
        # If it is a Windows server, other processes cannot delete the current working directory being used by celery
        # . If you want to delete it, you need to terminate the celery process. If it is a Linux server, the current
        # working directory of celery can be deleted by other processes.
        pass
    else:
        os.rmdir(t)
        with cwd_in_path():
            assert os.path.exists(t) is False
    os.chdir(now_cwd)


class test_gen_task_name:

    def test_no_module(self):
        app = Mock()
        app.name == '__main__'
        assert gen_task_name(app, 'foo', 'axsadaewe')