Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
work_item_id = 33
num_groups = 5
external_issue = ExternalIssue.objects.create(
organization_id=self.organization.id, integration_id=self.model.id, key=work_item_id
)
groups = [
self.create_linked_group(external_issue, self.project, GroupStatus.RESOLVED)
for _ in range(num_groups)
]
# Change so that state is changing from resolved to unresolved
work_item = self.set_workitem_state("Resolved", "Active")
with self.feature("organizations:integrations-issue-sync"):
resp = self.client.post(
absolute_uri("/extensions/vsts/issue-updated/"),
data=work_item,
HTTP_SHARED_SECRET=self.shared_secret,
)
assert resp.status_code == 200
group_ids = [g.id for g in groups]
assert (
len(Group.objects.filter(id__in=group_ids, status=GroupStatus.UNRESOLVED))
== num_groups
)
assert len(Activity.objects.filter(group_id__in=group_ids)) == num_groups
def get_group_description(self, request, group, event):
referrer = self.get_conf_key() + "_plugin"
output = [absolute_uri(group.get_absolute_url(params={"referrer": referrer}))]
body = self.get_group_body(request, group, event)
if body:
output.extend(["", "```", body, "```"])
return "\n".join(output)
def create_repository(self, organization, data, actor=None):
if actor is None:
raise NotImplementedError("Cannot create a repository anonymously")
client = self.get_client(actor)
try:
resp = client.create_hook(
data["name"],
{
"description": "sentry-bitbucket-repo-hook",
"url": absolute_uri(
"/plugins/bitbucket/organizations/{}/webhook/".format(organization.id)
),
"active": True,
"events": ["repo:push"],
},
)
except Exception as e:
self.raise_error(e, identity=client.auth)
else:
return {
"name": data["name"],
"external_id": data["external_id"],
"url": "https://bitbucket.org/{}".format(data["name"]),
"config": {"name": data["name"], "webhook_id": resp["uuid"]},
}
def create_repository(self, organization, data, actor=None):
if actor is None:
raise NotImplementedError("Cannot create a repository anonymously")
client = self.get_client(actor)
try:
resp = client.create_hook(
data["name"],
{
"description": "sentry-bitbucket-repo-hook",
"url": absolute_uri(
"/plugins/bitbucket/organizations/{}/webhook/".format(organization.id)
),
"active": True,
"events": ["repo:push"],
},
)
except Exception as e:
self.raise_error(e, identity=client.auth)
else:
return {
"name": data["name"],
"external_id": data["external_id"],
"url": "https://bitbucket.org/{}".format(data["name"]),
"config": {"name": data["name"], "webhook_id": resp["uuid"]},
}
def get_notification_settings_url(self):
return absolute_uri(reverse('sentry-account-settings-notifications'))
def generate_email_context(self, status):
trigger = self.action.alert_rule_trigger
alert_rule = trigger.alert_rule
return {
"link": absolute_uri(
reverse(
"sentry-incident",
kwargs={
"organization_slug": self.incident.organization.slug,
"incident_id": self.incident.identifier,
},
)
),
"rule_link": absolute_uri(
reverse(
"sentry-alert-rule",
kwargs={
"organization_slug": self.incident.organization.slug,
"alert_rule_id": self.action.alert_rule_trigger.alert_rule_id,
},
)
),
"incident_name": self.incident.title,
"aggregate": self.query_aggregations_display[QueryAggregations(alert_rule.aggregation)],
"query": alert_rule.query,
"threshold": trigger.alert_threshold
if status == TriggerStatus.ACTIVE
else trigger.resolve_threshold,
"status": self.status_display[status],
}
"signMetadata": avd.get("metadata_signed", False),
"wantMessagesSigned": avd.get("want_message_signed", False),
"wantAssertionsSigned": avd.get("want_assertion_signed", False),
"wantAssertionsEncrypted": avd.get("want_assertion_encrypted", False),
"signatureAlgorithm": avd.get("signature_algorithm", OneLogin_Saml2_Constants.RSA_SHA256),
"digestAlgorithm": avd.get("digest_algorithm", OneLogin_Saml2_Constants.SHA256),
"wantNameId": False,
"requestedAuthnContext": [
"urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport",
"urn:federation:authentication:windows",
],
}
# TODO(epurkhiser): This is also available in the helper and should probably come from there.
acs_url = absolute_uri(reverse("sentry-auth-organization-saml-acs", args=[org]))
sls_url = absolute_uri(reverse("sentry-auth-organization-saml-sls", args=[org]))
metadata_url = absolute_uri(reverse("sentry-auth-organization-saml-metadata", args=[org]))
saml_config = {
"strict": True,
"sp": {
"entityId": metadata_url,
"assertionConsumerService": {
"url": acs_url,
"binding": OneLogin_Saml2_Constants.BINDING_HTTP_POST,
},
"singleLogoutService": {
"url": sls_url,
"binding": OneLogin_Saml2_Constants.BINDING_HTTP_REDIRECT,
},
},
"security": security_config,
def _get_webhook_url(project, plugin_id, token):
return absolute_uri(
reverse(
"sentry-release-hook",
kwargs={
"plugin_id": plugin_id,
"project_id": project.id,
"signature": _get_signature(project.id, plugin_id, token),
},
def get_link_existing_issue_fields(self, request, group, event, **kwargs):
return [
{
"name": "issue_id",
"label": "Issue #",
"default": "",
"placeholder": "e.g. 1543",
"type": "text",
},
{
"name": "comment",
"label": "Comment",
"default": absolute_uri(
group.get_absolute_url(params={"referrer": "gitlab_plugin"})
),
"type": "textarea",
"help": ("Leave blank if you don't want to " "add a comment to the GitLab issue."),
"required": False,
},
def system_origin():
from sentry.utils.http import absolute_uri, origin_from_url
return origin_from_url(absolute_uri())