#!/usr/bin/env vpython3
# Copyright 2020 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import gerrit_util
import owners_client

alice = "alice@example.com"
bob = "bob@example.com"
chris = "chris@example.com"
dave = "dave@example.com"
emily = "emily@example.com"


class GerritClientTest(unittest.TestCase):
    def setUp(self):
        self.client = owners_client.GerritClient("host", "project", "branch")
        self.addCleanup(mock.patch.stopall)

    def testListOwners(self):
        mock.patch(
            "gerrit_util.GetOwnersForFile",
            return_value={
                "code_owners": [
                    {"account": {"email": "approver@example.com"}},
                    {
                        "account": {"email": "reviewer@example.com"},
                    },
                    {
                        "account": {"email": "missing@example.com"},
                    },
                    {
                        "account": {},
                    },
                ]
            },
        ).start()
        self.assertEqual(
            [
                "approver@example.com",
                "reviewer@example.com",
                "missing@example.com",
            ],
            self.client.ListOwners(os.path.join("bar", "everyone", "foo.txt")),
        )

        # Result should be cached.
        self.assertEqual(
            [
                "approver@example.com",
                "reviewer@example.com",
                "missing@example.com",
            ],
            self.client.ListOwners(os.path.join("bar", "everyone", "foo.txt")),
        )
        # Always use slashes as separators.
        gerrit_util.GetOwnersForFile.assert_called_once_with(
            "host",
            "project",
            "branch",
            "bar/everyone/foo.txt",
            resolve_all_users=False,
            highest_score_only=False,
            seed=mock.ANY,
        )

    def testListOwnersOwnedByAll(self):
        mock.patch(
            "gerrit_util.GetOwnersForFile",
            side_effect=[
                {
                    "code_owners": [
                        {
                            "account": {"email": "foo@example.com"},
                        },
                    ],
                    "owned_by_all_users": True,
                },
                {
                    "code_owners": [
                        {
                            "account": {"email": "bar@example.com"},
                        },
                    ],
                    "owned_by_all_users": False,
                },
            ],
        ).start()
        self.assertEqual(
            ["foo@example.com", self.client.EVERYONE],
            self.client.ListOwners("foo.txt"),
        )
        self.assertEqual(["bar@example.com"], self.client.ListOwners("bar.txt"))


class TestClient(owners_client.OwnersClient):
    def __init__(self, owners_by_path):
        super(TestClient, self).__init__()
        self.owners_by_path = owners_by_path

    def ListOwners(self, path):
        return self.owners_by_path[path]


class OwnersClientTest(unittest.TestCase):
    def setUp(self):
        self.owners = {}
        self.client = TestClient(self.owners)

    def testGetFilesApprovalStatus(self):
        self.client.owners_by_path = {
            "approved": ["approver@example.com"],
            "pending": ["reviewer@example.com"],
            "insufficient": ["insufficient@example.com"],
            "everyone": [owners_client.OwnersClient.EVERYONE],
        }
        self.assertEqual(
            self.client.GetFilesApprovalStatus(
                ["approved", "pending", "insufficient"],
                ["approver@example.com"],
                ["reviewer@example.com"],
            ),
            {
                "approved": owners_client.OwnersClient.APPROVED,
                "pending": owners_client.OwnersClient.PENDING,
                "insufficient": owners_client.OwnersClient.INSUFFICIENT_REVIEWERS,
            },
        )
        self.assertEqual(
            self.client.GetFilesApprovalStatus(
                ["everyone"], ["anyone@example.com"], []
            ),
            {"everyone": owners_client.OwnersClient.APPROVED},
        )
        self.assertEqual(
            self.client.GetFilesApprovalStatus(
                ["everyone"], [], ["anyone@example.com"]
            ),
            {"everyone": owners_client.OwnersClient.PENDING},
        )
        self.assertEqual(
            self.client.GetFilesApprovalStatus(["everyone"], [], []),
            {"everyone": owners_client.OwnersClient.INSUFFICIENT_REVIEWERS},
        )

    def testScoreOwners(self):
        self.client.owners_by_path = {"a": [alice, bob, chris]}
        self.assertEqual(
            self.client.ScoreOwners(self.client.owners_by_path.keys()),
            [alice, bob, chris],
        )

        self.client.owners_by_path = {
            "a": [alice, bob],
            "b": [bob],
            "c": [bob, chris],
        }
        self.assertEqual(
            self.client.ScoreOwners(self.client.owners_by_path.keys()),
            [alice, bob, chris],
        )

        self.client.owners_by_path = {
            "a": [alice, bob],
            "b": [bob],
            "c": [bob, chris],
        }
        self.assertEqual(
            self.client.ScoreOwners(
                self.client.owners_by_path.keys(), exclude=[chris]
            ),
            [alice, bob],
        )

        self.client.owners_by_path = {
            "a": [alice, bob, chris, dave],
            "b": [chris, bob, dave],
            "c": [chris, dave],
            "d": [alice, chris, dave],
        }
        self.assertEqual(
            self.client.ScoreOwners(self.client.owners_by_path.keys()),
            [alice, chris, bob, dave],
        )

    def assertSuggestsOwners(self, owners_by_path, exclude=None):
        self.client.owners_by_path = owners_by_path
        suggested = self.client.SuggestOwners(
            owners_by_path.keys(), exclude=exclude
        )

        # Owners should appear only once
        self.assertEqual(len(suggested), len(set(suggested)))

        # All paths should be covered.
        suggested = set(suggested)
        for owners in owners_by_path.values():
            self.assertTrue(suggested & set(owners))

        # No excluded owners should be present.
        if exclude:
            for owner in suggested:
                self.assertNotIn(owner, exclude)

    def testSuggestOwners(self):
        self.assertSuggestsOwners({})
        self.assertSuggestsOwners({"a": [alice]})
        self.assertSuggestsOwners({"abcd": [alice, bob, chris, dave]})
        self.assertSuggestsOwners(
            {"abcd": [alice, bob, chris, dave]}, exclude=[alice, bob]
        )
        self.assertSuggestsOwners(
            {
                "ae": [alice, emily],
                "be": [bob, emily],
                "ce": [chris, emily],
                "de": [dave, emily],
            }
        )
        self.assertSuggestsOwners(
            {
                "ad": [alice, dave],
                "cad": [chris, alice, dave],
                "ead": [emily, alice, dave],
                "bd": [bob, dave],
            }
        )
        self.assertSuggestsOwners(
            {"a": [alice], "b": [bob], "c": [chris], "ad": [alice, dave]}
        )
        self.assertSuggestsOwners(
            {
                "abc": [alice, bob, chris],
                "acb": [alice, chris, bob],
                "bac": [bob, alice, chris],
                "bca": [bob, chris, alice],
                "cab": [chris, alice, bob],
                "cba": [chris, bob, alice],
            }
        )

        # Check that we can handle a large amount of files with unrelated
        # owners.
        self.assertSuggestsOwners({str(x): [str(x)] for x in range(100)})

    def testBatchListOwners(self):
        self.client.owners_by_path = {
            "bar/everyone/foo.txt": [alice, bob],
            "bar/everyone/bar.txt": [bob],
            "bar/foo/": [bob, chris],
        }

        self.assertEqual(
            {
                "bar/everyone/foo.txt": [alice, bob],
                "bar/everyone/bar.txt": [bob],
                "bar/foo/": [bob, chris],
            },
            self.client.BatchListOwners(
                ["bar/everyone/foo.txt", "bar/everyone/bar.txt", "bar/foo/"]
            ),
        )

    def testSuggestMinimalOwners(self):
        self.client.owners_by_path = {
            "bar/everyone/foo.txt": [alice, bob, emily],
            "bar/everyone/bar.txt": [bob, emily],
            "bar/foo/": [bob, chris, emily],
            "baz/baz/baz": [chris, dave, emily],
        }

        self.assertEqual(
            [bob],
            self.client.SuggestMinimalOwners(
                ["bar/everyone/foo.txt", "bar/everyone/bar.txt", "bar/foo/"]
            ),
        )

        self.assertEqual(
            [chris],
            self.client.SuggestMinimalOwners(["bar/foo/", "baz/baz/baz"]),
        )

        # If no common owner exists, fallback to returning multiple owners
        self.assertEqual(
            [alice, bob, chris],
            self.client.SuggestMinimalOwners(
                [
                    "bar/everyone/foo.txt",
                    "bar/everyone/bar.txt",
                    "bar/foo/",
                    "baz/baz/baz",
                ],
                exclude=[emily],
            ),
        )


class GetCodeOwnersClientTest(unittest.TestCase):
    def setUp(self):
        mock.patch("gerrit_util.IsCodeOwnersEnabledOnHost").start()
        self.addCleanup(mock.patch.stopall)

    def testGetCodeOwnersClient_CodeOwnersEnabled(self):
        gerrit_util.IsCodeOwnersEnabledOnHost.return_value = True
        self.assertIsInstance(
            owners_client.GetCodeOwnersClient("host", "project", "branch"),
            owners_client.GerritClient,
        )

    def testGetCodeOwnersClient_CodeOwnersDisabled(self):
        gerrit_util.IsCodeOwnersEnabledOnHost.return_value = False
        with self.assertRaises(Exception):
            owners_client.GetCodeOwnersClient("", "", "")


if __name__ == "__main__":
    unittest.main()