# Copyright 2014 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unittests for gob_util.py"""

import base64
import http.client
import json
import tempfile
import time
import urllib.request

from chromite.lib import config_lib
from chromite.lib import cros_test_lib
from chromite.lib import gob_util
from chromite.lib import timeout_util


gob_util.TRY_LIMIT = 1


class FakeHTTPResponse(object):
    """Enough of a HTTPResponse for FetchUrl.

    See https://docs.python.org/3/library/http.client.html#httpresponse-objects
    for more details.
    """

    def __init__(
        self, body=b"", headers=(), reason=None, status=200, version=11
    ):
        if reason is None:
            reason = http.client.responses[status]

        self.body = body
        self.headers = dict(headers)
        self.msg = None
        self.reason = reason
        self.status = status
        self.version = version

    def read(self):
        return self.body

    def getheader(self, name, default=None):
        return self.headers.get(name, default)

    def getheaders(self):
        return tuple(self.headers.items())


class GobTest(cros_test_lib.MockTestCase):
    """Unittests that use mocks."""

    UTF8_DATA = b"That\xe2\x80\x99s an error. That\xe2\x80\x99s all we know."

    def setUp(self):
        self.PatchObject(gob_util, "CreateHttpReq", autospec=False)
        self.conn = self.PatchObject(urllib.request, "urlopen")

    def testUtf8Response(self):
        """Handle gerrit responses w/UTF8 in them."""
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=self.UTF8_DATA
        )
        gob_util.FetchUrl("", "")

    def testUtf8Response502(self):
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=self.UTF8_DATA, status=502
        )

        with self.assertRaises(gob_util.InternalGOBError):
            gob_util.FetchUrl("", "")

    def testConnectionTimeout(self):
        """Exercise the timeout process."""
        # To finish the test quickly, we need to shorten the timeout.
        self.PatchObject(gob_util, "REQUEST_TIMEOUT_SECONDS", 1)

        # Setup a 'hanging' network connection.
        def simulateHang(*_args, **_kwargs):
            time.sleep(30)
            self.fail("Would hang forever.")

        self.conn.return_value.__enter__.side_effect = simulateHang

        # Verify that we fail, with expected timeout error.
        with self.assertRaises(timeout_util.TimeoutError):
            gob_util.FetchUrl("", "")

    def testHtmlParser(self):
        """Verify that GOB error message is parsed properly."""
        html_data = """
<!DOCTYPE html>
 <html lang=en>
 <meta charset=utf-8>
 <meta name=viewport>
 <title>Error 403 (Forbidden)!!1</title>
 <style>"*{margin:0;padding:0}"</style>
 <div id="af-error-container">
 <a href=//www.google.com><span id=logo aria-label=Google></span></a>

 <p>Error <b>403.</b><br><ins>That's an error.<p>Too bad...</ins>
 </div>
 <div id="come other stuff">
    some other stuff
 </div>
<html>"""
        expected_parsed_data = """Error 403.
That's an error.

Too bad..."""
        ep = gob_util.ErrorParser()
        ep.feed(html_data)
        ep.close()
        self.assertEqual(expected_parsed_data, ep.ParsedDiv())

    def testCreateChange(self):
        body = json.dumps({"change_num": 123456}).encode()
        xss_protection_prefix = b")]}'\n"
        body = xss_protection_prefix + body
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=200
        )
        change_json = gob_util.CreateChange(
            "some.git.url", "project", "branch", "subject", True
        )
        self.assertEqual(change_json["change_num"], 123456)

        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=201
        )
        change_json = gob_util.CreateChange(
            "some.git.url", "project", "branch", "subject", True
        )
        self.assertEqual(change_json["change_num"], 123456)

    def testChangeEdit(self):
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body={}, status=204
        )
        gob_util.ChangeEdit(
            "some.git.url", 123456, "some/file/path", "some file contents"
        )

    def testPublishChangeEdit(self):
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body={}, status=204
        )
        gob_util.PublishChangeEdit("some.git.url", 123456)

    def testGetFileContents(self):
        expected_contents = "some file contents"
        body = base64.b64encode(expected_contents.encode())
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=200
        )
        contents = gob_util.GetFileContentsOnHead(
            "some.git.url", "some/file/path"
        )
        self.assertEqual(contents, expected_contents)
        contents = gob_util.GetFileContents("some.git.url", "some/file/path")
        self.assertEqual(contents, expected_contents)

    def testGetFileContentsFromGerrit(self):
        """Test for GetFileContentsFromGerrit()"""

        expected_contents = "some file contents"
        body = base64.b64encode(expected_contents.encode())
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=200
        )
        contents = gob_util.GetFileContentsFromGerrit(
            "some.git.url", "100000", "some/file/path"
        )
        self.assertEqual(contents, expected_contents)

    def testGetChangeMergeable(self):
        """Test for GetChangeMergeable()"""

        mergeable_info = {"mergeable": True}
        body = json.dumps(mergeable_info).encode()
        xss_protection_prefix = b")]}'\n"
        body = xss_protection_prefix + body
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=200
        )
        result = gob_util.GetChangeMergeable("some.git.url", "100000")
        self.assertEqual(result, mergeable_info)

    def testRebase(self):
        """Test for Rebase()"""

        change_info = {}
        body = json.dumps(change_info).encode()
        xss_protection_prefix = b")]}'\n"
        body = xss_protection_prefix + body
        self.conn.return_value.__enter__.return_value = FakeHTTPResponse(
            body=body, status=200
        )
        result = gob_util.Rebase("some.git.url", "100000")
        self.assertEqual(result, change_info)


class GetCookieTests(cros_test_lib.TestCase):
    """Unittests for GetCookies()"""

    def testSimple(self):
        with tempfile.NamedTemporaryFile(mode="w+") as f:
            f.write(".googlesource.com\tTRUE\t/f\tTRUE\t2147483647\to\tfoo=bar")
            f.flush()
            cookies = gob_util.GetCookies(
                "foo.googlesource.com", "/foo", [f.name]
            )
            self.assertEqual(cookies, {"o": "foo=bar"})
            cookies = gob_util.GetCookies("google.com", "/foo", [f.name])
            self.assertEqual(cookies, {})
            cookies = gob_util.GetCookies("foo.googlesource.com", "/", [f.name])
            self.assertEqual(cookies, {})


@cros_test_lib.pytestmark_network_test
class NetworkGobTest(cros_test_lib.TestCase):
    """Unittests that talk to real Gerrit."""

    def test200(self):
        """Test successful loading of change."""
        gob_util.FetchUrlJson(
            config_lib.GetSiteParams().EXTERNAL_GOB_HOST,
            "changes/227254/detail",
        )

    def test404(self):
        gob_util.FetchUrlJson(
            config_lib.GetSiteParams().EXTERNAL_GOB_HOST, "foo/bar/baz"
        )

    def test404Exception(self):
        with self.assertRaises(gob_util.GOBError) as ex:
            gob_util.FetchUrlJson(
                config_lib.GetSiteParams().EXTERNAL_GOB_HOST,
                "foo/bar/baz",
                ignore_404=False,
            )
        self.assertEqual(ex.exception.http_status, 404)