paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-10-25 09:03:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

test_classes: Type kwargs for client_get and friends.

Browse Source 
Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg
2021-12-16 23:14:22 -08:00
committed by Tim Abbott
parent 27977eddeb
commit dc18aadeb2
6 changed files with 177 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
zerver/lib/test_classes.py
View File

@@ -132,6 +132,12 @@ class UploadSerializeMixin(SerializeMixin):
super().setUpClass(*args, **kwargs)
# We could be more specific about which arguments are bool (Django's
# follow and secure, our intentionally_undocumented) and which are str
# (everything else), but explaining that to mypy is tedious.
ClientArg = Union[str, bool]
class ZulipTestCase(TestCase):
# Ensure that the test system just shows us diffs
maxDiff: Optional[int] = None
@@ -204,8 +210,9 @@ Output:
DEFAULT_SUBDOMAIN = "zulip"
TOKENIZED_NOREPLY_REGEX = settings.TOKENIZED_NOREPLY_EMAIL_ADDRESS.format(token="[a-z0-9_]{24}")
def set_http_headers(self, kwargs: Dict[str, Any]) -> None:
def set_http_headers(self, kwargs: Dict[str, ClientArg]) -> None:
if "subdomain" in kwargs:
assert isinstance(kwargs["subdomain"], str)
kwargs["HTTP_HOST"] = Realm.host_for_subdomain(kwargs["subdomain"])
del kwargs["subdomain"]
elif "HTTP_HOST" not in kwargs:
@@ -229,13 +236,13 @@ Output:
elif "HTTP_USER_AGENT" not in kwargs:
kwargs["HTTP_USER_AGENT"] = default_user_agent
def extract_api_suffix_url(self, url: str) -> Tuple[str, Dict[str, Any]]:
def extract_api_suffix_url(self, url: str) -> Tuple[str, Dict[str, List[str]]]:
"""
Function that extracts the URL after `/api/v1` or `/json` and also
returns the query data in the URL, if there is any.
"""
url_split = url.split("?")
data: Dict[str, Any] = {}
data = {}
if len(url_split) == 2:
data = urllib.parse.parse_qs(url_split[1])
url = url_split[0]
@@ -248,7 +255,7 @@ Output:
method: str,
result: HttpResponse,
data: Union[str, bytes, Dict[str, Any]],
http_headers: Dict[str, Any],
kwargs: Dict[str, ClientArg],
intentionally_undocumented: bool = False,
) -> None:
"""
@@ -276,6 +283,7 @@ Output:
content, url, method, str(result.status_code)
)
if response_validated:
http_headers = {k: v for k, v in kwargs.items() if isinstance(v, str)}
validate_request(
url,
method,
@@ -292,7 +300,7 @@ Output:
url: str,
info: Dict[str, Any] = {},
intentionally_undocumented: bool = False,
**kwargs: Any,
**kwargs: ClientArg,
) -> HttpResponse:
"""
We need to urlencode, since Django's function won't do it for us.
@@ -313,7 +321,7 @@ Output:
@instrument_url
def client_patch_multipart(
self, url: str, info: Dict[str, Any] = {}, **kwargs: Any
self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
"""
Use this for patch requests that have file uploads or
@@ -330,27 +338,31 @@ Output:
self.validate_api_response_openapi(url, "patch", result, info, kwargs)
return result
def json_patch(self, url: str, payload: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def json_patch(
self, url: str, payload: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
data = orjson.dumps(payload)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
return django_client.patch(url, data=data, content_type="application/json", **kwargs)
@instrument_url
def client_put(self, url: str, info: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def client_put(self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg) -> HttpResponse:
encoded = urllib.parse.urlencode(info)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
return django_client.put(url, encoded, **kwargs)
def json_put(self, url: str, payload: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def json_put(self, url: str, payload: Dict[str, Any] = {}, **kwargs: ClientArg) -> HttpResponse:
data = orjson.dumps(payload)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
return django_client.put(url, data=data, content_type="application/json", **kwargs)
@instrument_url
def client_delete(self, url: str, info: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def client_delete(
self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
encoded = urllib.parse.urlencode(info)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
@@ -359,14 +371,16 @@ Output:
return result
@instrument_url
def client_options(self, url: str, info: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def client_options(
self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
encoded = urllib.parse.urlencode(info)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
return django_client.options(url, encoded, **kwargs)
@instrument_url
def client_head(self, url: str, info: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
def client_head(self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg) -> HttpResponse:
encoded = urllib.parse.urlencode(info)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
@@ -377,9 +391,10 @@ Output:
self,
url: str,
info: Union[str, bytes, Dict[str, Any]] = {},
**kwargs: Any,
**kwargs: ClientArg,
) -> HttpResponse:
intentionally_undocumented: bool = kwargs.pop("intentionally_undocumented", False)
intentionally_undocumented = kwargs.pop("intentionally_undocumented", False)
assert isinstance(intentionally_undocumented, bool)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
result = django_client.post(url, info, **kwargs)
@@ -403,8 +418,9 @@ Output:
return match.func(req)
@instrument_url
def client_get(self, url: str, info: Dict[str, Any] = {}, **kwargs: Any) -> HttpResponse:
intentionally_undocumented: bool = kwargs.pop("intentionally_undocumented", False)
def client_get(self, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg) -> HttpResponse:
intentionally_undocumented = kwargs.pop("intentionally_undocumented", False)
assert isinstance(intentionally_undocumented, bool)
django_client = self.client # see WRAPPER_COMMENT
self.set_http_headers(kwargs)
result = django_client.get(url, info, **kwargs)
@@ -541,7 +557,7 @@ Output:
self.assertEqual(page_params["is_spectator"], False)
def login_with_return(
self, email: str, password: Optional[str] = None, **kwargs: Any
self, email: str, password: Optional[str] = None, **kwargs: ClientArg
) -> HttpResponse:
if password is None:
password = initial_password(email)
@@ -636,10 +652,10 @@ Output:
default_stream_groups: Sequence[str] = [],
source_realm_id: str = "",
key: Optional[str] = None,
realm_type: Optional[int] = Realm.ORG_TYPES["business"]["id"],
enable_marketing_emails: Optional[bool] = True,
is_demo_organization: Optional[bool] = False,
**kwargs: Any,
realm_type: int = Realm.ORG_TYPES["business"]["id"],
enable_marketing_emails: bool = True,
is_demo_organization: bool = False,
**kwargs: ClientArg,
) -> HttpResponse:
"""
Stage two of the two-step registration process.
@@ -734,33 +750,56 @@ Output:
credentials = f"{identifier}:{api_key}"
return "Basic " + base64.b64encode(credentials.encode()).decode()
def uuid_get(self, identifier: str, *args: Any, **kwargs: Any) -> HttpResponse:
def uuid_get(
self, identifier: str, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_uuid(identifier)
return self.client_get(*args, **kwargs)
return self.client_get(url, info, **kwargs)
def uuid_post(self, identifier: str, *args: Any, **kwargs: Any) -> HttpResponse:
def uuid_post(
self,
identifier: str,
url: str,
info: Union[str, bytes, Dict[str, Any]] = {},
**kwargs: ClientArg,
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_uuid(identifier)
return self.client_post(*args, **kwargs)
return self.client_post(url, info, **kwargs)
def api_get(self, user: UserProfile, *args: Any, **kwargs: Any) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_get(*args, **kwargs)
def api_post(
self, user: UserProfile, *args: Any, intentionally_undocumented: bool = False, **kwargs: Any
def api_get(
self, user: UserProfile, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_post(
*args, intentionally_undocumented=intentionally_undocumented, **kwargs
return self.client_get(url, info, **kwargs)
def api_post(
self,
user: UserProfile,
url: str,
info: Union[str, bytes, Dict[str, Any]] = {},
**kwargs: ClientArg,
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_post(url, info, **kwargs)
def api_patch(
self,
user: UserProfile,
url: str,
info: Dict[str, Any] = {},
intentionally_undocumented: bool = False,
**kwargs: ClientArg,
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_patch(
url, info, intentionally_undocumented=intentionally_undocumented, **kwargs
)
def api_patch(self, user: UserProfile, *args: Any, **kwargs: Any) -> HttpResponse:
def api_delete(
self, user: UserProfile, url: str, info: Dict[str, Any] = {}, **kwargs: ClientArg
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_patch(*args, **kwargs)
def api_delete(self, user: UserProfile, *args: Any, **kwargs: Any) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.client_delete(*args, **kwargs)
return self.client_delete(url, info, **kwargs)
def get_streams(self, user_profile: UserProfile) -> List[str]:
"""
@@ -1049,7 +1088,7 @@ Output:
invite_only: bool = False,
is_web_public: bool = False,
allow_fail: bool = False,
**kwargs: Any,
**kwargs: ClientArg,
) -> HttpResponse:
post_data = {
"subscriptions": orjson.dumps([{"name": stream} for stream in streams]).decode(),
@@ -1082,7 +1121,7 @@ Output:
user_profile: UserProfile,
url: str,
payload: Union[str, Dict[str, Any]],
**post_params: Any,
**post_params: ClientArg,
) -> Message:
"""
Send a webhook payload to the server, and verify that the
@@ -1561,9 +1600,20 @@ You can fix this by adding "{complete_event_type}" to ALL_EVENT_TYPES for this w
self.patch.start()
self.addCleanup(self.patch.stop)
def api_stream_message(self, user: UserProfile, *args: Any, **kwargs: Any) -> HttpResponse:
def api_stream_message(
self,
user: UserProfile,
fixture_name: str,
expected_topic: Optional[str] = None,
expected_message: Optional[str] = None,
content_type: Optional[str] = "application/json",
expect_noop: bool = False,
**kwargs: ClientArg,
) -> HttpResponse:
kwargs["HTTP_AUTHORIZATION"] = self.encode_user(user)
return self.check_webhook(*args, **kwargs)
return self.check_webhook(
fixture_name, expected_topic, expected_message, content_type, expect_noop, **kwargs
)
def check_webhook(
self,
@@ -1571,8 +1621,8 @@ You can fix this by adding "{complete_event_type}" to ALL_EVENT_TYPES for this w
expected_topic: Optional[str] = None,
expected_message: Optional[str] = None,
content_type: Optional[str] = "application/json",
expect_noop: Optional[bool] = False,
**kwargs: Any,
expect_noop: bool = False,
**kwargs: ClientArg,
) -> None:
"""
check_webhook is the main way to test "normal" webhooks that
@@ -1652,7 +1702,9 @@ one or more new messages.
fixture_name: str,
expected_message: str,
content_type: str = "application/json",
**kwargs: Any,
*,
sender: Optional[UserProfile] = None,
**kwargs: ClientArg,
) -> Message:
"""
For the rare cases that you are testing a webhook that sends
@@ -1668,8 +1720,9 @@ one or more new messages.
headers = get_fixture_http_headers(self.WEBHOOK_DIR_NAME, fixture_name)
headers = standardize_headers(headers)
kwargs.update(headers)
# The sender profile shouldn't be passed any further in kwargs, so we pop it.
sender = kwargs.pop("sender", self.test_user)
if sender is None:
sender = self.test_user
msg = self.send_webhook_payload(
sender,
@@ -1681,7 +1734,7 @@ one or more new messages.
return msg
def build_webhook_url(self, *args: Any, **kwargs: Any) -> str:
def build_webhook_url(self, *args: str, **kwargs: str) -> str:
url = self.URL_TEMPLATE
if url.find("api_key") >= 0:
api_key = get_api_key(self.test_user)

4
zerver/lib/test_helpers.py
View File

@@ -65,7 +65,7 @@ from zproject.backends import ExternalAuthDataDict, ExternalAuthResult
if TYPE_CHECKING:
# Avoid an import cycle; we only need these for type annotations.
from zerver.lib.test_classes import MigrationsTestCase, ZulipTestCase
from zerver.lib.test_classes import ClientArg, MigrationsTestCase, ZulipTestCase
class MockLDAP(fakeldap.MockLDAP):
@@ -359,7 +359,7 @@ def instrument_url(f: UrlFuncT) -> UrlFuncT:
else:
def wrapper(
self: "ZulipTestCase", url: str, info: object = {}, **kwargs: Any
self: "ZulipTestCase", url: str, info: object = {}, **kwargs: "ClientArg"
) -> HttpResponse:
start = time.time()
result = f(self, url, info, **kwargs)

2
zerver/openapi/openapi.py
View File

@@ -503,7 +503,7 @@ def validate_request(
url: str,
method: str,
data: Union[str, bytes, Dict[str, Any]],
http_headers: Dict[str, Any],
http_headers: Dict[str, str],
json_url: bool,
status_code: str,
intentionally_undocumented: bool = False,

6
zerver/tests/test_muting_topics.py
View File

@@ -1,5 +1,5 @@
from datetime import datetime, timedelta, timezone
from typing import Any, Dict
from typing import Any, Dict, List
from unittest import mock
from django.utils.timezone import now as timezone_now
@@ -90,7 +90,7 @@ class MutedTopicsTests(ZulipTestCase):
url = "/api/v1/users/me/subscriptions/muted_topics"
payloads = [
payloads: List[Dict[str, object]] = [
{"stream": stream.name, "topic": "Verona3", "op": "add"},
{"stream_id": stream.id, "topic": "Verona3", "op": "add"},
]
@@ -123,7 +123,7 @@ class MutedTopicsTests(ZulipTestCase):
recipient = stream.recipient
url = "/api/v1/users/me/subscriptions/muted_topics"
payloads = [
payloads: List[Dict[str, object]] = [
{"stream": stream.name, "topic": "vERONA3", "op": "remove"},
{"stream_id": stream.id, "topic": "vEroNA3", "op": "remove"},
]

1
zerver/tests/test_push_notifications.py
View File

@@ -871,6 +871,7 @@ class HandlePushNotificationTest(PushNotificationTest):
assert request.url is not None # allow mypy to infer url is present.
assert settings.PUSH_NOTIFICATION_BOUNCER_URL is not None
local_url = request.url.replace(settings.PUSH_NOTIFICATION_BOUNCER_URL, "")
assert isinstance(request.body, bytes)
result = self.uuid_post(
self.server_uuid, local_url, request.body, content_type="application/json"
)

152
zerver/webhooks/bitbucket2/tests.py
View File

@@ -1,4 +1,3 @@
from typing import Any, Dict
from unittest.mock import MagicMock, patch
from zerver.lib.test_classes import WebhookTestCase
@@ -117,172 +116,165 @@ class Bitbucket2HookTests(WebhookTestCase):
def test_bitbucket2_on_pull_request_created_event(self) -> None:
expected_message = "Tomasz created [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1) (assigned to Tomasz Kolek) from `new-branch` to `master`:\n\n~~~ quote\ndescription\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:created",
}
self.check_webhook(
"pull_request_created_or_updated", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_created_or_updated",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:created",
)
def test_bitbucket2_on_pull_request_created_without_reviewer_username_event(self) -> None:
expected_message = "Tomasz created [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1) (assigned to Tomasz Kolek) from `new-branch` to `master`:\n\n~~~ quote\ndescription\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:created",
}
self.check_webhook(
"pull_request_created_or_updated_without_username",
TOPIC_PR_EVENTS,
expected_message,
**kwargs,
HTTP_X_EVENT_KEY="pullrequest:created",
)
def test_bitbucket2_on_pull_request_created_with_custom_topic_in_url(self) -> None:
self.url = self.build_webhook_url(topic="notifications")
expected_topic = "notifications"
expected_message = "Tomasz created [PR #1 new commit](https://bitbucket.org/kolaszek/repository-name/pull-requests/1) (assigned to Tomasz Kolek) from `new-branch` to `master`:\n\n~~~ quote\ndescription\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:created",
}
self.check_webhook(
"pull_request_created_or_updated", expected_topic, expected_message, **kwargs
"pull_request_created_or_updated",
expected_topic,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:created",
)
def test_bitbucket2_on_pull_request_updated_event(self) -> None:
expected_message = "Tomasz updated [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1) (assigned to Tomasz Kolek) from `new-branch` to `master`:\n\n~~~ quote\ndescription\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:updated",
}
self.check_webhook(
"pull_request_created_or_updated", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_created_or_updated",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:updated",
)
def test_bitbucket2_on_pull_request_approved_event(self) -> None:
expected_message = "Tomasz approved [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1)."
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:approved",
}
self.check_webhook(
"pull_request_approved_or_unapproved", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_approved_or_unapproved",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:approved",
)
def test_bitbucket2_on_pull_request_approved_with_custom_topic_in_url(self) -> None:
self.url = self.build_webhook_url(topic="notifications")
expected_topic = "notifications"
expected_message = "Tomasz approved [PR #1 new commit](https://bitbucket.org/kolaszek/repository-name/pull-requests/1)."
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:approved",
}
self.check_webhook(
"pull_request_approved_or_unapproved", expected_topic, expected_message, **kwargs
"pull_request_approved_or_unapproved",
expected_topic,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:approved",
)
def test_bitbucket2_on_pull_request_unapproved_event(self) -> None:
expected_message = "Tomasz unapproved [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1)."
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:unapproved",
}
self.check_webhook(
"pull_request_approved_or_unapproved", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_approved_or_unapproved",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:unapproved",
)
def test_bitbucket2_on_pull_request_declined_event(self) -> None:
expected_message = "Tomasz rejected [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1)."
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:rejected",
}
self.check_webhook(
"pull_request_fulfilled_or_rejected", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_fulfilled_or_rejected",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:rejected",
)
def test_bitbucket2_on_pull_request_fulfilled_event(self) -> None:
expected_message = (
"Tomasz merged [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/1)."
)
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:fulfilled",
}
self.check_webhook(
"pull_request_fulfilled_or_rejected", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_fulfilled_or_rejected",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:fulfilled",
)
def test_bitbucket2_on_pull_request_comment_created_event(self) -> None:
expected_message = "Tomasz [commented](https://bitbucket.org/kolaszek/repository-name/pull-requests/3/_/diff#comment-20576503) on [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/3):\n\n~~~ quote\nComment1\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:comment_created",
}
self.check_webhook(
"pull_request_comment_action", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_comment_action",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:comment_created",
)
def test_bitbucket2_on_pull_request_comment_created_with_custom_topic_in_url(self) -> None:
self.url = self.build_webhook_url(topic="notifications")
expected_topic = "notifications"
expected_message = "Tomasz [commented](https://bitbucket.org/kolaszek/repository-name/pull-requests/3/_/diff#comment-20576503) on [PR #1 new commit](https://bitbucket.org/kolaszek/repository-name/pull-requests/3):\n\n~~~ quote\nComment1\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:comment_created",
}
self.check_webhook(
"pull_request_comment_action", expected_topic, expected_message, **kwargs
"pull_request_comment_action",
expected_topic,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:comment_created",
)
def test_bitbucket2_on_pull_request_comment_updated_event(self) -> None:
expected_message = "Tomasz updated a [comment](https://bitbucket.org/kolaszek/repository-name/pull-requests/3/_/diff#comment-20576503) on [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/3):\n\n~~~ quote\nComment1\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:comment_updated",
}
self.check_webhook(
"pull_request_comment_action", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_comment_action",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:comment_updated",
)
def test_bitbucket2_on_pull_request_comment_updated_with_custom_topic_in_url(self) -> None:
self.url = self.build_webhook_url(topic="notifications")
expected_topic = "notifications"
expected_message = "Tomasz updated a [comment](https://bitbucket.org/kolaszek/repository-name/pull-requests/3/_/diff#comment-20576503) on [PR #1 new commit](https://bitbucket.org/kolaszek/repository-name/pull-requests/3):\n\n~~~ quote\nComment1\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:comment_updated",
}
self.check_webhook(
"pull_request_comment_action", expected_topic, expected_message, **kwargs
"pull_request_comment_action",
expected_topic,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:comment_updated",
)
def test_bitbucket2_on_pull_request_comment_deleted_event(self) -> None:
expected_message = "Tomasz deleted a [comment](https://bitbucket.org/kolaszek/repository-name/pull-requests/3/_/diff#comment-20576503) on [PR #1](https://bitbucket.org/kolaszek/repository-name/pull-requests/3):\n\n~~~ quote\nComment1\n~~~"
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:comment_deleted",
}
self.check_webhook(
"pull_request_comment_action", TOPIC_PR_EVENTS, expected_message, **kwargs
"pull_request_comment_action",
TOPIC_PR_EVENTS,
expected_message,
HTTP_X_EVENT_KEY="pullrequest:comment_deleted",
)
def test_bitbucket2_on_repo_updated_event(self) -> None:
expected_message = "eeshangarg changed the website of the **new-name** repo to **http://zulipchat.com**.\neeshangarg changed the name of the **new-name** repo from **test-repo** to **new-name**.\neeshangarg changed the language of the **new-name** repo to **python**.\neeshangarg changed the full name of the **new-name** repo from **webhooktest/test-repo** to **webhooktest/new-name**.\neeshangarg changed the description of the **new-name** repo to **Random description.**"
expected_topic = "new-name"
kwargs: Dict[str, Any] = {"HTTP_X_EVENT_KEY": "repo:updated"}
self.check_webhook("repo_updated", expected_topic, expected_message, **kwargs)
self.check_webhook(
"repo_updated", expected_topic, expected_message, HTTP_X_EVENT_KEY="repo:updated"
)
def test_bitbucket2_on_push_one_tag_event(self) -> None:
expected_message = (
"Tomasz pushed tag [a](https://bitbucket.org/kolaszek/repository-name/commits/tag/a)."
)
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
self.check_webhook("push_one_tag", TOPIC, expected_message, **kwargs)
self.check_webhook(
"push_one_tag", TOPIC, expected_message, HTTP_X_EVENT_KEY="pullrequest:push"
)
def test_bitbucket2_on_push_remove_tag_event(self) -> None:
expected_message = (
"Tomasz removed tag [a](https://bitbucket.org/kolaszek/repository-name/commits/tag/a)."
)
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
self.check_webhook("push_remove_tag", TOPIC, expected_message, **kwargs)
self.check_webhook(
"push_remove_tag", TOPIC, expected_message, HTTP_X_EVENT_KEY="pullrequest:push"
)
def test_bitbucket2_on_push_more_than_one_tag_event(self) -> None:
expected_message = "Tomasz pushed tag [{name}](https://bitbucket.org/kolaszek/repository-name/commits/tag/{name})."
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
self.subscribe(self.test_user, self.STREAM_NAME)
payload = self.get_body("push_more_than_one_tag")
@@ -292,7 +284,7 @@ class Bitbucket2HookTests(WebhookTestCase):
self.url,
payload,
content_type="application/json",
**kwargs,
HTTP_X_EVENT_KEY="pullrequest:push",
)
msg = self.get_second_to_last_message()
@@ -312,10 +304,6 @@ class Bitbucket2HookTests(WebhookTestCase):
)
def test_bitbucket2_on_more_than_one_push_event(self) -> None:
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
self.subscribe(self.test_user, self.STREAM_NAME)
payload = self.get_body("more_than_one_push_event")
@@ -324,7 +312,7 @@ class Bitbucket2HookTests(WebhookTestCase):
self.url,
payload,
content_type="application/json",
**kwargs,
HTTP_X_EVENT_KEY="pullrequest:push",
)
msg = self.get_second_to_last_message()
@@ -345,9 +333,6 @@ class Bitbucket2HookTests(WebhookTestCase):
def test_bitbucket2_on_more_than_one_push_event_filtered_by_branches(self) -> None:
self.url = self.build_webhook_url(branches="master,development")
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
self.subscribe(self.test_user, self.STREAM_NAME)
payload = self.get_body("more_than_one_push_event")
@@ -357,7 +342,7 @@ class Bitbucket2HookTests(WebhookTestCase):
self.url,
payload,
content_type="application/json",
**kwargs,
HTTP_X_EVENT_KEY="pullrequest:push",
)
msg = self.get_second_to_last_message()
@@ -378,13 +363,12 @@ class Bitbucket2HookTests(WebhookTestCase):
def test_bitbucket2_on_more_than_one_push_event_filtered_by_branches_ignore(self) -> None:
self.url = self.build_webhook_url(branches="changes,development")
kwargs: Dict[str, Any] = {
"HTTP_X_EVENT_KEY": "pullrequest:push",
}
expected_message = (
"Tomasz pushed tag [a](https://bitbucket.org/kolaszek/repository-name/commits/tag/a)."
)
self.check_webhook("more_than_one_push_event", TOPIC, expected_message, **kwargs)
self.check_webhook(
"more_than_one_push_event", TOPIC, expected_message, HTTP_X_EVENT_KEY="pullrequest:push"
)
@patch("zerver.webhooks.bitbucket2.view.check_send_webhook_message")
def test_bitbucket2_on_push_event_filtered_by_branches_ignore(