Skip to content

Commit f0dfc06

Browse files
alexander-alderman-webbalexander-alderman-webb
committed
.
1 parent 279c244 commit f0dfc06

File tree

1 file changed

+249
-0
lines changed

1 file changed

+249
-0
lines changed

‎tests/integrations/django/test_db_transactions.py‎

Lines changed: 249 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,131 @@ def test_db_no_autocommit_executemany(sentry_init, client, capture_events):
150150
) or conn_params.get("dbname")
151151

152152

153+
@pytest.mark.forked
154+
@pytest_mark_django_db_decorator(transaction=True)
155+
def test_db_no_autocommit_rollback_execute(sentry_init, client, capture_events):
156+
sentry_init(
157+
integrations=[DjangoIntegration()],
158+
traces_sample_rate=1.0,
159+
)
160+
161+
if "postgres" not in connections:
162+
pytest.skip("postgres tests disabled")
163+
164+
# trigger Django to open a new connection by marking the existing one as None.
165+
connections["postgres"].connection = None
166+
167+
events = capture_events()
168+
169+
client.get(reverse("postgres_insert_orm_no_autocommit_rollback"))
170+
171+
(event,) = events
172+
173+
# Ensure operation is rolled back
174+
assert not User.objects.using("postgres").exists()
175+
176+
assert event["contexts"]["trace"]["origin"] == "auto.http.django"
177+
178+
commit_spans = [
179+
span
180+
for span in event["spans"]
181+
if span["data"].get(SPANDATA.DB_OPERATION) == DBOPERATION.ROLLBACK
182+
]
183+
assert len(commit_spans) == 1
184+
commit_span = commit_spans[0]
185+
assert commit_span["origin"] == "auto.db.django"
186+
187+
# Verify other database attributes
188+
assert commit_span["data"].get(SPANDATA.DB_SYSTEM) == "postgresql"
189+
conn_params = connections["postgres"].get_connection_params()
190+
assert commit_span["data"].get(SPANDATA.DB_NAME) is not None
191+
assert commit_span["data"].get(SPANDATA.DB_NAME) == conn_params.get(
192+
"database"
193+
) or conn_params.get("dbname")
194+
assert commit_span["data"].get(SPANDATA.SERVER_ADDRESS) == os.environ.get(
195+
"SENTRY_PYTHON_TEST_POSTGRES_HOST", "localhost"
196+
)
197+
assert commit_span["data"].get(SPANDATA.SERVER_PORT) == os.environ.get(
198+
"SENTRY_PYTHON_TEST_POSTGRES_PORT", "5432"
199+
)
200+
201+
202+
@pytest.mark.forked
203+
@pytest_mark_django_db_decorator(transaction=True)
204+
def test_db_no_autocommit_rollback_executemany(sentry_init, client, capture_events):
205+
sentry_init(
206+
integrations=[DjangoIntegration()],
207+
traces_sample_rate=1.0,
208+
)
209+
210+
events = capture_events()
211+
212+
with start_transaction(name="test_transaction"):
213+
from django.db import connection, transaction
214+
215+
cursor = connection.cursor()
216+
217+
query = """INSERT INTO auth_user (
218+
password,
219+
is_superuser,
220+
username,
221+
first_name,
222+
last_name,
223+
email,
224+
is_staff,
225+
is_active,
226+
date_joined
227+
)
228+
VALUES ('password', false, %s, %s, %s, %s, false, true, %s);"""
229+
230+
query_list = (
231+
(
232+
"user1",
233+
"John",
234+
"Doe",
235+
"[email protected]",
236+
datetime(1970, 1, 1),
237+
),
238+
(
239+
"user2",
240+
"Max",
241+
"Mustermann",
242+
"[email protected]",
243+
datetime(1970, 1, 1),
244+
),
245+
)
246+
247+
transaction.set_autocommit(False)
248+
cursor.executemany(query, query_list)
249+
transaction.rollback()
250+
transaction.set_autocommit(True)
251+
252+
(event,) = events
253+
254+
# Ensure operation is rolled back
255+
assert not User.objects.exists()
256+
257+
assert event["contexts"]["trace"]["origin"] == "manual"
258+
assert event["spans"][0]["origin"] == "auto.db.django"
259+
260+
commit_spans = [
261+
span
262+
for span in event["spans"]
263+
if span["data"].get(SPANDATA.DB_OPERATION) == DBOPERATION.ROLLBACK
264+
]
265+
assert len(commit_spans) == 1
266+
commit_span = commit_spans[0]
267+
assert commit_span["origin"] == "auto.db.django"
268+
269+
# Verify other database attributes
270+
assert commit_span["data"].get(SPANDATA.DB_SYSTEM) == "sqlite"
271+
conn_params = connection.get_connection_params()
272+
assert commit_span["data"].get(SPANDATA.DB_NAME) is not None
273+
assert commit_span["data"].get(SPANDATA.DB_NAME) == conn_params.get(
274+
"database"
275+
) or conn_params.get("dbname")
276+
277+
153278
@pytest.mark.forked
154279
@pytest_mark_django_db_decorator(transaction=True)
155280
def test_db_atomic_execute(sentry_init, client, capture_events):
@@ -270,3 +395,127 @@ def test_db_atomic_executemany(sentry_init, client, capture_events):
270395
assert commit_span["data"].get(SPANDATA.DB_NAME) == conn_params.get(
271396
"database"
272397
) or conn_params.get("dbname")
398+
399+
400+
@pytest.mark.forked
401+
@pytest_mark_django_db_decorator(transaction=True)
402+
def test_db_atomic_rollback_execute(sentry_init, client, capture_events):
403+
sentry_init(
404+
integrations=[DjangoIntegration()],
405+
send_default_pii=True,
406+
traces_sample_rate=1.0,
407+
)
408+
409+
if "postgres" not in connections:
410+
pytest.skip("postgres tests disabled")
411+
412+
# trigger Django to open a new connection by marking the existing one as None.
413+
connections["postgres"].connection = None
414+
415+
events = capture_events()
416+
417+
client.get(reverse("postgres_insert_orm_atomic_rollback"))
418+
419+
(event,) = events
420+
421+
# Ensure operation is rolled back
422+
assert not User.objects.using("postgres").exists()
423+
424+
assert event["contexts"]["trace"]["origin"] == "auto.http.django"
425+
426+
commit_spans = [
427+
span
428+
for span in event["spans"]
429+
if span["data"].get(SPANDATA.DB_OPERATION) == DBOPERATION.ROLLBACK
430+
]
431+
assert len(commit_spans) == 1
432+
commit_span = commit_spans[0]
433+
assert commit_span["origin"] == "auto.db.django"
434+
435+
# Verify other database attributes
436+
assert commit_span["data"].get(SPANDATA.DB_SYSTEM) == "postgresql"
437+
conn_params = connections["postgres"].get_connection_params()
438+
assert commit_span["data"].get(SPANDATA.DB_NAME) is not None
439+
assert commit_span["data"].get(SPANDATA.DB_NAME) == conn_params.get(
440+
"database"
441+
) or conn_params.get("dbname")
442+
assert commit_span["data"].get(SPANDATA.SERVER_ADDRESS) == os.environ.get(
443+
"SENTRY_PYTHON_TEST_POSTGRES_HOST", "localhost"
444+
)
445+
assert commit_span["data"].get(SPANDATA.SERVER_PORT) == os.environ.get(
446+
"SENTRY_PYTHON_TEST_POSTGRES_PORT", "5432"
447+
)
448+
449+
450+
@pytest.mark.forked
451+
@pytest_mark_django_db_decorator(transaction=True)
452+
def test_db_atomic_rollback_executemany(sentry_init, client, capture_events):
453+
sentry_init(
454+
integrations=[DjangoIntegration()],
455+
send_default_pii=True,
456+
traces_sample_rate=1.0,
457+
)
458+
459+
events = capture_events()
460+
461+
with start_transaction(name="test_transaction"):
462+
from django.db import connection, transaction
463+
464+
with transaction.atomic():
465+
cursor = connection.cursor()
466+
467+
query = """INSERT INTO auth_user (
468+
password,
469+
is_superuser,
470+
username,
471+
first_name,
472+
last_name,
473+
email,
474+
is_staff,
475+
is_active,
476+
date_joined
477+
)
478+
VALUES ('password', false, %s, %s, %s, %s, false, true, %s);"""
479+
480+
query_list = (
481+
(
482+
"user1",
483+
"John",
484+
"Doe",
485+
"[email protected]",
486+
datetime(1970, 1, 1),
487+
),
488+
(
489+
"user2",
490+
"Max",
491+
"Mustermann",
492+
"[email protected]",
493+
datetime(1970, 1, 1),
494+
),
495+
)
496+
cursor.executemany(query, query_list)
497+
transaction.set_rollback(True)
498+
499+
(event,) = events
500+
501+
# Ensure operation is rolled back
502+
assert not User.objects.exists()
503+
504+
assert event["contexts"]["trace"]["origin"] == "manual"
505+
506+
commit_spans = [
507+
span
508+
for span in event["spans"]
509+
if span["data"].get(SPANDATA.DB_OPERATION) == DBOPERATION.ROLLBACK
510+
]
511+
assert len(commit_spans) == 1
512+
commit_span = commit_spans[0]
513+
assert commit_span["origin"] == "auto.db.django"
514+
515+
# Verify other database attributes
516+
assert commit_span["data"].get(SPANDATA.DB_SYSTEM) == "sqlite"
517+
conn_params = connection.get_connection_params()
518+
assert commit_span["data"].get(SPANDATA.DB_NAME) is not None
519+
assert commit_span["data"].get(SPANDATA.DB_NAME) == conn_params.get(
520+
"database"
521+
) or conn_params.get("dbname")

0 commit comments

Comments
 (0)