Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_processes_resource_change_task_not_called_without_error_created(self, delay):
event = self.store_event(
data={
"message": "Foo bar",
"level": "error",
"exception": {"type": "Foo", "value": "shits on fiah yo"},
"timestamp": iso_format(timezone.now()),
},
project_id=self.project.id,
assert_no_errors=False,
)
self.create_service_hook(
project=self.project, organization=self.project.organization, actor=self.user, events=[]
)
post_process_group(
event=event, is_new=False, is_regression=False, is_new_group_environment=False
)
assert not delay.called
self.project1 = self.create_project(teams=[self.team1])
self.project2 = self.create_project(teams=[self.team2])
event1 = self.store_event(
data={
"fingerprint": ["group1"],
"timestamp": iso_format(before_now(seconds=3)),
"user": {"email": "foo@example.com"},
},
project_id=self.project1.id,
)
event2 = self.store_event(
data={
"fingerprint": ["group2"],
"timestamp": iso_format(before_now(seconds=2)),
"user": {"email": "bar@example.com"},
},
project_id=self.project1.id,
)
event3 = self.store_event(
data={
"fingerprint": ["group3"],
"timestamp": iso_format(before_now(seconds=1)),
"user": {"email": "foo@example.com"},
},
project_id=self.project2.id,
)
self.group1 = event1.group
self.group2 = event2.group
def test_date_params_included(self):
# Create an event that is outside the date range
self.store_event(
data={
"event_id": "d" * 32,
"message": "very bad",
"type": "default",
"platform": "python",
"timestamp": iso_format(before_now(days=2)),
"tags": {"foo": "1"},
},
project_id=self.project.id,
)
result = discover.get_pagination_ids(
self.event,
"foo:1",
{"project_id": [self.project.id], "end": self.min_ago, "start": self.day_ago},
)
assert result.previous == "a" * 32
assert result.next == "c" * 32
assert result.oldest == "a" * 32
assert result.latest == "c" * 32
def test_with_field_and_reference_event_invalid(self):
with self.feature("organizations:events-v2"):
response = self.client.get(
self.url,
format="json",
data={
"start": iso_format(self.day_ago),
"end": iso_format(self.day_ago + timedelta(hours=1, minutes=59)),
"interval": "1h",
"referenceEvent": "nope-invalid",
"yAxis": "count()",
},
)
assert response.status_code == 400, response.content
assert "reference" in response.content
def test_error_field(self):
data = load_data("php")
data["timestamp"] = iso_format(before_now(seconds=1))
event = self.store_event(data=data, project_id=self.project.id)
slug = "{}:{}".format(self.project.slug, event.event_id)
ref = discover.ReferenceEvent(
self.organization, slug, ["error.value", "error.type", "error.handled"]
)
result = discover.create_reference_event_conditions(ref)
assert result == [
["error.value", "=", "This is a test exception sent from the Raven CLI."],
["error.type", "=", "Exception"],
]
def test_all_projects(self):
project = self.create_project(organization=self.org)
self.event = self.store_event(
data={
"message": "other message",
"platform": "python",
"timestamp": iso_format(self.now - timedelta(minutes=1)),
},
project_id=project.id,
)
with self.feature("organizations:discover"):
url = reverse("sentry-api-0-discover-query", args=[self.org.slug])
response = self.client.post(
url,
{
"projects": [-1],
"fields": ["message", "platform.name"],
"range": "1d",
"orderby": "-timestamp",
"start": None,
"end": None,
},
def test_minimal(self):
kwargs = {
"message": "hello",
"tags": {"foo": "bar"},
"timestamp": iso_format(before_now(seconds=1)),
}
resp = self._postWithHeader(kwargs)
assert resp.status_code == 200, resp.content
event_id = json.loads(resp.content)["id"]
instance = self.get_event(event_id)
assert instance.message == "hello"
assert instance.data["logentry"] == {"formatted": "hello"}
assert instance.title == instance.data["title"] == "hello"
assert instance.location is instance.data.get("location", None) is None
assert tagstore.get_tag_key(self.project.id, None, "foo") is not None
assert tagstore.get_tag_value(self.project.id, None, "foo", "bar") is not None
def setUp(self):
super(SnubaDiscoverEventStorageTest, self).setUp()
self.min_ago = iso_format(before_now(minutes=1))
self.two_min_ago = iso_format(before_now(minutes=2))
self.project1 = self.create_project()
self.project2 = self.create_project()
self.event1 = self.store_event(
data={
"event_id": "a" * 32,
"type": "default",
"platform": "python",
"fingerprint": ["group1"],
"timestamp": self.two_min_ago,
"tags": {"foo": "1"},
},
project_id=self.project1.id,
)
self.event2 = self.store_event(
data={
def setUp(self):
super(SnubaEventStorageTest, self).setUp()
self.min_ago = iso_format(before_now(minutes=1))
self.two_min_ago = iso_format(before_now(minutes=2))
self.project1 = self.create_project()
self.project2 = self.create_project()
self.event1 = self.store_event(
data={
"event_id": "a" * 32,
"type": "default",
"platform": "python",
"fingerprint": ["group1"],
"timestamp": self.two_min_ago,
"tags": {"foo": "1"},
},
project_id=self.project1.id,
)
self.event2 = self.store_event(
def test_start_end_query(self):
project = self.create_project()
self.store_event(
data={"tags": {"foo": "bar"}, "timestamp": iso_format(before_now(days=15))},
project_id=project.id,
)
self.login_as(user=self.user)
url = reverse(
"sentry-api-0-project-tagkey-values",
kwargs={
"organization_slug": project.organization.slug,
"project_slug": project.slug,
"key": "foo",
},
)
response = self.client.get(
url