7d423bde-8caf-4c8b-8bf5-12ad3f831949

Committed 22 Aug 2025 01:06PM UTC coverage: 45.781% (+0.06%) from 45.719%

Build # 7d423bde-8caf-4c8b-8bf5-12ad3f831949

Build Type

push

circleci

Committed by

mpasternak

Commit Message

Merge branch 'release/v202508.1201'

Run Details

11 of 18 new or added lines in 3 files covered. (61.11%)

1245 existing lines in 104 files now uncovered.

17481 of 38184 relevant lines covered (45.78%)

1.18 hits per line

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

54.55

src/pbn_api/models/sentdata.py

from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.db.models import JSONField

from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType

from bpp import const
from bpp.models import LinkDoPBNMixin


class SentDataManager(models.Manager):
    def get_for_rec(self, rec):
        return self.get(
            object_id=rec.pk, content_type=ContentType.objects.get_for_model(rec)
        )

    def check_if_needed(self, rec, data: dict):
        try:
            sd = self.get_for_rec(rec)
        except SentData.DoesNotExist:
            return True

        if sd.data_sent != data:
            return True

        if not sd.uploaded_okay:
            return True

        return False

    def updated(
        self, rec, data: dict, pbn_uid_id=None, uploaded_okay=True, exception=None
    ):
        try:
            sd = self.get_for_rec(rec)
        except SentData.DoesNotExist:
            self.create(
                object=rec,
                data_sent=data,
                uploaded_okay=uploaded_okay,
                pbn_uid_id=pbn_uid_id,
                exception=exception,
            )
            return

        sd.data_sent = data
        sd.uploaded_okay = uploaded_okay
        sd.exception = exception
        sd.pbn_uid_id = pbn_uid_id
        sd.save()

    def ids_for_model(self, model):
        return self.filter(content_type=ContentType.objects.get_for_model(model))

    def bad_uploads(self, model):
        return (
            self.ids_for_model(model)
            .filter(uploaded_okay=False)
            .values_list("object_id", flat=True)
            .distinct()
        )


class SentData(LinkDoPBNMixin, models.Model):
    url_do_pbn = const.LINK_PBN_DO_PUBLIKACJI
    content_type = models.ForeignKey(
        "contenttypes.ContentType", on_delete=models.CASCADE
    )
    object_id = models.PositiveIntegerField(db_index=True)

    object = GenericForeignKey()

    data_sent = JSONField("Wysłane dane")
    last_updated_on = models.DateTimeField("Data operacji", auto_now=True)

    uploaded_okay = models.BooleanField(
        "Wysłano poprawnie", default=True, db_index=True
    )
    exception = models.TextField("Kod błędu", max_length=65535, blank=True, null=True)

    pbn_uid = models.ForeignKey(
        "pbn_api.Publication",
        verbose_name="Publikacja z PBN",
        blank=True,
        null=True,
        on_delete=models.SET_NULL,
    )

    typ_rekordu = models.CharField(max_length=50, blank=True, null=True)

    objects = SentDataManager()

    class Meta:
        verbose_name = "Informacja o wysłanych danych"
        verbose_name_plural = "Informacje o wysłanych danych"

    object.verbose_name = "Rekord"

    def __str__(self):
        return (
            f"Informacja o wysłanych do PBN danych dla rekordu ({self.content_type_id},{self.object_id}) "
            f"z dnia {self.last_updated_on} (status: {'OK' if self.uploaded_okay else 'ERR'})"
        )

    def link_do_pbn_wartosc_id(self):
        return self.pbn_uid_id

    def rekord_w_bpp(self):
        try:
            return self.object
        except ObjectDoesNotExist:
            pass

    def save(
        self, force_insert=False, force_update=False, using=None, update_fields=None
    ):

        if update_fields and "data_sent" in update_fields:
            if self.typ_rekordu != self.data_sent.get("type"):
                update_fields.append("typ_rekordu")

        self.typ_rekordu = self.data_sent.get("type")
        return super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )

1	from django.core.exceptions import ObjectDoesNotExist	3✔
2	from django.db import models	3✔
3	from django.db.models import JSONField	3✔
4
5	from django.contrib.contenttypes.fields import GenericForeignKey	3✔
6	from django.contrib.contenttypes.models import ContentType	3✔
7
8	from bpp import const	3✔
9	from bpp.models import LinkDoPBNMixin	3✔
10
11
12	class SentDataManager(models.Manager):	3✔
13	def get_for_rec(self, rec):	3✔
14	return self.get(	×
15	object_id=rec.pk, content_type=ContentType.objects.get_for_model(rec)
16	)
17
18	def check_if_needed(self, rec, data: dict):	3✔
19	try:	×
20	sd = self.get_for_rec(rec)	×
21	except SentData.DoesNotExist:	×
22	return True	×
23
24	if sd.data_sent != data:	×
25	return True	×
26
27	if not sd.uploaded_okay:	×
28	return True	×
29
30	return False	×
31
32	def updated(	3✔
33	self, rec, data: dict, pbn_uid_id=None, uploaded_okay=True, exception=None
34	):
35	try:	×
36	sd = self.get_for_rec(rec)	×
37	except SentData.DoesNotExist:	×
38	self.create(	×
39	object=rec,
40	data_sent=data,
41	uploaded_okay=uploaded_okay,
42	pbn_uid_id=pbn_uid_id,
43	exception=exception,
44	)
45	return	×
46
47	sd.data_sent = data	×
48	sd.uploaded_okay = uploaded_okay	×
49	sd.exception = exception	×
50	sd.pbn_uid_id = pbn_uid_id	×
51	sd.save()	×
52
53	def ids_for_model(self, model):	3✔
54	return self.filter(content_type=ContentType.objects.get_for_model(model))	×
55
56	def bad_uploads(self, model):	3✔
57	return (	×
58	self.ids_for_model(model)
59	.filter(uploaded_okay=False)
60	.values_list("object_id", flat=True)
61	.distinct()
62	)
63
64
65	class SentData(LinkDoPBNMixin, models.Model):	3✔
66	url_do_pbn = const.LINK_PBN_DO_PUBLIKACJI	3✔
67	content_type = models.ForeignKey(	3✔
68	"contenttypes.ContentType", on_delete=models.CASCADE
69	)
70	object_id = models.PositiveIntegerField(db_index=True)	3✔
71
72	object = GenericForeignKey()	3✔
73
74	data_sent = JSONField("Wysłane dane")	3✔
75	last_updated_on = models.DateTimeField("Data operacji", auto_now=True)	3✔
76
77	uploaded_okay = models.BooleanField(	3✔
78	"Wysłano poprawnie", default=True, db_index=True
79	)
80	exception = models.TextField("Kod błędu", max_length=65535, blank=True, null=True)	3✔
81
82	pbn_uid = models.ForeignKey(	3✔
83	"pbn_api.Publication",
84	verbose_name="Publikacja z PBN",
85	blank=True,
86	null=True,
87	on_delete=models.SET_NULL,
88	)
89
90	typ_rekordu = models.CharField(max_length=50, blank=True, null=True)	3✔
91
92	objects = SentDataManager()	3✔
93
94	class Meta:	3✔
95	verbose_name = "Informacja o wysłanych danych"	3✔
96	verbose_name_plural = "Informacje o wysłanych danych"	3✔
97
98	object.verbose_name = "Rekord"	3✔
99
100	def __str__(self):	3✔
101	return (	×
102	f"Informacja o wysłanych do PBN danych dla rekordu ({self.content_type_id},{self.object_id}) "
103	f"z dnia {self.last_updated_on} (status: {'OK' if self.uploaded_okay else 'ERR'})"
104	)
105
106	def link_do_pbn_wartosc_id(self):	3✔
107	return self.pbn_uid_id	×
108
109	def rekord_w_bpp(self):	3✔
110	try:	×
111	return self.object	×
112	except ObjectDoesNotExist:	×
113	pass	×
114
115	def save(	3✔
116	self, force_insert=False, force_update=False, using=None, update_fields=None
117	):
118
UNCOV 119	if update_fields and "data_sent" in update_fields:	1✔
120	if self.typ_rekordu != self.data_sent.get("type"):	×
121	update_fields.append("typ_rekordu")	×
122
UNCOV 123	self.typ_rekordu = self.data_sent.get("type")	1✔
UNCOV 124	return super().save(	1✔
125	force_insert=force_insert,
126	force_update=force_update,
127	using=using,
128	update_fields=update_fields,
129	)

iplweb / bpp / 7d423bde-8caf-4c8b-8bf5-12ad3f831949

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