# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.


import unittest

from pyhugegraph.utils.exceptions import NotFoundError
from tests.client_utils import ClientUtils


class TestAuthManager(unittest.TestCase):
    client = None
    auth = None

    @classmethod
    def setUpClass(cls):
        cls.client = ClientUtils()
        cls.auth = cls.client.auth

    @classmethod
    def tearDownClass(cls):
        cls.client.clear_graph_all_data()

    def setUp(self):
        users = self.auth.list_users()
        for user in users["users"]:
            if user["user_creator"] != "system":
                self.auth.delete_user(user["id"])

        groups = self.auth.list_groups()
        for group in groups["groups"]:
            self.auth.delete_group(group["id"])

        belongs = self.auth.list_belongs()
        for belong in belongs["belongs"]:
            self.auth.delete_belong(belong["id"])

        targets = self.auth.list_targets()
        for target in targets["targets"]:
            self.auth.delete_target(target["id"])

        accesses = self.auth.list_accesses()
        for access in accesses["accesses"]:
            self.auth.revoke_accesses(access["id"])

    def tearDown(self):
        pass

    def test_user_operations(self):
        users = self.auth.list_users()
        self.assertTrue(users["users"][0]["user_creator"] == "system")

        # {'user_password': '',
        # 'user_update': '2024-01-26 02:45:57.317', 'user_name': 'test_user',
        # 'user_creator': 'admin', 'id': '-30:test_user', 'user_create': '2024-01-26 02:45:57.317'}
        user = self.auth.create_user("test_user", "password")
        self.assertEqual(user["user_name"], "test_user")

        user = self.auth.get_user(user["id"])
        self.assertEqual(user["user_name"], "test_user")

        # Modify the user
        user = self.auth.modify_user(user["id"], user_email="hugegraph@apache.org")
        self.assertEqual(user["user_email"], "hugegraph@apache.org")

        # Delete the user
        self.auth.delete_user(user["id"])
        # Verify the user was deleted
        try:
            user = self.auth.get_user(user["id"])
        except NotFoundError as e:
            self.assertTrue(f"Can\\'t find user with id \\'{user['id']}\\'" in str(e))

    def test_group_operations(self):
        # Create a group
        group = self.auth.create_group("test_group")
        self.assertEqual(group["group_name"], "test_group")

        groups = self.auth.list_groups()
        self.assertEqual(groups["groups"][0]["group_name"], "test_group")

        # Get the group
        group = self.auth.get_group(group["id"])
        self.assertEqual(group["group_name"], "test_group")

        # Modify the group
        group = self.auth.modify_group(
            group["id"], group_description="test_description"
        )
        self.assertEqual(group["group_description"], "test_description")

        # Delete the group
        self.auth.delete_group(group["id"])
        # Verify the group was deleted
        try:
            group = self.auth.get_group(group["id"])
        except NotFoundError as e:
            self.assertTrue(f"Can\\'t find group with id \\'{group['id']}\\'" in str(e))

    def test_target_operations(self):
        # Create a target
        target = self.auth.create_target(
            "test_target",
            "graph1",
            "127.0.0.1:8080",
            [{"type": "VERTEX", "label": "person", "properties": {"city": "Beijing"}}],
        )
        # Verify the target was created
        self.assertEqual(target["target_name"], "test_target")

        # Get the target
        target = self.auth.get_target(target["id"])
        self.assertEqual(target["target_name"], "test_target")

        # Modify the target
        target = self.auth.update_target(
            target["id"],
            "test_target",
            "graph1",
            "127.0.0.1:8080",
            [{"type": "VERTEX", "label": "person", "properties": {"city": "Shanghai"}}],
        )
        # Verify the target was modified
        self.assertEqual(
            target["target_resources"][0]["properties"]["city"], "Shanghai"
        )

        # Delete the target
        self.auth.delete_target(target["id"])
        # Verify the target was deleted
        with self.assertRaises(Exception):
            self.auth.get_target(target["id"])

    def test_belong_operations(self):
        user = self.auth.create_user("all", "password")
        group = self.auth.create_group("all")

        # Create a belong
        # {'belong_create': '2024-01-29 02:04:41.161', 'belong_creator': 'admin',
        # 'belong_update': '2024-01-29 02:04:41.161', 'id': 'S-30:all>-49>>S-36:all',
        # 'user': '-30:all', 'group': '-36:all'}
        belong = self.auth.create_belong(user["id"], group["id"])
        # Verify the belong was created
        self.assertEqual(belong["user"], user["id"])
        self.assertEqual(belong["group"], group["id"])

        belongs = self.auth.list_belongs()
        self.assertEqual(belongs["belongs"][0]["user"], user["id"])

        # Get the belong
        belong = self.auth.get_belong(belong["id"])
        self.assertEqual(belong["user"], user["id"])
        self.assertEqual(belong["group"], group["id"])

        # Modify the belong
        belong = self.auth.update_belong(belong["id"], "WRITE")
        # Verify the belong was modified
        self.assertEqual(belong["belong_description"], "WRITE")

        # Delete the belong
        self.auth.delete_belong(belong["id"])
        # Verify the belong was deleted
        with self.assertRaises(Exception):
            self.auth.get_belong(belong["id"])

    def test_access_operations(self):
        # Create a permission
        group = self.auth.create_group("test_group")
        target = self.auth.create_target(
            "test_target",
            "graph1",
            "127.0.0.1:8080",
            [{"type": "VERTEX", "label": "person", "properties": {"city": "Beijing"}}],
        )
        access = self.auth.grant_accesses(group["id"], target["id"], "READ")
        # Verify the permission was created
        self.assertEqual(access["group"], group["id"])
        self.assertEqual(access["target"], target["id"])
        self.assertEqual(access["access_permission"], "READ")

        accesses = self.auth.list_accesses()
        self.assertEqual(accesses["accesses"][0]["group"], group["id"])

        # Get the permission
        access = self.auth.get_accesses(access["id"])
        self.assertEqual(access["group"], group["id"])
        self.assertEqual(access["target"], target["id"])
        self.assertEqual(access["access_permission"], "READ")

        # Modify the permission
        access = self.auth.modify_accesses(access["id"], "test_description")
        # Verify the permission was modified
        self.assertEqual(access["access_description"], "test_description")

        # Delete the permission
        self.auth.revoke_accesses(access["id"])
        # Verify the permission was deleted
        with self.assertRaises(Exception):
            self.auth.get_accesses(access["id"])