Files
2024-02-15 17:01:50 +00:00

152 lines
4.8 KiB
Python

from datetime import datetime, timezone
from sentry.models.commit import Commit
from sentry.models.commitauthor import CommitAuthor
from sentry.models.repository import Repository
from sentry.testutils.cases import APITestCase
from sentry.testutils.silo import region_silo_test
from sentry_plugins.bitbucket.testutils import PUSH_EVENT_EXAMPLE
BAD_IP = "109.111.111.10"
BITBUCKET_IP_IN_RANGE = "104.192.143.10"
BITBUCKET_IP = "34.198.178.64"
class WebhookTest(APITestCase):
def test_get(self):
project = self.project # force creation
url = f"/plugins/bitbucket/organizations/{project.organization.id}/webhook/"
response = self.client.get(url)
assert response.status_code == 405
def test_unregistered_event(self):
project = self.project # force creation
url = f"/plugins/bitbucket/organizations/{project.organization.id}/webhook/"
response = self.client.post(
path=url,
data=PUSH_EVENT_EXAMPLE,
content_type="application/json",
HTTP_X_EVENT_KEY="UnregisteredEvent",
REMOTE_ADDR=BITBUCKET_IP,
)
assert response.status_code == 204
response = self.client.post(
path=url,
data=PUSH_EVENT_EXAMPLE,
content_type="application/json",
HTTP_X_EVENT_KEY="UnregisteredEvent",
REMOTE_ADDR=BITBUCKET_IP_IN_RANGE,
)
assert response.status_code == 204
def test_invalid_signature_ip(self):
project = self.project # force creation
url = f"/plugins/bitbucket/organizations/{project.organization.id}/webhook/"
response = self.client.post(
path=url,
data=PUSH_EVENT_EXAMPLE,
content_type="application/json",
HTTP_X_EVENT_KEY="repo:push",
REMOTE_ADDR=BAD_IP,
)
assert response.status_code == 401
@region_silo_test
class PushEventWebhookTest(APITestCase):
def test_simple(self):
project = self.project # force creation
url = f"/plugins/bitbucket/organizations/{project.organization.id}/webhook/"
Repository.objects.create(
organization_id=project.organization.id,
external_id="{c78dfb25-7882-4550-97b1-4e0d38f32859}",
provider="bitbucket",
name="maxbittker/newsdiffs",
)
response = self.client.post(
path=url,
data=PUSH_EVENT_EXAMPLE,
content_type="application/json",
HTTP_X_EVENT_KEY="repo:push",
REMOTE_ADDR=BITBUCKET_IP,
)
assert response.status_code == 204
commit_list = list(
Commit.objects.filter(organization_id=project.organization_id)
.select_related("author")
.order_by("-date_added")
)
assert len(commit_list) == 1
commit = commit_list[0]
assert commit.key == "e0e377d186e4f0e937bdb487a23384fe002df649"
assert commit.message == "README.md edited online with Bitbucket"
assert commit.author.name == "Max Bittker"
assert commit.author.email == "[email protected]"
assert commit.author.external_id is None
assert commit.date_added == datetime(2017, 5, 24, 1, 5, 47, tzinfo=timezone.utc)
def test_anonymous_lookup(self):
project = self.project # force creation
url = f"/plugins/bitbucket/organizations/{project.organization.id}/webhook/"
Repository.objects.create(
organization_id=project.organization.id,
external_id="{c78dfb25-7882-4550-97b1-4e0d38f32859}",
provider="bitbucket",
name="maxbittker/newsdiffs",
)
CommitAuthor.objects.create(
external_id="bitbucket:baxterthehacker",
organization_id=project.organization_id,
email="[email protected]",
name="bĂ xterthehacker",
)
response = self.client.post(
path=url,
data=PUSH_EVENT_EXAMPLE,
content_type="application/json",
HTTP_X_EVENT_KEY="repo:push",
REMOTE_ADDR=BITBUCKET_IP,
)
assert response.status_code == 204
commit_list = list(
Commit.objects.filter(organization_id=project.organization_id)
.select_related("author")
.order_by("-date_added")
)
# should be skipping the #skipsentry commit
assert len(commit_list) == 1
commit = commit_list[0]
assert commit.key == "e0e377d186e4f0e937bdb487a23384fe002df649"
assert commit.message == "README.md edited online with Bitbucket"
assert commit.author.name == "Max Bittker"
assert commit.author.email == "[email protected]"
assert commit.author.external_id is None
assert commit.date_added == datetime(2017, 5, 24, 1, 5, 47, tzinfo=timezone.utc)