import mock
import pytest
import time
from django.contrib.auth.models import Group
from django.core.exceptions import ValidationError

from addons.github.tests import factories
from addons.osfstorage.models import OsfStorageFile
from framework.auth import Auth
from django.contrib.auth.models import AnonymousUser
from django.contrib.contenttypes.models import ContentType
from framework.exceptions import PermissionsError
from osf.models import OSFGroup, Node, OSFUser, OSFGroupLog, NodeLog
from osf.utils.permissions import MANAGER, MEMBER, MANAGE, READ, WRITE, ADMIN
from website.notifications.utils import get_all_node_subscriptions
from website.osf_groups import signals as group_signals
from .factories import (
    NodeFactory,
    ProjectFactory,
    AuthUserFactory,
    OSFGroupFactory
)

pytestmark = pytest.mark.django_db

@pytest.fixture()
def manager():
    return AuthUserFactory()

@pytest.fixture()
def member():
    return AuthUserFactory()

@pytest.fixture()
def user():
    return AuthUserFactory()

@pytest.fixture()
def user_two():
    return AuthUserFactory()

@pytest.fixture()
def user_three():
    return AuthUserFactory()

@pytest.fixture()
def auth(manager):
    return Auth(manager)

@pytest.fixture()
def project(manager):
    return ProjectFactory(creator=manager)

@pytest.fixture()
def osf_group(manager, member):
    osf_group = OSFGroupFactory(creator=manager)
    osf_group.make_member(member)
    return osf_group

class TestOSFGroup:

    def test_osf_group_creation(self, manager, member, user_two, fake):
        osf_group = OSFGroup.objects.create(name=fake.bs(), creator=manager)
        # OSFGroup creator given manage permissions
        assert osf_group.has_permission(manager, MANAGE) is True
        assert osf_group.has_permission(user_two, MANAGE) is False

        assert manager in osf_group.managers
        assert manager in osf_group.members
        assert manager not in osf_group.members_only

        user_two.is_superuser = True
        user_two.save()

        # Superusers don't have permission to group
        assert osf_group.has_permission(user_two, MEMBER) is False

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_make_manager(self, mock_send_mail, manager, member, user_two, user_three, osf_group):
        # no permissions
        with pytest.raises(PermissionsError):
            osf_group.make_manager(user_two, Auth(user_three))

        # member only
        with pytest.raises(PermissionsError):
            osf_group.make_manager(user_two, Auth(member))

        # manage permissions
        osf_group.make_manager(user_two, Auth(manager))
        assert osf_group.has_permission(user_two, MANAGE) is True
        assert user_two in osf_group.managers
        assert user_two in osf_group.members
        assert mock_send_mail.call_count == 1

        # upgrade to manager
        osf_group.make_manager(member, Auth(manager))
        assert osf_group.has_permission(member, MANAGE) is True
        assert member in osf_group.managers
        assert member in osf_group.members
        # upgrading an existing member does not re-send an email
        assert mock_send_mail.call_count == 1

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_make_member(self, mock_send_mail, manager, member, user_two, user_three, osf_group):
        # no permissions
        with pytest.raises(PermissionsError):
            osf_group.make_member(user_two, Auth(user_three))

        # member only
        with pytest.raises(PermissionsError):
            osf_group.make_member(user_two, Auth(member))

        # manage permissions
        osf_group.make_member(user_two, Auth(manager))
        assert osf_group.has_permission(user_two, MANAGE) is False
        assert user_two not in osf_group.managers
        assert user_two in osf_group.members
        assert mock_send_mail.call_count == 1

        # downgrade to member, sole manager
        with pytest.raises(ValueError):
            osf_group.make_member(manager, Auth(manager))

        # downgrade to member
        osf_group.make_manager(user_two, Auth(manager))
        assert user_two in osf_group.managers
        assert user_two in osf_group.members
        osf_group.make_member(user_two, Auth(manager))
        assert user_two not in osf_group.managers
        assert user_two in osf_group.members
        assert mock_send_mail.call_count == 1

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_add_unregistered_member(self, mock_send_mail, manager, member, osf_group, user_two):
        test_fullname = 'Test User'
        test_email = 'test_member@cos.io'
        test_manager_email = 'test_manager@cos.io'

        # Email already exists
        with pytest.raises(ValueError):
            osf_group.add_unregistered_member(test_fullname, user_two.username, auth=Auth(manager))

        # Test need manager perms to add
        with pytest.raises(PermissionsError):
            osf_group.add_unregistered_member(test_fullname, test_email, auth=Auth(member))

        # Add member
        osf_group.add_unregistered_member(test_fullname, test_email, auth=Auth(manager))
        assert mock_send_mail.call_count == 1
        unreg_user = OSFUser.objects.get(username=test_email)
        assert unreg_user in osf_group.members
        assert unreg_user not in osf_group.managers
        assert osf_group.has_permission(unreg_user, MEMBER) is True
        assert osf_group._id in unreg_user.unclaimed_records

        # Attempt to add unreg user as a member
        with pytest.raises(ValueError):
            osf_group.add_unregistered_member(test_fullname, test_email, auth=Auth(manager))

        # Add unregistered manager
        osf_group.add_unregistered_member(test_fullname, test_manager_email, auth=Auth(manager), role=MANAGER)
        assert mock_send_mail.call_count == 2
        unreg_manager = OSFUser.objects.get(username=test_manager_email)
        assert unreg_manager in osf_group.members
        assert unreg_manager in osf_group.managers
        assert osf_group.has_permission(unreg_manager, MEMBER) is True
        assert osf_group._id in unreg_manager.unclaimed_records

        # Add unregistered member with blacklisted email
        with pytest.raises(ValidationError):
            osf_group.add_unregistered_member(test_fullname, 'test@example.com', auth=Auth(manager), role=MANAGER)

    def test_remove_member(self, manager, member, user_three, osf_group):
        new_member = AuthUserFactory()
        osf_group.make_member(new_member)
        assert new_member not in osf_group.managers
        assert new_member in osf_group.members

        # no permissions
        with pytest.raises(PermissionsError):
            osf_group.remove_member(new_member, Auth(user_three))

        # member only
        with pytest.raises(PermissionsError):
            osf_group.remove_member(new_member, Auth(member))

        # manage permissions
        osf_group.remove_member(new_member, Auth(manager))
        assert new_member not in osf_group.managers
        assert new_member not in osf_group.members

        # Remove self - member can remove themselves
        osf_group.remove_member(member, Auth(member))
        assert member not in osf_group.managers
        assert member not in osf_group.members

    def test_remove_manager(self, manager, member, user_three, osf_group):
        new_manager = AuthUserFactory()
        osf_group.make_manager(new_manager)
        # no permissions
        with pytest.raises(PermissionsError):
            osf_group.remove_member(new_manager, Auth(user_three))

        # member only
        with pytest.raises(PermissionsError):
            osf_group.remove_member(new_manager, Auth(member))

        # manage permissions
        osf_group.remove_member(new_manager, Auth(manager))
        assert new_manager not in osf_group.managers
        assert new_manager not in osf_group.members

        # can't remove last manager
        with pytest.raises(ValueError):
            osf_group.remove_member(manager, Auth(manager))
        assert manager in osf_group.managers
        assert manager in osf_group.members

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_notify_group_member_email_does_not_send_before_throttle_expires(self, mock_send_mail, manager, osf_group):
        member = AuthUserFactory()
        assert member.member_added_email_records == {}
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager))
        assert mock_send_mail.call_count == 1

        record = member.member_added_email_records[osf_group._id]
        assert record is not None
        # 2nd call does not send email because throttle period has not expired
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager))
        assert member.member_added_email_records[osf_group._id] == record
        assert mock_send_mail.call_count == 1

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_notify_group_member_email_sends_after_throttle_expires(self, mock_send_mail, osf_group, member, manager):
        throttle = 0.5

        member = AuthUserFactory()
        assert member.member_added_email_records == {}
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager), throttle=throttle)
        assert mock_send_mail.call_count == 1

        time.sleep(1)  # throttle period expires
        # 2nd call does not send email because throttle period has not expired
        assert member.member_added_email_records[osf_group._id] is not None
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager), throttle=throttle)
        assert mock_send_mail.call_count == 2

    @mock.patch('website.osf_groups.views.mails.send_mail')
    def test_notify_group_unregistered_member_throttle(self, mock_send_mail, osf_group, member, manager):
        throttle = 0.5

        member = AuthUserFactory()
        member.is_registered = False
        member.add_unclaimed_record(osf_group, referrer=manager, given_name='grapes mcgee', email='grapes@cos.io')
        member.save()
        assert member.member_added_email_records == {}
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager), throttle=throttle)
        assert mock_send_mail.call_count == 1

        assert member.member_added_email_records[osf_group._id] is not None
        # 2nd call does not send email because throttle period has not expired
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager))
        assert mock_send_mail.call_count == 1

        time.sleep(1)  # throttle period expires
        # 2nd call does not send email because throttle period has not expired
        assert member.member_added_email_records[osf_group._id] is not None
        group_signals.member_added.send(osf_group, user=member, permission=WRITE, auth=Auth(manager), throttle=throttle)
        assert mock_send_mail.call_count == 2

    def test_rename_osf_group(self, manager, member, user_two, osf_group):
        new_name = 'Platform Team'
        # no permissions
        with pytest.raises(PermissionsError):
            osf_group.set_group_name(new_name, Auth(user_two))

        # member only
        with pytest.raises(PermissionsError):
            osf_group.set_group_name(new_name, Auth(member))

        # manage permissions
        osf_group.set_group_name(new_name, Auth(manager))
        osf_group.save()

        assert osf_group.name == new_name

    def test_remove_group(self, manager, member, osf_group):
        osf_group_name = osf_group.name
        manager_group_name = osf_group.manager_group.name
        member_group_name = osf_group.member_group.name

        osf_group.remove_group(Auth(manager))
        assert not OSFGroup.objects.filter(name=osf_group_name).exists()
        assert not Group.objects.filter(name=manager_group_name).exists()
        assert not Group.objects.filter(name=member_group_name).exists()

        assert manager_group_name not in manager.groups.values_list('name', flat=True)

    def test_remove_group_node_perms(self, manager, member, osf_group, project):
        project.add_osf_group(osf_group, ADMIN)
        assert project.has_permission(member, ADMIN) is True

        osf_group.remove_group(Auth(manager))

        assert project.has_permission(member, ADMIN) is False

    def test_user_groups_property(self, manager, member, osf_group):
        assert osf_group in manager.osf_groups
        assert osf_group in member.osf_groups

        other_group = OSFGroupFactory()

        assert other_group not in manager.osf_groups
        assert other_group not in member.osf_groups

    def test_user_group_roles(self, manager, member, user_three, osf_group):
        assert manager.group_role(osf_group) == MANAGER
        assert member.group_role(osf_group) == MEMBER
        assert user_three.group_role(osf_group) is None

    def test_replace_contributor(self, manager, member, osf_group):
        user = osf_group.add_unregistered_member('test_user', 'test@cos.io', auth=Auth(manager))
        assert user in osf_group.members
        assert user not in osf_group.managers
        assert (
            osf_group._id in
            user.unclaimed_records.keys()
        )
        osf_group.replace_contributor(user, member)
        assert user not in osf_group.members
        assert user not in osf_group.managers
        assert osf_group.has_permission(member, MEMBER) is True
        assert osf_group.has_permission(user, MEMBER) is False

        # test unclaimed_records is removed
        assert (
            osf_group._id not in
            user.unclaimed_records.keys()
        )

    def test_get_users_with_perm_osf_groups(self, project, manager, member, osf_group):
        # Explicitly added as a contributor
        read_users = project.get_users_with_perm(READ)
        write_users = project.get_users_with_perm(WRITE)
        admin_users = project.get_users_with_perm(ADMIN)
        assert len(project.get_users_with_perm(READ)) == 1
        assert len(project.get_users_with_perm(WRITE)) == 1
        assert len(project.get_users_with_perm(ADMIN)) == 1
        assert manager in read_users
        assert manager in write_users
        assert manager in admin_users

        # Added through osf groups
        project.add_osf_group(osf_group, WRITE)
        read_users = project.get_users_with_perm(READ)