# Copyright 2020 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License")
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


"""Unit tests for the permissions module.

Verifies the access control logic across different user roles
(admin, editor, owner, etc.) and feature visibility states.
"""

import werkzeug.exceptions  # Flask HTTP stuff.

import testing_config  # Must be imported before the module under test.

# from google.appengine.api import users
from framework import basehandlers, permissions, users
from framework.basehandlers import Route
from internals import core_models, user_models


class MockHandler(basehandlers.BaseHandler):
    """A mock handler used for testing permissions logic."""

    def __init__(self):
        """Initialize the mock handler."""
        self.called_with = None
        self.common_data = {}

    def get_common_data(self):
        """Get common data."""
        return self.common_data

    @permissions.require_admin_site
    def do_get(self, *args):
        """Handle GET requests."""
        self.called_with = args
        return {'message': 'did get'}

    @permissions.require_admin_site
    def do_post(self, *args):
        """Handle POST requests."""
        self.called_with = args
        return {'message': 'did post'}


test_app = basehandlers.FlaskApplication(
    __name__,
    [
        Route('/path', MockHandler),
    ],
    debug=True,
)


class PermissionFunctionTests(testing_config.CustomTestCase):
    """Tests for standalone permission functions."""

    def setUp(self):
        """Set up the test environment."""
        self.users = []
        self.app_user = user_models.AppUser(email='registered@example.com')
        self.app_user.put()
        self.users.append(self.app_user)

        self.app_admin = user_models.AppUser(email='admin@example.com')
        self.app_admin.is_admin = True
        self.app_admin.put()
        self.users.append(self.app_admin)

        self.app_editor = user_models.AppUser(email='editor@example.com')
        self.app_editor.is_site_editor = True
        self.app_editor.put()
        self.users.append(self.app_editor)

        self.feature_owner = user_models.AppUser(
            email='feature_owner@example.com'
        )
        self.feature_owner.put()
        self.users.append(self.feature_owner)

        self.feature_editor = user_models.AppUser(
            email='feature_editor@example.com'
        )  # noqa: E501
        self.feature_editor.put()
        self.users.append(self.feature_editor)

        self.spec_mentor = user_models.AppUser(email='mentor@example.com')
        self.spec_mentor.put()
        self.users.append(self.spec_mentor)

        # Feature for checking permissions against
        self.feature_1 = core_models.FeatureEntry(
            name='feature one',
            summary='sum',
            creator_email='feature_creator@example.com',
            owner_emails=['feature_owner@example.com'],
            editor_emails=['feature_editor@example.com'],
            spec_mentor_emails=['mentor@example.com'],
            category=1,
        )
        self.feature_1.put()
        self.feature_id = self.feature_1.key.integer_id()

        # Feature for checking permissions against
        self.feature_confidential = core_models.FeatureEntry(
            name='feature one',
            summary='sum',
            creator_email='feature_creator@example.com',
            owner_emails=['feature_owner@example.com'],
            editor_emails=['feature_editor@example.com'],
            spec_mentor_emails=['mentor@example.com'],
            category=1,
            confidential=True,
        )
        self.feature_confidential.put()
        self.feature_confidential_id = (
            self.feature_confidential.key.integer_id()
        )

    def tearDown(self):
        """Clean up the test environment."""
        for user in self.users:
            user.delete()
        self.feature_1.key.delete()
        self.feature_confidential.key.delete()

    def check_function_results(
        self,
        func,
        additional_args,
        unregistered='missing',
        registered='missing',
        special='missing',
        site_editor='missing',
        admin='missing',
        anon='missing',
    ):
        """Test func under six conditions and check expected results."""
        # Test unregistered users
        testing_config.sign_in('unregistered@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(unregistered, func(user, *additional_args))

        # Test registered users
        testing_config.sign_in('registered@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(registered, func(user, *additional_args))

        # Test special users (registered orgs)
        for domain in permissions.REGISTERED_USER_ORGANIZATIONS:
            testing_config.sign_in(f'user{domain}', 123)
            user = users.get_current_user()
            self.assertEqual(special, func(user, *additional_args))

        # Test site editor user
        testing_config.sign_in('editor@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(site_editor, func(user, *additional_args))

        # Test admin users
        testing_config.sign_in('admin@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(admin, func(user, *additional_args))

        # Test anonymous visitors
        testing_config.sign_out()
        user = users.get_current_user()
        self.assertEqual(anon, func(user, *additional_args))

    def check_function_results_with_feature(
        self,
        func,
        additional_args,
        unregistered='missing',
        registered='missing',
        feature_owner='missing',
        feature_editor='missing',
        creator='missing',
        site_editor='missing',
        admin='missing',
        spec_mentor='missing',
    ):
        """Test func in the context of a specific feature id."""
        # Test unregistered users
        testing_config.sign_in('unregistered@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(unregistered, func(user, *additional_args))

        # Test registered users
        testing_config.sign_in('registered@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(registered, func(user, *additional_args))

        # Test feature owners
        testing_config.sign_in('feature_owner@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(feature_owner, func(user, *additional_args))

        # Test feature editors
        testing_config.sign_in('feature_editor@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(feature_editor, func(user, *additional_args))

        # Test feature editors
        testing_config.sign_in('feature_creator@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(creator, func(user, *additional_args))

        # Test site editor user
        testing_config.sign_in('editor@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(site_editor, func(user, *additional_args))

        # Test admin users
        testing_config.sign_in('admin@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(admin, func(user, *additional_args))

        # Test spec mentor users
        testing_config.sign_in('mentor@example.com', 123)
        user = users.get_current_user()
        self.assertEqual(spec_mentor, func(user, *additional_args))

    def test_can_admin_site(self):
        """Test can admin site."""
        self.check_function_results(
            permissions.can_admin_site,
            tuple(),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=False,
            admin=True,
            anon=False,
        )

    def test_can_view_feature(self):
        """Test can view feature."""
        self.check_function_results(
            permissions.can_view_feature,
            (None,),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=False,
            admin=False,
            anon=False,
        )

        self.check_function_results_with_feature(
            permissions.can_view_feature,
            (self.feature_1,),
            unregistered=True,
            registered=True,
            feature_owner=True,
            feature_editor=True,
            creator=True,
            site_editor=True,
            admin=True,
            spec_mentor=True,
        )

        self.check_function_results_with_feature(
            permissions.can_view_feature,
            (self.feature_confidential,),
            unregistered=False,
            registered=False,
            feature_owner=True,
            feature_editor=True,
            creator=True,
            site_editor=False,
            admin=True,
            spec_mentor=False,
        )

    def test_can_create_feature(self):
        """Test can create feature."""
        self.check_function_results(
            permissions.can_create_feature,
            tuple(),
            unregistered=False,
            registered=True,
            special=True,
            site_editor=True,
            admin=True,
            anon=False,
        )

    def test_can_create_feature__registered_orgs(self):
        """Users from any registered org domain can create features."""
        for domain in permissions.REGISTERED_USER_ORGANIZATIONS:
            testing_config.sign_in(f'user{domain}', 123)
            user = users.get_current_user()
            self.assertTrue(permissions.can_create_feature(user))

    def test_can_create_feature__unregistered_orgs(self):
        """Unregistered users from non-whitelisted orgs cannot create features."""
        unregistered_emails = [
            'user@example.com',
            'user@otherdomain.com',
            'user@fakechromium.org.evil.com',
        ]
        for email in unregistered_emails:
            testing_config.sign_in(email, 123)
            user = users.get_current_user()
            self.assertFalse(permissions.can_create_feature(user))

    def test_can_edit_any_feature(self):
        """Test can edit any feature."""
        self.check_function_results(
            permissions.can_edit_any_feature,
            tuple(),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=True,
            admin=True,
            anon=False,
        )

    def test_can_review_release_notes__anon_other_site_admin(self):
        """Test can review release notes with anon other site admin."""
        self.check_function_results(
            permissions.can_review_release_notes,
            tuple(),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=False,
            admin=True,
            anon=False,
        )

    def test_can_review_release_notes__allow_list(self):
        """Test can review release notes with allow list."""
        testing_config.sign_in(permissions.RELEASE_NOTE_REVIEWERS[0], 123)
        user = users.get_current_user()
        self.assertTrue(permissions.can_review_release_notes(user))

    def test_can_edit_feature(self):
        """Test can edit feature."""
        self.check_function_results(
            permissions.can_edit_feature,
            (None,),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=True,
            admin=True,
            anon=False,
        )

        # Check in context of specific feature.
        self.check_function_results_with_feature(
            permissions.can_edit_feature,
            (self.feature_1,),
            unregistered=False,
            registered=False,
            feature_owner=True,
            feature_editor=True,
            creator=True,
            site_editor=True,
            admin=True,
            spec_mentor=True,
        )

    def test_can_review_gate(self):
        """Test can review gate."""
        approvers = []
        self.check_function_results(
            permissions.can_review_gate,
            (self.feature_1, None, approvers),
            unregistered=False,
            registered=False,
            special=False,
            site_editor=False,
            admin=True,
            anon=False,
        )

        approvers = ['registered@example.com']
        self.check_function_results(
            permissions.can_review_gate,
            (self.feature_1, None, approvers),
            unregistered=False,
            registered=True,
            special=False,
            site_editor=False,
            admin=True,
            anon=False,
        )

    def test_can_view_enterprise_release_notes(self):
        # Test non google or chromium user
        """Test can view enterprise release notes."""
        testing_config.sign_in('user@example.com', 123)
        self.assertEqual(
            False,
            permissions.is_google_or_chromium_account(users.get_current_user()),
        )  # noqa: E501

        # Test google user
        testing_config.sign_in('user@google.com', 123)
        self.assertEqual(
            True,
            permissions.is_google_or_chromium_account(users.get_current_user()),
        )  # noqa: E501

        # Test chromium user
        testing_config.sign_in('user@chromium.org', 123)
        self.assertEqual(
            True,
            permissions.is_google_or_chromium_account(users.get_current_user()),
        )  # noqa: E501


class RequireAdminSiteTests(testing_config.CustomTestCase):
    """Tests for the require_admin_site decorator."""

    def setUp(self):
        """Set up the test environment."""
        self.app_user = user_models.AppUser(email='registered@example.com')
        self.app_user.put()

        self.app_editor = user_models.AppUser(email='editor@example.com')
        self.app_editor.is_site_editor = True
        self.app_editor.put()

        self.app_admin = user_models.AppUser(email='admin@example.com')
        self.app_admin.is_admin = True
        self.app_admin.put()

    def tearDown(self):
        """Clean up the test environment."""
        self.app_user.delete()
        self.app_editor.delete()
        self.app_admin.delete()

    def test_require_admin_site__unregistered_user(self):
        """Wrapped method rejects call from an unregistered user."""
        handler = MockHandler()
        testing_config.sign_in('unregistered@example.com', 123)
        with test_app.test_request_context('/path', method='POST'):
            with self.assertRaises(werkzeug.exceptions.Forbidden):
                handler.do_post()
        self.assertEqual(handler.called_with, None)

    def test_require_admin_site__registered_user(self):
        """Wrapped method rejects call from registered non-admin user."""
        handler = MockHandler()
        testing_config.sign_in('registered@example.com', 123)
        with test_app.test_request_context('/path', method='POST'):
            with self.assertRaises(werkzeug.exceptions.Forbidden):
                handler.do_post()
        self.assertEqual(handler.called_with, None)

    def test_require_admin_site__googler(self):
        """Wrapped method rejects call from googler."""
        handler = MockHandler()
        testing_config.sign_in('user@google.com', 123)
        with test_app.test_request_context('/path', method='POST'):
            with self.assertRaises(werkzeug.exceptions.Forbidden):
                handler.do_post()
        self.assertEqual(handler.called_with, None)

    def test_require_admin_site__editor(self):
        """Wrapped method rejects call from a site editor."""
        handler = MockHandler()
        testing_config.sign_in('editor@example.com', 123)
        with test_app.test_request_context('/path', method='POST'):
            with self.assertRaises(werkzeug.exceptions.Forbidden):
                handler.do_post()
        self.assertEqual(handler.called_with, None)

    def test_require_admin_site__admin(self):
        """Wrapped method accepts call from an admin user."""
        handler = MockHandler()
        testing_config.sign_in('admin@example.com', 123)
        with test_app.test_request_context('/path'):
            actual_response = handler.do_get(123, 234)
        self.assertEqual(handler.called_with, (123, 234))
        self.assertEqual({'message': 'did get'}, actual_response)

        with test_app.test_request_context('/path', method='POST'):
            actual_response = handler.do_post(345, 456)
        self.assertEqual(handler.called_with, (345, 456))
        self.assertEqual({'message': 'did post'}, actual_response)

    def test_require_admin_site__anon(self):
        """Wrapped method rejects call from anon, but offers sign-in."""
        handler = MockHandler()
        testing_config.sign_out()
        with test_app.test_request_context('/path'):
            actual_response = handler.do_get(123, 234)
        self.assertEqual(handler.called_with, None)
        self.assertEqual(302, actual_response.status_code)

        with test_app.test_request_context('/path', method='POST'):
            with self.assertRaises(werkzeug.exceptions.Forbidden):
                handler.do_post(345, 456)
        self.assertEqual(handler.called_with, None)
