3c407b5b-8434-4f25-9e12-005b14d23a79

Committed 07 Sep 2025 02:53PM UTC coverage: 33.615% (-8.7%) from 42.328%

Build # 3c407b5b-8434-4f25-9e12-005b14d23a79

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202509.1221'

Run Details

1 of 45 new or added lines in 2 files covered. (2.22%)

3993 existing lines in 129 files now uncovered.

16522 of 49150 relevant lines covered (33.62%)

0.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

36.54

src/pbn_api/tests/test_tasks.py

from unittest.mock import MagicMock

import pytest
from model_bakery import baker

from pbn_api.models import PBN_Export_Queue
from pbn_api.models.queue import SendStatus
from pbn_api.tasks import (
    kolejka_ponow_wysylke_prac_po_zalogowaniu,
    kolejka_wyczysc_wpisy_bez_rekordow,
    task_sprobuj_wyslac_do_pbn,
)

from django.contrib.contenttypes.models import ContentType

from bpp.models import Wydawnictwo_Ciagle


@pytest.mark.django_db(transaction=True)
def test_kolejka_wyczysc_wpisy_bez_rekordow():
    # Jeżeli NIE damy content_type i nie wpiszemy klasy, to losowo dostaniemy klase np bpp.cache.Autorzy
    # albo bpp.cache.Rekord, a tam klucz jest złożony i będzie, że nie może castnąć int do int[],
    # albo dostaniemy tabelę cache temporary (cpaq)
    baker.make(
        PBN_Export_Queue,
        object_id=0xBEEF,
        content_type=ContentType.objects.get_for_model(Wydawnictwo_Ciagle),
    )
    assert PBN_Export_Queue.objects.count() == 1
    kolejka_wyczysc_wpisy_bez_rekordow()
    assert PBN_Export_Queue.objects.count() == 0


@pytest.mark.django_db(transaction=True)
def test_kolejka_wyczysc_wpisy_bez_rekordow_missing_table():
    """Test that the function handles missing database tables gracefully."""
    from bpp.models.cache.punktacja import Cache_Punktacja_Autora_Sum

    # Use ContentType for Cache_Punktacja_Autora_Sum which uses temporary table
    content_type = ContentType.objects.get_for_model(Cache_Punktacja_Autora_Sum)

    # Create a PBN_Export_Queue entry pointing to the temporary table that doesn't exist
    baker.make(PBN_Export_Queue, object_id=0xBEEF, content_type=content_type)
    assert PBN_Export_Queue.objects.count() == 1

    # This should handle the missing table gracefully and delete the orphaned record
    kolejka_wyczysc_wpisy_bez_rekordow()
    assert PBN_Export_Queue.objects.count() == 0


@pytest.mark.django_db
def test_kolejka_ponow_wysylke_prac_po_zalogowaniu(wydawnictwo_ciagle, mocker):
    peq = baker.make(
        PBN_Export_Queue,
        rekord_do_wysylki=wydawnictwo_ciagle,
        retry_after_user_authorised=True,
        wysylke_zakonczono=None,
    )

    task_sprobuj_wyslac_do_pbn = mocker.patch(
        "pbn_api.tasks.task_sprobuj_wyslac_do_pbn"
    )

    kolejka_ponow_wysylke_prac_po_zalogowaniu(peq.zamowil.pk)

    task_sprobuj_wyslac_do_pbn.delay.assert_called_once()


@pytest.mark.parametrize(
    "send_status",
    [SendStatus.RETRY_MUCH_LATER, SendStatus.RETRY_SOON, SendStatus.RETRY_LATER],
)
def test_task_sprobuj_wyslac_do_pbn_retry_later(mocker, send_status):
    task_sprobuj_wyslac_do_pbn_apply_async = mocker.patch(
        "pbn_api.tasks.task_sprobuj_wyslac_do_pbn.apply_async"
    )

    wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")

    mock_peq = MagicMock()
    mock_peq.send_to_pbn.return_value = send_status

    wait_for_object.return_value = mock_peq

    task_sprobuj_wyslac_do_pbn(5)

    # Upewnij się, ze zadanie jest uruchamiane ponownie, ale ciut później
    task_sprobuj_wyslac_do_pbn_apply_async.assert_called_once()


@pytest.mark.parametrize(
    "send_status",
    [
        SendStatus.FINISHED_ERROR,
        SendStatus.FINISHED_OKAY,
        SendStatus.RETRY_AFTER_USER_AUTHORISED,
    ],
)
def test_task_sprobuj_wyslac_do_pbn_finished(mocker, send_status):
    mocker.patch("pbn_api.tasks.task_sprobuj_wyslac_do_pbn.apply_async")

    wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")

    mock_peq = MagicMock()
    mock_peq.send_to_pbn.return_value = send_status

    wait_for_object.return_value = mock_peq

    task_sprobuj_wyslac_do_pbn(5)


def test_task_sprobuj_wyslac_do_pbn_raises(mocker):
    wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")

    mock_peq = MagicMock()
    mock_peq.send_to_pbn.side_return_value = 0xBEEF

    wait_for_object.return_value = mock_peq

    with pytest.raises(NotImplementedError):
        task_sprobuj_wyslac_do_pbn(5)

1	from unittest.mock import MagicMock	1✔
2
3	import pytest	1✔
4	from model_bakery import baker	1✔
5
6	from pbn_api.models import PBN_Export_Queue	1✔
7	from pbn_api.models.queue import SendStatus	1✔
8	from pbn_api.tasks import (	1✔
9	kolejka_ponow_wysylke_prac_po_zalogowaniu,
10	kolejka_wyczysc_wpisy_bez_rekordow,
11	task_sprobuj_wyslac_do_pbn,
12	)
13
14	from django.contrib.contenttypes.models import ContentType	1✔
15
16	from bpp.models import Wydawnictwo_Ciagle	1✔
17
18
19	@pytest.mark.django_db(transaction=True)	1✔
20	def test_kolejka_wyczysc_wpisy_bez_rekordow():	1✔
21	# Jeżeli NIE damy content_type i nie wpiszemy klasy, to losowo dostaniemy klase np bpp.cache.Autorzy
22	# albo bpp.cache.Rekord, a tam klucz jest złożony i będzie, że nie może castnąć int do int[],
23	# albo dostaniemy tabelę cache temporary (cpaq)
24	baker.make(	×
25	PBN_Export_Queue,
26	object_id=0xBEEF,
27	content_type=ContentType.objects.get_for_model(Wydawnictwo_Ciagle),
28	)
UNCOV 29	assert PBN_Export_Queue.objects.count() == 1	×
UNCOV 30	kolejka_wyczysc_wpisy_bez_rekordow()	×
UNCOV 31	assert PBN_Export_Queue.objects.count() == 0	×
32
33
34	@pytest.mark.django_db(transaction=True)	1✔
35	def test_kolejka_wyczysc_wpisy_bez_rekordow_missing_table():	1✔
36	"""Test that the function handles missing database tables gracefully."""
UNCOV 37	from bpp.models.cache.punktacja import Cache_Punktacja_Autora_Sum	×
38
39	# Use ContentType for Cache_Punktacja_Autora_Sum which uses temporary table
UNCOV 40	content_type = ContentType.objects.get_for_model(Cache_Punktacja_Autora_Sum)	×
41
42	# Create a PBN_Export_Queue entry pointing to the temporary table that doesn't exist
UNCOV 43	baker.make(PBN_Export_Queue, object_id=0xBEEF, content_type=content_type)	×
UNCOV 44	assert PBN_Export_Queue.objects.count() == 1	×
45
46	# This should handle the missing table gracefully and delete the orphaned record
UNCOV 47	kolejka_wyczysc_wpisy_bez_rekordow()	×
UNCOV 48	assert PBN_Export_Queue.objects.count() == 0	×
49
50
51	@pytest.mark.django_db	1✔
52	def test_kolejka_ponow_wysylke_prac_po_zalogowaniu(wydawnictwo_ciagle, mocker):	1✔
53	peq = baker.make(	×
54	PBN_Export_Queue,
55	rekord_do_wysylki=wydawnictwo_ciagle,
56	retry_after_user_authorised=True,
57	wysylke_zakonczono=None,
58	)
59
60	task_sprobuj_wyslac_do_pbn = mocker.patch(	×
61	"pbn_api.tasks.task_sprobuj_wyslac_do_pbn"
62	)
63
UNCOV 64	kolejka_ponow_wysylke_prac_po_zalogowaniu(peq.zamowil.pk)	×
65
UNCOV 66	task_sprobuj_wyslac_do_pbn.delay.assert_called_once()	×
67
68
69	@pytest.mark.parametrize(	1✔
70	"send_status",
71	[SendStatus.RETRY_MUCH_LATER, SendStatus.RETRY_SOON, SendStatus.RETRY_LATER],
72	)
73	def test_task_sprobuj_wyslac_do_pbn_retry_later(mocker, send_status):	1✔
74	task_sprobuj_wyslac_do_pbn_apply_async = mocker.patch(	×
75	"pbn_api.tasks.task_sprobuj_wyslac_do_pbn.apply_async"
76	)
77
UNCOV 78	wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")	×
79
UNCOV 80	mock_peq = MagicMock()	×
81	mock_peq.send_to_pbn.return_value = send_status	×
82
UNCOV 83	wait_for_object.return_value = mock_peq	×
84
85	task_sprobuj_wyslac_do_pbn(5)	×
86
87	# Upewnij się, ze zadanie jest uruchamiane ponownie, ale ciut później
88	task_sprobuj_wyslac_do_pbn_apply_async.assert_called_once()	×
89
90
91	@pytest.mark.parametrize(	1✔
92	"send_status",
93	[
94	SendStatus.FINISHED_ERROR,
95	SendStatus.FINISHED_OKAY,
96	SendStatus.RETRY_AFTER_USER_AUTHORISED,
97	],
98	)
99	def test_task_sprobuj_wyslac_do_pbn_finished(mocker, send_status):	1✔
UNCOV 100	mocker.patch("pbn_api.tasks.task_sprobuj_wyslac_do_pbn.apply_async")	×
101
UNCOV 102	wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")	×
103
UNCOV 104	mock_peq = MagicMock()	×
UNCOV 105	mock_peq.send_to_pbn.return_value = send_status	×
106
UNCOV 107	wait_for_object.return_value = mock_peq	×
108
UNCOV 109	task_sprobuj_wyslac_do_pbn(5)	×
110
111
112	def test_task_sprobuj_wyslac_do_pbn_raises(mocker):	1✔
UNCOV 113	wait_for_object = mocker.patch("pbn_api.tasks.wait_for_object")	×
114
UNCOV 115	mock_peq = MagicMock()	×
UNCOV 116	mock_peq.send_to_pbn.side_return_value = 0xBEEF	×
117
UNCOV 118	wait_for_object.return_value = mock_peq	×
119
UNCOV 120	with pytest.raises(NotImplementedError):	×
UNCOV 121	task_sprobuj_wyslac_do_pbn(5)	×

iplweb / bpp / 3c407b5b-8434-4f25-9e12-005b14d23a79

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous