mirror of
https://github.com/hanzoai/sentinel.git
synced 2026-09-29 22:34:33 +00:00
112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
from sentry.db.models.query import in_iexact
|
|
from sentry.models.organization import Organization
|
|
from sentry.models.user import User
|
|
from sentry.models.userreport import UserReport
|
|
from sentry.testutils.cases import TestCase
|
|
from sentry.testutils.silo import no_silo_test
|
|
from sentry.utils.query import (
|
|
RangeQuerySetWrapper,
|
|
RangeQuerySetWrapperWithProgressBar,
|
|
RangeQuerySetWrapperWithProgressBarApprox,
|
|
bulk_delete_objects,
|
|
)
|
|
|
|
|
|
class InIexactQueryTest(TestCase):
|
|
def test_basic(self):
|
|
self.create_organization(slug="SlugA")
|
|
self.create_organization(slug="slugB")
|
|
self.create_organization(slug="slugc")
|
|
|
|
assert Organization.objects.filter(in_iexact("slug", ["sluga", "slugb"])).count() == 2
|
|
assert Organization.objects.filter(in_iexact("slug", ["slugC"])).count() == 1
|
|
assert Organization.objects.filter(in_iexact("slug", [])).count() == 0
|
|
|
|
|
|
@no_silo_test
|
|
class RangeQuerySetWrapperTest(TestCase):
|
|
range_wrapper = RangeQuerySetWrapper
|
|
|
|
def test_basic(self):
|
|
total = 10
|
|
|
|
for _ in range(total):
|
|
self.create_user()
|
|
|
|
qs = User.objects.all()
|
|
|
|
assert len(list(self.range_wrapper(qs, step=2))) == total
|
|
assert len(list(self.range_wrapper(qs, limit=5))) == 5
|
|
|
|
def test_loop_and_delete(self):
|
|
total = 10
|
|
for _ in range(total):
|
|
self.create_user()
|
|
|
|
qs = User.objects.all()
|
|
|
|
for user in self.range_wrapper(qs, step=2):
|
|
user.delete()
|
|
|
|
assert User.objects.all().count() == 0
|
|
|
|
def test_empty(self):
|
|
qs = User.objects.all()
|
|
assert len(list(self.range_wrapper(qs, step=2))) == 0
|
|
|
|
|
|
@no_silo_test
|
|
class RangeQuerySetWrapperWithProgressBarTest(RangeQuerySetWrapperTest):
|
|
range_wrapper = RangeQuerySetWrapperWithProgressBar
|
|
|
|
|
|
@no_silo_test
|
|
class RangeQuerySetWrapperWithProgressBarApproxTest(RangeQuerySetWrapperTest):
|
|
range_wrapper = RangeQuerySetWrapperWithProgressBarApprox
|
|
|
|
|
|
class BulkDeleteObjectsTest(TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
UserReport.objects.all().delete()
|
|
|
|
def test_basic(self):
|
|
total = 10
|
|
records = []
|
|
for i in range(total):
|
|
records.append(self.create_userreport(project=self.project, event_id=str(i) * 32))
|
|
|
|
result = bulk_delete_objects(UserReport, id__in=[r.id for r in records])
|
|
assert result, "Could be more work to do"
|
|
assert len(UserReport.objects.all()) == 0
|
|
|
|
def test_basic_tuple(self):
|
|
total = 10
|
|
records = []
|
|
for i in range(total):
|
|
records.append(self.create_userreport(project=self.project, event_id=str(i) * 32))
|
|
|
|
result = bulk_delete_objects(UserReport, id__in=tuple([r.id for r in records]))
|
|
assert result, "Could be more work to do"
|
|
assert len(UserReport.objects.all()) == 0
|
|
|
|
def test_basic_set(self):
|
|
total = 10
|
|
records = []
|
|
for i in range(total):
|
|
records.append(self.create_userreport(project=self.project, event_id=str(i) * 32))
|
|
|
|
result = bulk_delete_objects(UserReport, id__in={r.id for r in records})
|
|
assert result, "Could be more work to do"
|
|
assert len(UserReport.objects.all()) == 0
|
|
|
|
def test_limiting(self):
|
|
total = 10
|
|
records = []
|
|
for i in range(total):
|
|
records.append(self.create_userreport(project=self.project, event_id=str(i) * 32))
|
|
|
|
result = bulk_delete_objects(UserReport, id__in=[r.id for r in records], limit=5)
|
|
assert result, "Still more work to do"
|
|
assert len(UserReport.objects.all()) == 5
|