# Copyright 2017 Google LLC
#
# 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.

from __future__ import unicode_literals
from unittest import mock

import pytest

from google.api import auth_pb2
from google.api_core import path_template


@pytest.mark.parametrize(
    "tmpl, args, kwargs, expected_result",
    [
        # Basic positional params
        ["/v1/*", ["a"], {}, "/v1/a"],
        ["/v1/**", ["a/b"], {}, "/v1/a/b"],
        ["/v1/*/*", ["a", "b"], {}, "/v1/a/b"],
        ["/v1/*/*/**", ["a", "b", "c/d"], {}, "/v1/a/b/c/d"],
        # Basic named params
        ["/v1/{name}", [], {"name": "parent"}, "/v1/parent"],
        ["/v1/{name=**}", [], {"name": "parent/child"}, "/v1/parent/child"],
        # Named params with a sub-template
        ["/v1/{name=parent/*}", [], {"name": "parent/child"}, "/v1/parent/child"],
        [
            "/v1/{name=parent/**}",
            [],
            {"name": "parent/child/object"},
            "/v1/parent/child/object",
        ],
        # Combining positional and named params
        ["/v1/*/{name}", ["a"], {"name": "parent"}, "/v1/a/parent"],
        ["/v1/{name}/*", ["a"], {"name": "parent"}, "/v1/parent/a"],
        [
            "/v1/{parent}/*/{child}/*",
            ["a", "b"],
            {"parent": "thor", "child": "thorson"},
            "/v1/thor/a/thorson/b",
        ],
        ["/v1/{name}/**", ["a/b"], {"name": "parent"}, "/v1/parent/a/b"],
        # Combining positional and named params with sub-templates.
        [
            "/v1/{name=parent/*}/*",
            ["a"],
            {"name": "parent/child"},
            "/v1/parent/child/a",
        ],
        [
            "/v1/*/{name=parent/**}",
            ["a"],
            {"name": "parent/child/object"},
            "/v1/a/parent/child/object",
        ],
    ],
)
def test_expand_success(tmpl, args, kwargs, expected_result):
    result = path_template.expand(tmpl, *args, **kwargs)
    assert result == expected_result
    assert path_template.validate(tmpl, result)


@pytest.mark.parametrize(
    "tmpl, args, kwargs, exc_match",
    [
        # Missing positional arg.
        ["v1/*", [], {}, "Positional"],
        # Missing named arg.
        ["v1/{name}", [], {}, "Named"],
    ],
)
def test_expanded_failure(tmpl, args, kwargs, exc_match):
    with pytest.raises(ValueError, match=exc_match):
        path_template.expand(tmpl, *args, **kwargs)


@pytest.mark.parametrize(
    "request_obj, field, expected_result",
    [
        [{"field": "stringValue"}, "field", "stringValue"],
        [{"field": "stringValue"}, "nosuchfield", None],
        [{"field": "stringValue"}, "field.subfield", None],
        [{"field": {"subfield": "stringValue"}}, "field", None],
        [{"field": {"subfield": "stringValue"}}, "field.subfield", "stringValue"],
        [{"field": {"subfield": [1, 2, 3]}}, "field.subfield", [1, 2, 3]],
        [{"field": {"subfield": "stringValue"}}, "field", None],
        [{"field": {"subfield": "stringValue"}}, "field.nosuchfield", None],
        [
            {"field": {"subfield": {"subsubfield": "stringValue"}}},
            "field.subfield.subsubfield",
            "stringValue",
        ],
        ["string", "field", None],
    ],
)
def test_get_field(request_obj, field, expected_result):
    result = path_template.get_field(request_obj, field)
    assert result == expected_result


@pytest.mark.parametrize(
    "request_obj, field, expected_result",
    [
        [{"field": "stringValue"}, "field", {}],
        [{"field": "stringValue"}, "nosuchfield", {"field": "stringValue"}],
        [{"field": "stringValue"}, "field.subfield", {"field": "stringValue"}],
        [{"field": {"subfield": "stringValue"}}, "field.subfield", {"field": {}}],
        [
            {"field": {"subfield": "stringValue", "q": "w"}, "e": "f"},
            "field.subfield",
            {"field": {"q": "w"}, "e": "f"},
        ],
        [
            {"field": {"subfield": "stringValue"}},
            "field.nosuchfield",
            {"field": {"subfield": "stringValue"}},
        ],
        [
            {"field": {"subfield": {"subsubfield": "stringValue", "q": "w"}}},
            "field.subfield.subsubfield",
            {"field": {"subfield": {"q": "w"}}},
        ],
        ["string", "field", "string"],
        ["string", "field.subfield", "string"],
    ],
)
def test_delete_field(request_obj, field, expected_result):
    path_template.delete_field(request_obj, field)
    assert request_obj == expected_result


@pytest.mark.parametrize(
    "tmpl, path",
    [
        # Single segment template, but multi segment value
        ["v1/*", "v1/a/b"],
        ["v1/*/*", "v1/a/b/c"],
        # Single segement named template, but multi segment value
        ["v1/{name}", "v1/a/b"],
        ["v1/{name}/{value}", "v1/a/b/c"],
        # Named value with a sub-template but invalid value
        ["v1/{name=parent/*}", "v1/grandparent/child"],
    ],
)
def test_validate_failure(tmpl, path):
    assert not path_template.validate(tmpl, path)


def test__expand_variable_match_unexpected():
    match = mock.Mock(spec=["group"])
    match.group.return_value = None
    with pytest.raises(ValueError, match="Unknown"):
        path_template._expand_variable_match([], {}, match)


def test__replace_variable_with_pattern():
    match = mock.Mock(spec=["group"])
    match.group.return_value = None
    with pytest.raises(ValueError, match="Unknown"):
        path_template._replace_variable_with_pattern(match)


@pytest.mark.parametrize(
    "http_options, message, request_kwargs, expected_result",
    [
        [
            [["get", "/v1/no/template", ""]],
            None,
            {"foo": "bar"},
            ["get", "/v1/no/template", {}, {"foo": "bar"}],
        ],
        [
            [["get", "/v1/no/template", ""]],
            auth_pb2.AuthenticationRule(selector="bar"),
            {},
            [
                "get",
                "/v1/no/template",
                None,
                auth_pb2.AuthenticationRule(selector="bar"),
            ],
        ],
        # Single templates
        [
            [["get", "/v1/{field}", ""]],
            None,
            {"field": "parent"},
            ["get", "/v1/parent", {}, {}],
        ],
        [
            [["get", "/v1/{selector}", ""]],
            auth_pb2.AuthenticationRule(selector="parent"),
            {},
            ["get", "/v1/parent", None, auth_pb2.AuthenticationRule()],
        ],
        [
            [["get", "/v1/{field.sub}", ""]],
            None,
            {"field": {"sub": "parent"}, "foo": "bar"},
            ["get", "/v1/parent", {}, {"field": {}, "foo": "bar"}],
        ],
        [
            [["get", "/v1/{oauth.canonical_scopes}", ""]],
            auth_pb2.AuthenticationRule(
                selector="bar",
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="parent"),
            ),
            {},
            [
                "get",
                "/v1/parent",
                None,
                auth_pb2.AuthenticationRule(
                    selector="bar", oauth=auth_pb2.OAuthRequirements()
                ),
            ],
        ],
    ],
)
def test_transcode_base_case(http_options, message, request_kwargs, expected_result):
    http_options, expected_result = helper_test_transcode(http_options, expected_result)
    result = path_template.transcode(http_options, message, **request_kwargs)
    assert result == expected_result


@pytest.mark.parametrize(
    "http_options, message, request_kwargs, expected_result",
    [
        [
            [["get", "/v1/{field.subfield}", ""]],
            None,
            {"field": {"subfield": "parent"}, "foo": "bar"},
            ["get", "/v1/parent", {}, {"field": {}, "foo": "bar"}],
        ],
        [
            [["get", "/v1/{oauth.canonical_scopes}", ""]],
            auth_pb2.AuthenticationRule(
                selector="bar",
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="parent"),
            ),
            {},
            [
                "get",
                "/v1/parent",
                None,
                auth_pb2.AuthenticationRule(
                    selector="bar", oauth=auth_pb2.OAuthRequirements()
                ),
            ],
        ],
        [
            [["get", "/v1/{field.subfield.subsubfield}", ""]],
            None,
            {"field": {"subfield": {"subsubfield": "parent"}}, "foo": "bar"},
            ["get", "/v1/parent", {}, {"field": {"subfield": {}}, "foo": "bar"}],
        ],
        [
            [["get", "/v1/{field.subfield1}/{field.subfield2}", ""]],
            None,
            {"field": {"subfield1": "parent", "subfield2": "child"}, "foo": "bar"},
            ["get", "/v1/parent/child", {}, {"field": {}, "foo": "bar"}],
        ],
        [
            [["get", "/v1/{selector}/{oauth.canonical_scopes}", ""]],
            auth_pb2.AuthenticationRule(
                selector="parent",
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="child"),
            ),
            {"field": {"subfield1": "parent", "subfield2": "child"}, "foo": "bar"},
            [
                "get",
                "/v1/parent/child",
                None,
                auth_pb2.AuthenticationRule(oauth=auth_pb2.OAuthRequirements()),
            ],
        ],
    ],
)
def test_transcode_subfields(http_options, message, request_kwargs, expected_result):
    http_options, expected_result = helper_test_transcode(http_options, expected_result)
    result = path_template.transcode(http_options, message, **request_kwargs)
    assert result == expected_result


@pytest.mark.parametrize(
    "http_options, message, request_kwargs, expected_result",
    [
        # Single segment wildcard
        [
            [["get", "/v1/{field=*}", ""]],
            None,
            {"field": "parent"},
            ["get", "/v1/parent", {}, {}],
        ],
        [
            [["get", "/v1/{selector=*}", ""]],
            auth_pb2.AuthenticationRule(selector="parent"),
            {},
            ["get", "/v1/parent", None, auth_pb2.AuthenticationRule()],
        ],
        [
            [["get", "/v1/{field=a/*/b/*}", ""]],
            None,
            {"field": "a/parent/b/child", "foo": "bar"},
            ["get", "/v1/a/parent/b/child", {}, {"foo": "bar"}],
        ],
        [
            [["get", "/v1/{selector=a/*/b/*}", ""]],
            auth_pb2.AuthenticationRule(
                selector="a/parent/b/child", allow_without_credential=True
            ),
            {},
            [
                "get",
                "/v1/a/parent/b/child",
                None,
                auth_pb2.AuthenticationRule(allow_without_credential=True),
            ],
        ],
        # Double segment wildcard
        [
            [["get", "/v1/{field=**}", ""]],
            None,
            {"field": "parent/p1"},
            ["get", "/v1/parent/p1", {}, {}],
        ],
        [
            [["get", "/v1/{selector=**}", ""]],
            auth_pb2.AuthenticationRule(selector="parent/p1"),
            {},
            ["get", "/v1/parent/p1", None, auth_pb2.AuthenticationRule()],
        ],
        [
            [["get", "/v1/{field=a/**/b/**}", ""]],
            None,
            {"field": "a/parent/p1/b/child/c1", "foo": "bar"},
            ["get", "/v1/a/parent/p1/b/child/c1", {}, {"foo": "bar"}],
        ],
        [
            [["get", "/v1/{selector=a/**/b/**}", ""]],
            auth_pb2.AuthenticationRule(
                selector="a/parent/p1/b/child/c1", allow_without_credential=True
            ),
            {},
            [
                "get",
                "/v1/a/parent/p1/b/child/c1",
                None,
                auth_pb2.AuthenticationRule(allow_without_credential=True),
            ],
        ],
        # Combined single and double segment wildcard
        [
            [["get", "/v1/{field=a/*/b/**}", ""]],
            None,
            {"field": "a/parent/b/child/c1"},
            ["get", "/v1/a/parent/b/child/c1", {}, {}],
        ],
        [
            [["get", "/v1/{selector=a/*/b/**}", ""]],
            auth_pb2.AuthenticationRule(selector="a/parent/b/child/c1"),
            {},
            ["get", "/v1/a/parent/b/child/c1", None, auth_pb2.AuthenticationRule()],
        ],
        [
            [["get", "/v1/{field=a/**/b/*}/v2/{name}", ""]],
            None,
            {"field": "a/parent/p1/b/child", "name": "first", "foo": "bar"},
            ["get", "/v1/a/parent/p1/b/child/v2/first", {}, {"foo": "bar"}],
        ],
        [
            [["get", "/v1/{selector=a/**/b/*}/v2/{oauth.canonical_scopes}", ""]],
            auth_pb2.AuthenticationRule(
                selector="a/parent/p1/b/child",
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="first"),
            ),
            {"field": "a/parent/p1/b/child", "name": "first", "foo": "bar"},
            [
                "get",
                "/v1/a/parent/p1/b/child/v2/first",
                None,
                auth_pb2.AuthenticationRule(oauth=auth_pb2.OAuthRequirements()),
            ],
        ],
    ],
)
def test_transcode_with_wildcard(
    http_options, message, request_kwargs, expected_result
):
    http_options, expected_result = helper_test_transcode(http_options, expected_result)
    result = path_template.transcode(http_options, message, **request_kwargs)
    assert result == expected_result


@pytest.mark.parametrize(
    "http_options, message, request_kwargs, expected_result",
    [
        # Single field body
        [
            [["post", "/v1/no/template", "data"]],
            None,
            {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
            ["post", "/v1/no/template", {"id": 1, "info": "some info"}, {"foo": "bar"}],
        ],
        [
            [["post", "/v1/no/template", "oauth"]],
            auth_pb2.AuthenticationRule(
                selector="bar",
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="child"),
            ),
            {},
            [
                "post",
                "/v1/no/template",
                auth_pb2.OAuthRequirements(canonical_scopes="child"),
                auth_pb2.AuthenticationRule(selector="bar"),
            ],
        ],
        [
            [["post", "/v1/{field=a/*}/b/{name=**}", "data"]],
            None,
            {
                "field": "a/parent",
                "name": "first/last",
                "data": {"id": 1, "info": "some info"},
                "foo": "bar",
            },
            [
                "post",
                "/v1/a/parent/b/first/last",
                {"id": 1, "info": "some info"},
                {"foo": "bar"},
            ],
        ],
        [
            [["post", "/v1/{selector=a/*}/b/{oauth.canonical_scopes=**}", "oauth"]],
            auth_pb2.AuthenticationRule(
                selector="a/parent",
                allow_without_credential=True,
                requirements=[auth_pb2.AuthRequirement(provider_id="p")],
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="first/last"),
            ),
            {},
            [
                "post",
                "/v1/a/parent/b/first/last",
                auth_pb2.OAuthRequirements(),
                auth_pb2.AuthenticationRule(
                    requirements=[auth_pb2.AuthRequirement(provider_id="p")],
                    allow_without_credential=True,
                ),
            ],
        ],
        # Wildcard body
        [
            [["post", "/v1/{field=a/*}/b/{name=**}", "*"]],
            None,
            {
                "field": "a/parent",
                "name": "first/last",
                "data": {"id": 1, "info": "some info"},
                "foo": "bar",
            },
            [
                "post",
                "/v1/a/parent/b/first/last",
                {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
                {},
            ],
        ],
        [
            [["post", "/v1/{selector=a/*}/b/{oauth.canonical_scopes=**}", "*"]],
            auth_pb2.AuthenticationRule(
                selector="a/parent",
                allow_without_credential=True,
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="first/last"),
            ),
            {
                "field": "a/parent",
                "name": "first/last",
                "data": {"id": 1, "info": "some info"},
                "foo": "bar",
            },
            [
                "post",
                "/v1/a/parent/b/first/last",
                auth_pb2.AuthenticationRule(
                    allow_without_credential=True, oauth=auth_pb2.OAuthRequirements()
                ),
                auth_pb2.AuthenticationRule(),
            ],
        ],
    ],
)
def test_transcode_with_body(http_options, message, request_kwargs, expected_result):
    http_options, expected_result = helper_test_transcode(http_options, expected_result)
    result = path_template.transcode(http_options, message, **request_kwargs)
    assert result == expected_result


@pytest.mark.parametrize(
    "http_options, message, request_kwargs, expected_result",
    [
        # Additional bindings
        [
            [
                ["post", "/v1/{field=a/*}/b/{name=**}", "extra_data"],
                ["post", "/v1/{field=a/*}/b/{name=**}", "*"],
            ],
            None,
            {
                "field": "a/parent",
                "name": "first/last",
                "data": {"id": 1, "info": "some info"},
                "foo": "bar",
            },
            [
                "post",
                "/v1/a/parent/b/first/last",
                {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
                {},
            ],
        ],
        [
            [
                [
                    "post",
                    "/v1/{selector=a/*}/b/{oauth.canonical_scopes=**}",
                    "extra_data",
                ],
                ["post", "/v1/{selector=a/*}/b/{oauth.canonical_scopes=**}", "*"],
            ],
            auth_pb2.AuthenticationRule(
                selector="a/parent",
                allow_without_credential=True,
                oauth=auth_pb2.OAuthRequirements(canonical_scopes="first/last"),
            ),
            {},
            [
                "post",
                "/v1/a/parent/b/first/last",
                auth_pb2.AuthenticationRule(
                    allow_without_credential=True, oauth=auth_pb2.OAuthRequirements()
                ),
                auth_pb2.AuthenticationRule(),
            ],
        ],
        [
            [
                ["get", "/v1/{field=a/*}/b/{name=**}", ""],
                ["get", "/v1/{field=a/*}/b/first/last", ""],
            ],
            None,
            {"field": "a/parent", "foo": "bar"},
            ["get", "/v1/a/parent/b/first/last", {}, {"foo": "bar"}],
        ],
        [
            [
                ["get", "/v1/{selector=a/*}/b/{oauth.allow_without_credential=**}", ""],
                ["get", "/v1/{selector=a/*}/b/first/last", ""],
            ],
            auth_pb2.AuthenticationRule(
                selector="a/parent",
                allow_without_credential=True,
                oauth=auth_pb2.OAuthRequirements(),
            ),
            {},
            [
                "get",
                "/v1/a/parent/b/first/last",
                None,
                auth_pb2.AuthenticationRule(
                    allow_without_credential=True, oauth=auth_pb2.OAuthRequirements()
                ),
            ],
        ],
    ],
)
def test_transcode_with_additional_bindings(
    http_options, message, request_kwargs, expected_result
):
    http_options, expected_result = helper_test_transcode(http_options, expected_result)
    result = path_template.transcode(http_options, message, **request_kwargs)
    assert result == expected_result


@pytest.mark.parametrize(
    "http_options, message, request_kwargs",
    [
        [[["get", "/v1/{name}", ""]], None, {"foo": "bar"}],
        [[["get", "/v1/{selector}", ""]], auth_pb2.AuthenticationRule(), {}],
        [[["get", "/v1/{name}", ""]], auth_pb2.AuthenticationRule(), {}],
        [[["get", "/v1/{name}", ""]], None, {"name": "first/last"}],
        [
            [["get", "/v1/{selector}", ""]],
            auth_pb2.AuthenticationRule(selector="first/last"),
            {},
        ],
        [[["get", "/v1/{name=mr/*/*}", ""]], None, {"name": "first/last"}],
        [
            [["get", "/v1/{selector=mr/*/*}", ""]],
            auth_pb2.AuthenticationRule(selector="first/last"),
            {},
        ],
        [[["post", "/v1/{name}", "data"]], None, {"name": "first/last"}],
        [
            [["post", "/v1/{selector}", "data"]],
            auth_pb2.AuthenticationRule(selector="first"),
            {},
        ],
        [[["post", "/v1/{first_name}", "data"]], None, {"last_name": "last"}],
        [
            [["post", "/v1/{first_name}", ""]],
            auth_pb2.AuthenticationRule(selector="first"),
            {},
        ],
    ],
)
def test_transcode_fails(http_options, message, request_kwargs):
    http_options, _ = helper_test_transcode(http_options, range(4))
    with pytest.raises(ValueError) as exc_info:
        path_template.transcode(http_options, message, **request_kwargs)
    assert str(exc_info.value).count("URI") == len(http_options)


def helper_test_transcode(http_options_list, expected_result_list):
    http_options = []
    for opt_list in http_options_list:
        http_option = {"method": opt_list[0], "uri": opt_list[1]}
        if opt_list[2]:
            http_option["body"] = opt_list[2]
        http_options.append(http_option)

    expected_result = {
        "method": expected_result_list[0],
        "uri": expected_result_list[1],
        "query_params": expected_result_list[3],
    }
    if expected_result_list[2]:
        expected_result["body"] = expected_result_list[2]
    return (http_options, expected_result)