curl_param_value_generators: Strengthen types.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg
2020-06-30 13:47:16 -07:00
parent 96cf928d27
commit a7bac82f2e

View File

@@ -8,25 +8,26 @@ from zerver.lib.events import do_events_register
from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_classes import ZulipTestCase
from zerver.models import Client, Message, UserGroup, UserPresence, get_realm from zerver.models import Client, Message, UserGroup, UserPresence, get_realm
GENERATOR_FUNCTIONS: Dict[str, Callable[..., Dict[Any, Any]]] = dict() GENERATOR_FUNCTIONS: Dict[str, Callable[[], Dict[str, object]]] = dict()
REGISTERED_GENERATOR_FUNCTIONS: Set[str] = set() REGISTERED_GENERATOR_FUNCTIONS: Set[str] = set()
CALLED_GENERATOR_FUNCTIONS: Set[str] = set() CALLED_GENERATOR_FUNCTIONS: Set[str] = set()
helpers = ZulipTestCase() helpers = ZulipTestCase()
def openapi_param_value_generator(endpoints: List[str]) -> Callable[[Callable[..., Any]], def openapi_param_value_generator(
Callable[..., Any]]: endpoints: List[str],
) -> Callable[[Callable[[], Dict[str, object]]], Callable[[], Dict[str, object]]]:
"""This decorator is used to register openapi param value genarator functions """This decorator is used to register openapi param value genarator functions
with endpoints. Example usage: with endpoints. Example usage:
@openapi_param_value_generator(["/messages/render:post"]) @openapi_param_value_generator(["/messages/render:post"])
def ... def ...
""" """
def wrapper(generator_func: Callable[..., Dict[Any, Any]]) -> Callable[..., Dict[Any, Any]]: def wrapper(generator_func: Callable[[], Dict[str, object]]) -> Callable[[], Dict[str, object]]:
@wraps(generator_func) @wraps(generator_func)
def _record_calls_wrapper(*args: Any, **kwargs: Any) -> Dict[Any, Any]: def _record_calls_wrapper() -> Dict[str, object]:
CALLED_GENERATOR_FUNCTIONS.add(generator_func.__name__) CALLED_GENERATOR_FUNCTIONS.add(generator_func.__name__)
return generator_func(*args, **kwargs) return generator_func()
REGISTERED_GENERATOR_FUNCTIONS.add(generator_func.__name__) REGISTERED_GENERATOR_FUNCTIONS.add(generator_func.__name__)
for endpoint in endpoints: for endpoint in endpoints:
@@ -35,13 +36,14 @@ def openapi_param_value_generator(endpoints: List[str]) -> Callable[[Callable[..
return _record_calls_wrapper return _record_calls_wrapper
return wrapper return wrapper
def patch_openapi_example_values(entry: str, params: List[Dict[str, Any]], def patch_openapi_example_values(
request_body: Optional[Dict[str, Any]]=None) \ entry: str, params: List[Dict[str, Any]],
-> Tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]]: request_body: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Dict[str, object]], Optional[Dict[str, object]]]:
if entry not in GENERATOR_FUNCTIONS: if entry not in GENERATOR_FUNCTIONS:
return params, request_body return params, request_body
func = GENERATOR_FUNCTIONS[entry] func = GENERATOR_FUNCTIONS[entry]
realm_example_values: Dict[str, Any] = func() realm_example_values: Dict[str, object] = func()
for param in params: for param in params:
param_name = param["name"] param_name = param["name"]
@@ -60,13 +62,13 @@ def patch_openapi_example_values(entry: str, params: List[Dict[str, Any]],
@openapi_param_value_generator(["/messages/{message_id}:get", "/messages/{message_id}/history:get", @openapi_param_value_generator(["/messages/{message_id}:get", "/messages/{message_id}/history:get",
"/messages/{message_id}:patch", "/messages/{message_id}:delete"]) "/messages/{message_id}:patch", "/messages/{message_id}:delete"])
def iago_message_id() -> Dict[str, int]: def iago_message_id() -> Dict[str, object]:
return { return {
"message_id": helpers.send_stream_message(helpers.example_user("iago"), "Denmark"), "message_id": helpers.send_stream_message(helpers.example_user("iago"), "Denmark"),
} }
@openapi_param_value_generator(["/messages/{message_id}/reactions:delete"]) @openapi_param_value_generator(["/messages/{message_id}/reactions:delete"])
def add_emoji_to_message() -> Dict[str, List[Dict[None, None]]]: def add_emoji_to_message() -> Dict[str, object]:
user_profile = helpers.example_user('iago') user_profile = helpers.example_user('iago')
# from OpenAPI format data in zulip.yaml # from OpenAPI format data in zulip.yaml
@@ -81,7 +83,7 @@ def add_emoji_to_message() -> Dict[str, List[Dict[None, None]]]:
return {} return {}
@openapi_param_value_generator(["/messages/flags:post"]) @openapi_param_value_generator(["/messages/flags:post"])
def update_flags_message_ids() -> Dict[str, List[int]]: def update_flags_message_ids() -> Dict[str, object]:
stream_name = "Venice" stream_name = "Venice"
helpers.subscribe(helpers.example_user("iago"), stream_name) helpers.subscribe(helpers.example_user("iago"), stream_name)
@@ -93,27 +95,27 @@ def update_flags_message_ids() -> Dict[str, List[int]]:
} }
@openapi_param_value_generator(["/mark_stream_as_read:post", "/users/me/{stream_id}/topics:get"]) @openapi_param_value_generator(["/mark_stream_as_read:post", "/users/me/{stream_id}/topics:get"])
def get_venice_stream_id() -> Dict[str, int]: def get_venice_stream_id() -> Dict[str, object]:
return { return {
"stream_id": helpers.get_stream_id("Venice"), "stream_id": helpers.get_stream_id("Venice"),
} }
@openapi_param_value_generator(["/streams/{stream_id}:patch"]) @openapi_param_value_generator(["/streams/{stream_id}:patch"])
def update_stream() -> Dict[str, Any]: def update_stream() -> Dict[str, object]:
stream = helpers.subscribe(helpers.example_user("iago"), "temp_stream 1") stream = helpers.subscribe(helpers.example_user("iago"), "temp_stream 1")
return { return {
"stream_id": stream.id, "stream_id": stream.id,
} }
@openapi_param_value_generator(["/streams/{stream_id}:delete"]) @openapi_param_value_generator(["/streams/{stream_id}:delete"])
def create_temp_stream_and_get_id() -> Dict[str, int]: def create_temp_stream_and_get_id() -> Dict[str, object]:
stream = helpers.subscribe(helpers.example_user("iago"), "temp_stream 2") stream = helpers.subscribe(helpers.example_user("iago"), "temp_stream 2")
return { return {
"stream_id": stream.id, "stream_id": stream.id,
} }
@openapi_param_value_generator(["/mark_topic_as_read:post"]) @openapi_param_value_generator(["/mark_topic_as_read:post"])
def get_denmark_stream_id_and_topic() -> Dict[str, Any]: def get_denmark_stream_id_and_topic() -> Dict[str, object]:
stream_name = "Denmark" stream_name = "Denmark"
topic_name = "Tivoli Gardens" topic_name = "Tivoli Gardens"
@@ -126,7 +128,7 @@ def get_denmark_stream_id_and_topic() -> Dict[str, Any]:
} }
@openapi_param_value_generator(["/users/me/subscriptions/properties:post"]) @openapi_param_value_generator(["/users/me/subscriptions/properties:post"])
def update_subscription_data() -> Dict[str, List[Dict[str, Any]]]: def update_subscription_data() -> Dict[str, object]:
profile = helpers.example_user("iago") profile = helpers.example_user("iago")
helpers.subscribe(profile, "Verona") helpers.subscribe(profile, "Verona")
helpers.subscribe(profile, "social") helpers.subscribe(profile, "social")
@@ -138,7 +140,7 @@ def update_subscription_data() -> Dict[str, List[Dict[str, Any]]]:
} }
@openapi_param_value_generator(["/users/me/subscriptions:delete"]) @openapi_param_value_generator(["/users/me/subscriptions:delete"])
def delete_subscription_data() -> Dict[str, List[Dict[None, None]]]: def delete_subscription_data() -> Dict[str, object]:
iago = helpers.example_user("iago") iago = helpers.example_user("iago")
zoe = helpers.example_user("ZOE") zoe = helpers.example_user("ZOE")
helpers.subscribe(iago, "Verona") helpers.subscribe(iago, "Verona")
@@ -148,7 +150,7 @@ def delete_subscription_data() -> Dict[str, List[Dict[None, None]]]:
return {} return {}
@openapi_param_value_generator(["/events:get"]) @openapi_param_value_generator(["/events:get"])
def get_events() -> Dict[str, Any]: def get_events() -> Dict[str, object]:
profile = helpers.example_user("iago") profile = helpers.example_user("iago")
helpers.subscribe(profile, "Verona") helpers.subscribe(profile, "Verona")
client = Client.objects.create(name="curl-test-client-1") client = Client.objects.create(name="curl-test-client-1")
@@ -160,7 +162,7 @@ def get_events() -> Dict[str, Any]:
} }
@openapi_param_value_generator(["/events:delete"]) @openapi_param_value_generator(["/events:delete"])
def delete_event_queue() -> Dict[str, Any]: def delete_event_queue() -> Dict[str, object]:
profile = helpers.example_user("iago") profile = helpers.example_user("iago")
client = Client.objects.create(name="curl-test-client-2") client = Client.objects.create(name="curl-test-client-2")
response = do_events_register(profile, client, event_types=['message']) response = do_events_register(profile, client, event_types=['message'])
@@ -170,40 +172,40 @@ def delete_event_queue() -> Dict[str, Any]:
} }
@openapi_param_value_generator(["/users/{email}/presence:get"]) @openapi_param_value_generator(["/users/{email}/presence:get"])
def get_user_presence() -> Dict[None, None]: def get_user_presence() -> Dict[str, object]:
iago = helpers.example_user("iago") iago = helpers.example_user("iago")
client = Client.objects.create(name="curl-test-client-3") client = Client.objects.create(name="curl-test-client-3")
update_user_presence(iago, client, timezone_now(), UserPresence.ACTIVE, False) update_user_presence(iago, client, timezone_now(), UserPresence.ACTIVE, False)
return {} return {}
@openapi_param_value_generator(["/users:post"]) @openapi_param_value_generator(["/users:post"])
def create_user() -> Dict[str, str]: def create_user() -> Dict[str, object]:
return { return {
"email": helpers.nonreg_email("test"), "email": helpers.nonreg_email("test"),
} }
@openapi_param_value_generator(["/user_groups/create:post"]) @openapi_param_value_generator(["/user_groups/create:post"])
def create_user_group_data() -> Dict[str, Any]: def create_user_group_data() -> Dict[str, object]:
return { return {
"members": [helpers.example_user("hamlet").id, helpers.example_user("othello").id], "members": [helpers.example_user("hamlet").id, helpers.example_user("othello").id],
} }
@openapi_param_value_generator(["/user_groups/{group_id}:patch", "/user_groups/{group_id}:delete"]) @openapi_param_value_generator(["/user_groups/{group_id}:patch", "/user_groups/{group_id}:delete"])
def get_temp_user_group_id() -> Dict[str, str]: def get_temp_user_group_id() -> Dict[str, object]:
user_group, _ = UserGroup.objects.get_or_create(name="temp", realm=get_realm("zulip")) user_group, _ = UserGroup.objects.get_or_create(name="temp", realm=get_realm("zulip"))
return { return {
"group_id": user_group.id, "group_id": user_group.id,
} }
@openapi_param_value_generator(["/realm/filters/{filter_id}:delete"]) @openapi_param_value_generator(["/realm/filters/{filter_id}:delete"])
def remove_realm_filters() -> Dict[str, Any]: def remove_realm_filters() -> Dict[str, object]:
filter_id = do_add_realm_filter(get_realm("zulip"), "#(?P<id>[0-9]{2,8})", "https://github.com/zulip/zulip/pull/%(id)s") filter_id = do_add_realm_filter(get_realm("zulip"), "#(?P<id>[0-9]{2,8})", "https://github.com/zulip/zulip/pull/%(id)s")
return { return {
"filter_id": filter_id, "filter_id": filter_id,
} }
@openapi_param_value_generator(["/realm/emoji/{emoji_name}:post", "/user_uploads:post"]) @openapi_param_value_generator(["/realm/emoji/{emoji_name}:post", "/user_uploads:post"])
def upload_custom_emoji() -> Dict[str, Any]: def upload_custom_emoji() -> Dict[str, object]:
return { return {
"filename": "zerver/tests/images/animated_img.gif", "filename": "zerver/tests/images/animated_img.gif",
} }