blob: 2c5216e06188a1d2f1bf98e702957b5bcbb6c657 [file] [log] [blame]
# 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
import mock
import pytest
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, request_kwargs, expected_result",
[
[
[["get", "/v1/no/template", ""]],
{"foo": "bar"},
["get", "/v1/no/template", {}, {"foo": "bar"}],
],
# Single templates
[
[["get", "/v1/{field}", ""]],
{"field": "parent"},
["get", "/v1/parent", {}, {}],
],
[
[["get", "/v1/{field.sub}", ""]],
{"field": {"sub": "parent"}, "foo": "bar"},
["get", "/v1/parent", {}, {"field": {}, "foo": "bar"}],
],
],
)
def test_transcode_base_case(http_options, request_kwargs, expected_result):
http_options, expected_result = helper_test_transcode(http_options, expected_result)
result = path_template.transcode(http_options, **request_kwargs)
assert result == expected_result
@pytest.mark.parametrize(
"http_options, request_kwargs, expected_result",
[
[
[["get", "/v1/{field.subfield}", ""]],
{"field": {"subfield": "parent"}, "foo": "bar"},
["get", "/v1/parent", {}, {"field": {}, "foo": "bar"}],
],
[
[["get", "/v1/{field.subfield.subsubfield}", ""]],
{"field": {"subfield": {"subsubfield": "parent"}}, "foo": "bar"},
["get", "/v1/parent", {}, {"field": {"subfield": {}}, "foo": "bar"}],
],
[
[["get", "/v1/{field.subfield1}/{field.subfield2}", ""]],
{"field": {"subfield1": "parent", "subfield2": "child"}, "foo": "bar"},
["get", "/v1/parent/child", {}, {"field": {}, "foo": "bar"}],
],
],
)
def test_transcode_subfields(http_options, request_kwargs, expected_result):
http_options, expected_result = helper_test_transcode(http_options, expected_result)
result = path_template.transcode(http_options, **request_kwargs)
assert result == expected_result
@pytest.mark.parametrize(
"http_options, request_kwargs, expected_result",
[
# Single segment wildcard
[
[["get", "/v1/{field=*}", ""]],
{"field": "parent"},
["get", "/v1/parent", {}, {}],
],
[
[["get", "/v1/{field=a/*/b/*}", ""]],
{"field": "a/parent/b/child", "foo": "bar"},
["get", "/v1/a/parent/b/child", {}, {"foo": "bar"}],
],
# Double segment wildcard
[
[["get", "/v1/{field=**}", ""]],
{"field": "parent/p1"},
["get", "/v1/parent/p1", {}, {}],
],
[
[["get", "/v1/{field=a/**/b/**}", ""]],
{"field": "a/parent/p1/b/child/c1", "foo": "bar"},
["get", "/v1/a/parent/p1/b/child/c1", {}, {"foo": "bar"}],
],
# Combined single and double segment wildcard
[
[["get", "/v1/{field=a/*/b/**}", ""]],
{"field": "a/parent/b/child/c1"},
["get", "/v1/a/parent/b/child/c1", {}, {}],
],
[
[["get", "/v1/{field=a/**/b/*}/v2/{name}", ""]],
{"field": "a/parent/p1/b/child", "name": "first", "foo": "bar"},
["get", "/v1/a/parent/p1/b/child/v2/first", {}, {"foo": "bar"}],
],
],
)
def test_transcode_with_wildcard(http_options, request_kwargs, expected_result):
http_options, expected_result = helper_test_transcode(http_options, expected_result)
result = path_template.transcode(http_options, **request_kwargs)
assert result == expected_result
@pytest.mark.parametrize(
"http_options, request_kwargs, expected_result",
[
# Single field body
[
[["post", "/v1/no/template", "data"]],
{"data": {"id": 1, "info": "some info"}, "foo": "bar"},
["post", "/v1/no/template", {"id": 1, "info": "some info"}, {"foo": "bar"}],
],
[
[["post", "/v1/{field=a/*}/b/{name=**}", "data"]],
{
"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"},
],
],
# Wildcard body
[
[["post", "/v1/{field=a/*}/b/{name=**}", "*"]],
{
"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"},
{},
],
],
],
)
def test_transcode_with_body(http_options, request_kwargs, expected_result):
http_options, expected_result = helper_test_transcode(http_options, expected_result)
result = path_template.transcode(http_options, **request_kwargs)
assert result == expected_result
@pytest.mark.parametrize(
"http_options, request_kwargs, expected_result",
[
# Additional bindings
[
[
["post", "/v1/{field=a/*}/b/{name=**}", "extra_data"],
["post", "/v1/{field=a/*}/b/{name=**}", "*"],
],
{
"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"},
{},
],
],
[
[
["get", "/v1/{field=a/*}/b/{name=**}", ""],
["get", "/v1/{field=a/*}/b/first/last", ""],
],
{"field": "a/parent", "foo": "bar"},
["get", "/v1/a/parent/b/first/last", {}, {"foo": "bar"}],
],
],
)
def test_transcode_with_additional_bindings(
http_options, request_kwargs, expected_result
):
http_options, expected_result = helper_test_transcode(http_options, expected_result)
result = path_template.transcode(http_options, **request_kwargs)
assert result == expected_result
@pytest.mark.parametrize(
"http_options, request_kwargs",
[
[[["get", "/v1/{name}", ""]], {"foo": "bar"}],
[[["get", "/v1/{name}", ""]], {"name": "first/last"}],
[[["get", "/v1/{name=mr/*/*}", ""]], {"name": "first/last"}],
[[["post", "/v1/{name}", "data"]], {"name": "first/last"}],
],
)
def test_transcode_fails(http_options, request_kwargs):
http_options, _ = helper_test_transcode(http_options, range(4))
with pytest.raises(ValueError):
path_template.transcode(http_options, **request_kwargs)
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)