diff --git a/pid_provider/models.py b/pid_provider/models.py index 5dea0639..7544809f 100644 --- a/pid_provider/models.py +++ b/pid_provider/models.py @@ -37,6 +37,8 @@ zero_to_none, compare, QueryBuilderPidProviderXML, + fix_get_article_data, + fix_get_data_to_compare, ) from tracker.models import BaseEvent, UnexpectedEvent @@ -615,15 +617,33 @@ def data(self): "v2": self.v2, "aop_pid": self.aop_pid, "pkg_name": self.pkg_name, - "finger_print": self.current_version and self.current_version.finger_print, - "created": self.created and self.created.isoformat(), - "updated": self.updated and self.updated.isoformat(), - "record_status": "updated" if self.updated else "created", "registered_in_core": self.registered_in_core, + "ppx_id": self.id } - _data.update(self.get_readable_data()) + _data["registered_data"] = self.get_readable_data() + _data.update(self.record_status) return _data + @property + def record_status(self): + """Retorna os timestamps e o estado do registro ('created' ou 'updated'). + Calcula a variação entre `created` e `updated` para definir o estado de + persistência. Utilizado para direcionar o código HTTP de resposta: + - 'created' -> 201 Created + - 'updated' -> 200 OK ou 204 No Content + Returns: + dict: Dicionário com `created`, `updated` em ISO 8601 e `record_status`. + """ + d = {} + if self.created: + d["created"] = self.created.isoformat() + d["record_status"] = "created" + if self.updated: + d["updated"] = self.updated.isoformat() + if (self.updated - self.created).total_seconds() > 1: + d["record_status"] = "updated" + return d + @classmethod @profile_classmethod def get_xml_with_pre(cls, v3): @@ -657,25 +677,35 @@ def is_aop(self): return True def get_readable_data(self): - if self.readable_data: - return self.readable_data + readable_data = self.readable_data or {} + if readable_data: + try: + readable_data.pop("partial_body") + except KeyError: + pass + return readable_data if self.xml_with_pre: - return self.xml_with_pre.get_article_data() + return fix_get_article_data(self.xml_with_pre) return {} @property def data_to_compare(self): + data = {} readable = self.get_readable_data() - titles = readable.get("article_titles") - body_fragment = readable.get("body_fragment") - return { - "article_titles": titles or self.xml_with_pre.article_titles_texts, + if readable: + titles = readable.get("article_titles") + body_fragment = readable.get("body_fragment") + if titles: + data["article_titles"] = titles + if body_fragment: + data["body_fragment"] = body_fragment + data.update({ "z_surnames": self.z_surnames, "z_collab": self.z_collab, "z_links": self.z_links, - "z_partial_body": self.z_partial_body, - "body_fragment": body_fragment or self.xml_with_pre.get_body_fragment(PARTIAL_BODY_MAX), - } + "body_fragment_fingerprint": self.z_partial_body, + }) + return data @classmethod @profile_classmethod @@ -758,7 +788,7 @@ def register( input_data = {} input_data.update(xml_with_pre.data) - input_data.update(xml_with_pre.get_article_data()) + input_data.update(fix_get_article_data(xml_with_pre)) input_data["origin"] = origin response["input_data"] = input_data @@ -997,6 +1027,9 @@ def is_updated( ) return + if not registered.readable_data: + return + # verifica se é necessário atualizar if registered.is_equal_to(xml_with_pre): # XML fornecido é igual ao registrado, não precisa continuar @@ -1074,12 +1107,12 @@ def select_record(xml_adapter, selection_results): checamos truthiness (nunca .exists()/.count() sobre queryset). """ unmatched_items = {} - xml_adapter_data_to_compare = xml_adapter.get_data_to_compare() + xml_adapter_data_to_compare = fix_get_data_to_compare(xml_adapter) for label, results in selection_results: if not results: continue - result = PidProviderXML.get_best_match(results, xml_adapter_data_to_compare) + result = PidProviderXML.get_best_match(results, xml_adapter_data_to_compare) matched = result.get("matched") unmatched = result.get("unmatched") @@ -1117,14 +1150,15 @@ def get_record_by_pid_v3(cls, xml_adapter): # pid v3 é inédito raise cls.DoesNotExist - xml_adapter_data_to_compare = xml_adapter.get_data_to_compare() + xml_adapter_data_to_compare = fix_get_data_to_compare(xml_adapter) result = PidProviderXML.get_best_match(results, xml_adapter_data_to_compare) + registered = result.get("registered") if not registered: - xml_data = xml_adapter.xml_with_pre.get_article_data(PARTIAL_BODY_MAX) - items = [item.data for item in results] raise PidProviderXMLPidV3ConflictError( - _(f"{xml_pid_v3} belongs to {items}, not to {xml_data}") + _("{} do not belong to {}. Result: {}").format( + xml_pid_v3, xml_adapter_data_to_compare, result, + ) ) return registered @@ -1150,10 +1184,10 @@ def get_best_match(results, xml_adapter_data): ao consumir o retorno, nunca acesso direto. - ``"unmatched"``: presente apenas se houver ao menos 1 - candidato com ``percentual_score`` <= 0.6. Lista de + candidato com ``percentual_score`` <= min_rate. Lista de ``item.data`` desses candidatos. - ``"registered"``: presente apenas se houver ao menos 1 - candidato aprovado (score > 0.6). Contém o OBJETO + candidato aprovado (score > min_rate). Contém o OBJETO ``PidProviderXML`` (não o dict ``.data``) do candidato com maior score — em caso de empate, o critério de desempate é ``updated`` mais recente e, em seguida, maior ``id``. @@ -1166,20 +1200,26 @@ def get_best_match(results, xml_adapter_data): detail = {} found = [] items = {} + responses = {} + min_rate = 0.6 + if len(xml_adapter_data) <= 4: + min_rate = 0.49 for item in results: item_data = item.data_to_compare response = compare(item_data, xml_adapter_data) items[item.id] = item + responses[item.id] = response found.append((response["percentual_score"], item.updated.isoformat(), item.id)) found = sorted(found, reverse=True) matched = [] unmatched = [] for percentual_score, updated, item_id in found: - if percentual_score > 0.6: - matched.append(items[item_id].data) + data = {"data": items[item_id].data, "response": responses[item_id]} + if percentual_score > min_rate: + matched.append(data) else: - unmatched.append(items[item_id].data) + unmatched.append(data) if matched: detail["registered"] = items[found[0][-1]] if len(matched) > 1: @@ -1210,7 +1250,7 @@ def _add_data(self, xml_adapter, registered_in_core): self.z_links = xml_adapter.z_links self.z_partial_body = xml_adapter.xml_with_pre.body_fragment_fingerprint - self.readable_data = xml_adapter.xml_with_pre.get_article_data() + self.readable_data = fix_get_article_data(xml_adapter.xml_with_pre) @profile_method def _add_dates(self, xml_adapter, origin_date, available_since): @@ -1382,6 +1422,8 @@ def is_registered( response["registered"] = True response.update(registered.data) response["is_equal"] = registered.is_equal_to(xml_with_pre) + if not registered.readable_data: + response["is_equal"] = False return response except Exception as e: exc_type, exc_value, exc_traceback = sys.exc_info() diff --git a/pid_provider/query_params.py b/pid_provider/query_params.py index 461e69f7..77df860e 100644 --- a/pid_provider/query_params.py +++ b/pid_provider/query_params.py @@ -7,8 +7,70 @@ from pid_provider import exceptions +def fix_get_data_to_compare(xml_adapter): + """ + packtools 4.16.11 + { + ... + "z_partial_body": self.z_partial_body, + ... + } + packtools > 4.17.0 + { + ... + "body_fragment_fingerprint": self.xml_with_pre.body_fragment_fingerprint, + ... + } + """ + data = xml_adapter.get_data_to_compare() + # independentemente da release do packtools, + # o valor para z_partial_body na comparação é body_fragment_fingerprint + data["body_fragment_fingerprint"] = xml_adapter.xml_with_pre.body_fragment_fingerprint + return data + + +def fix_get_article_data(xml_with_pre, max_length=300): + """ + Wrapper de compatibilidade em torno de xml_with_pre.get_article_data(). + + Remove a chave legada "partial_body" do dict retornado (substituída + por "body_fragment" nas versões atuais do packtools), evitando que + código que consome esse dict dependa de uma chave que pode não + refletir mais o valor realmente usado nas comparações de corpo do + artigo. + """ + try: + data = xml_with_pre.readable_data + except AttributeError: + data = xml_with_pre.get_article_data(max_length) + try: + data.pop("partial_body") + except KeyError: + pass + return data + + def compare(registered_items, input_data): """ + Compara, item a item, os valores registrados (registered_items) com + os valores do XML de entrada (input_data). + + Para cada label em registered_items, obtém o valor correspondente em + input_data via `.get(label)` — um label ausente em input_data é + tratado como None (não é pulado). Delega a comparação individual a + compare_items() e agrega os scores. + + Returns + ------- + dict + { + "items": lista de resultados de compare_items() (um por label), + "total_score": soma dos scores individuais, + "percentual_score": total_score / len(items), + } + + Levanta ZeroDivisionError se registered_items estiver vazio (items + fica vazio e a divisão por zero não é tratada explicitamente). """ total_score = 0 items = [] @@ -24,6 +86,15 @@ def compare(registered_items, input_data): def compare_lists(registered, xml_adapter_titles): + """ + Compara duas listas de textos (ex.: títulos de artigo) por + similaridade de conjunto de palavras. + + Retorna 1 se as listas forem idênticas, 0 se qualquer uma das duas + estiver vazia/None, ou o resultado de how_similar() entre as + palavras únicas de cada lista (ordenadas e unidas em uma única + string), caso contrário. + """ if xml_adapter_titles == registered: return 1 if not xml_adapter_titles: @@ -40,6 +111,20 @@ def compare_lists(registered, xml_adapter_titles): def compare_items(label, registered, input_data): + """ + Compara um único item entre o valor registrado e o valor de entrada. + + - Se `registered` for uma lista (ex.: títulos), delega a + compare_lists(). + - Caso os dois valores, normalizados (falsy vira None), sejam + iguais, o score é 1. + - Caso contrário, o score vem de how_similar() entre os dois valores + (None é tratado como string vazia). + + Retorna um dict {"label": label, "score": score}, incluindo também + "registered" quando o score não é 1 — útil para inspecionar + divergências. + """ if isinstance(registered, list): score = compare_lists(registered, input_data) elif (input_data or None) == (registered or None): @@ -49,10 +134,16 @@ def compare_items(label, registered, input_data): response = {"label": label, "score": score} if score != 1: response["registered"] = registered + response["input_data"] = input_data return response def get_score(registered, xml_data, min_value, max_value): + """ + Score binário simples: max_value se registered == xml_data e ambos + truthy; min_value se ambos forem iguais mas falsy (ex.: None == + None); 0 caso contrário. + """ if registered == xml_data: if registered: return max_value @@ -61,6 +152,13 @@ def get_score(registered, xml_data, min_value, max_value): def zero_to_none(data): + """ + Normaliza um campo numérico textual: retorna None se `data` for + falsy; retorna `data` sem alteração se não for composto só de + dígitos; e converte para None quando o valor numérico for zero + (nos demais casos, mantém `data` como string, sem converter para + int). + """ if not data: return if not data.isdigit(): @@ -80,45 +178,70 @@ class QueryBuilderPidProviderXML: def __init__(self, xml_adapter): """ - Inicializa o construtor de queries obtendo os dicionários de dados do adaptador. - + Inicializa o construtor de queries obtendo os dados do adaptador. + Parameters ---------- xml_adapter : PidProviderXMLAdapter Adaptador com dados do XML para busca + + Define + ------ + z_body_fragment : fingerprint sha256 de um fragmento estável do + corpo do artigo (XMLWithPre.body_fragment_fingerprint), + acessado direto do xml_with_pre — não requer nenhuma + mudança no packtools nem no PidProviderXMLAdapter. É mais + robusto que z_partial_body (que é só o primeiro parágrafo + não vazio e pode colidir entre artigos diferentes, ex.: + rótulos de seção genéricos como "ARTIGO DE REVISÃO"). + z_partial_body : hash legado do corpo do artigo + (xml_with_pre.z_partial_body), mantido apenas para casar + com registros antigos. + adapter_data : dict bruto de xml_adapter.data. + xml_with_pre_data : dict normalizado retornado por + fix_get_article_data(xml_adapter.xml_with_pre, 300) — já + sem a chave legada "partial_body". """ self.xml_adapter = xml_adapter - # Centraliza o acesso aos dados brutos e normalizados (hashes de 64 chars) - # z_body: fingerprint sha256 do corpo INTEIRO do artigo - # (XMLWithPre.body_fingerprint), acessado direto do xml_with_pre. - # z_body_fragment: fingerprint sha256 de um fragmento do artigo - # (XMLWithPre.body_fragment_fingerprint), acessado direto do - # xml_with_pre — não requer nenhuma mudança no packtools nem no - # PidProviderXMLAdapter. É mais robusto que z_partial_body (que - # é só o primeiro parágrafo não vazio e pode colidir entre - # artigos diferentes, ex.: rótulos de seção genéricos como - # "ARTIGO DE REVISÃO"). - self.z_body = xml_adapter.xml_with_pre.body_fingerprint self.z_body_fragment = xml_adapter.xml_with_pre.body_fragment_fingerprint + self.z_partial_body = xml_adapter.z_partial_body self.adapter_data = xml_adapter.data - self.compare_data = xml_adapter.get_data_to_compare() - self.xml_with_pre_data = xml_adapter.xml_with_pre.get_article_data(300) + self.xml_with_pre_data = fix_get_article_data(xml_adapter.xml_with_pre, 300) @property def pkg_name_list(self): - # --- Resolução Consolidada de Package Names --- + """ + Consolida, em um único set, todos os nomes de pacote possíveis + para o artigo: o nome enviado via parâmetro no construtor, o + nome oficial atual calculado pelo packtools (sps_pkg_name) e + todos os nomes depreciados/alternativos já usados no passado. + Valores falsy são descartados. + """ pkg_names = set() - # 1. Nome enviado originalmente via parâmetro no construtor if self.xml_adapter.pkg_name: pkg_names.add(self.xml_adapter.pkg_name) - # 2. Nome oficial atual gerado pelo motor de cálculo do XML if self.xml_adapter.sps_pkg_name: pkg_names.add(self.xml_adapter.sps_pkg_name) - # 3. Consolida todas as listas de nomes depreciados/alternativos pkg_names.update(self.xml_adapter.xml_with_pre.deprecated_sps_pkg_name_list) return set(item for item in pkg_names if item) def validate_input_data(self): + """ + Garante que o XML de entrada tem parâmetros suficientes para + localizar um registro existente. + + Levanta: + - RequiredPublicationYearErrorToGetPidProviderXMLError se não + houver ano de publicação; + - RequiredISSNErrorToGetPidProviderXMLError se não houver ISSN + eletrônico nem impresso; + - NotEnoughParametersToGetPidProviderXMLError se, além do + ano/ISSN, não houver nenhum dado de localização do artigo + (elocation_id/fpage/lpage/etc.) nem nenhum dado textual + (títulos, sobrenomes, colaboradores, links ou fragmento do + corpo) que permita diferenciar o artigo de outros do mesmo + fascículo. + """ if not self.adapter_data.get("pub_year"): raise exceptions.RequiredPublicationYearErrorToGetPidProviderXMLError() issn_electronic = self.adapter_data.get("issn_electronic") @@ -135,7 +258,7 @@ def validate_input_data(self): self.xml_with_pre_data.get("surnames"), self.xml_with_pre_data.get("collab"), self.xml_with_pre_data.get("links"), - self.xml_with_pre_data.get("partial_body"), + self.xml_with_pre_data.get("body_fragment"), ] if any(items): return @@ -150,24 +273,19 @@ def identifier_queries(self): """ q = Q() - # PIDs diretos do xml_adapter (não envelopados no data dict) v3 = self.xml_adapter.v3 v2 = self.xml_adapter.v2 aop_pid = self.xml_adapter.aop_pid - # PID v3 - máxima prioridade if v3: q |= Q(v3=v3) - # PID v2 if v2: q |= Q(v2=v2) - # AOP PID if aop_pid: q |= Q(v2=aop_pid) | Q(aop_pid=aop_pid) - # Package names históricos e atuais pkg_names = self.pkg_name_list if pkg_names: q |= Q(pkg_name__in=pkg_names) @@ -237,9 +355,10 @@ def partial_body_query(self): dois formatos possíveis de hash, dependendo de quando o registro foi salvo: - - legado: hash de z_partial_body (primeiro parágrafo não vazio - do corpo, via xml_adapter.z_partial_body); - - atual: fingerprint do parcial do artigo + - legado: self.z_partial_body + (xml_adapter.xml_with_pre.z_partial_body) — hash do primeiro + parágrafo não vazio do corpo; + - atual: self.z_body_fragment (xml_with_pre.body_fragment_fingerprint), gravado no mesmo campo z_partial_body a partir desta correção (sem necessidade de migração/backfill). @@ -256,8 +375,7 @@ def partial_body_query(self): preservando o comportamento equivalente ao antigo `Q(z_partial_body=None)` (que o Django traduz para IS NULL). """ - z_partial_body = self.adapter_data.get("z_partial_body") - candidates = set(v for v in (z_partial_body, self.z_body_fragment, self.z_body) if v) + candidates = set(v for v in (self.z_partial_body, self.z_body_fragment) if v) if candidates: return Q(z_partial_body__in=candidates) return Q(z_partial_body__isnull=True) @@ -265,7 +383,9 @@ def partial_body_query(self): @property def article_data_query(self): """ - Constrói query para busca por dados textuais codificados (hashes sha256). + Constrói query para busca por dados textuais codificados (hashes + sha256 de sobrenomes, colaboradores e links), combinada com + partial_body_query (hash/fingerprint do corpo do artigo). """ z_surnames = self.adapter_data.get("z_surnames") z_collab = self.adapter_data.get("z_collab") @@ -278,6 +398,13 @@ def article_data_query(self): ) & self.partial_body_query def get_article_data_query(self, issue): + """ + Combina article_data_query com os parâmetros de fascículo e + localização do artigo (quando `issue` é truthy), ou exige que + todos os campos de localização estejam nulos (quando `issue` é + falsy) — caso de artigos sem paginação/localização definida + (ex.: ahead-of-print). + """ if issue: return ( self.article_data_query & @@ -294,4 +421,4 @@ def get_article_data_query(self, issue): fpage__isnull=True, lpage__isnull=True, ) - ) \ No newline at end of file + ) diff --git a/pid_provider/tests/test_fix_pid_v2.py b/pid_provider/tests/test_fix_pid_v2.py new file mode 100644 index 00000000..5e8dc57f --- /dev/null +++ b/pid_provider/tests/test_fix_pid_v2.py @@ -0,0 +1,142 @@ +"""Testes para FixPidV2 (pid_provider/models.py).""" +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.db import IntegrityError +from django.test import TestCase + +from pid_provider.models import FixPidV2, PidProviderXML + +User = get_user_model() + + +class FixPidV2GetTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpid-user", password="x") + self.ppx = PidProviderXML.objects.create(creator=self.user, v3="ABC", v2="V2-OLD") + + def test_raises_value_error_when_pid_provider_xml_is_falsy(self): + with self.assertRaises(ValueError): + FixPidV2.get(pid_provider_xml=None) + + def test_raises_does_not_exist_when_no_record(self): + with self.assertRaises(FixPidV2.DoesNotExist): + FixPidV2.get(pid_provider_xml=self.ppx) + + +class FixPidV2CreateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpid-create", password="x") + self.ppx = PidProviderXML.objects.create(creator=self.user, v3="ABC", v2="V2-OLD") + + def test_raises_value_error_when_correct_equals_incorrect(self): + with self.assertRaises(ValueError): + FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2="V2-OLD", correct_pid_v2="V2-OLD" + ) + + def test_raises_value_error_when_correct_pid_v2_missing(self): + with self.assertRaises(ValueError): + FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2="V2-OLD", correct_pid_v2=None + ) + + def test_raises_value_error_when_incorrect_pid_v2_missing(self): + with self.assertRaises(ValueError): + FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2=None, correct_pid_v2="V2-NEW" + ) + + def test_creates_new_record(self): + obj = FixPidV2.create( + self.user, + self.ppx, + incorrect_pid_v2="V2-OLD", + correct_pid_v2="V2-NEW", + fixed_in_core=False, + fixed_in_upload=True, + ) + + self.assertIsNotNone(obj.pk) + self.assertEqual(obj.incorrect_pid_v2, "V2-OLD") + self.assertEqual(obj.correct_pid_v2, "V2-NEW") + self.assertTrue(obj.fixed_in_upload) + + def test_falls_back_to_get_on_integrity_error(self): + existing = FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2="V2-OLD", correct_pid_v2="V2-NEW" + ) + + with patch.object(FixPidV2, "save", side_effect=IntegrityError): + result = FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2="V2-OLD", correct_pid_v2="V2-OTHER" + ) + + self.assertEqual(result.pk, existing.pk) + + +class FixPidV2CreateOrUpdateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpid-cou", password="x") + self.ppx = PidProviderXML.objects.create(creator=self.user, v3="ABC", v2="V2-OLD") + + def test_creates_when_none_exists(self): + obj = FixPidV2.create_or_update( + self.user, + self.ppx, + incorrect_pid_v2="V2-OLD", + correct_pid_v2="V2-NEW", + fixed_in_core=True, + fixed_in_upload=False, + ) + self.assertIsNotNone(obj.pk) + self.assertTrue(obj.fixed_in_core) + + def test_updates_existing_merging_truthy_flags_only(self): + """ + fixed_in_core/fixed_in_upload são combinados com `novo or antigo`: + um novo valor falsy (None/False) NÃO apaga um valor truthy já + persistido. + """ + existing = FixPidV2.create( + self.user, + self.ppx, + incorrect_pid_v2="V2-OLD", + correct_pid_v2="V2-NEW", + fixed_in_core=True, + fixed_in_upload=None, + ) + + updated = FixPidV2.create_or_update( + self.user, + self.ppx, + fixed_in_core=None, + fixed_in_upload=True, + ) + + self.assertEqual(updated.pk, existing.pk) + self.assertTrue(updated.fixed_in_core) # preservado + self.assertTrue(updated.fixed_in_upload) # setado agora + self.assertEqual(updated.updated_by, self.user) + + +class FixPidV2GetOrCreateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpid-goc", password="x") + self.ppx = PidProviderXML.objects.create(creator=self.user, v3="ABC", v2="V2-OLD") + + def test_returns_existing_without_creating_new(self): + existing = FixPidV2.create( + self.user, self.ppx, incorrect_pid_v2="V2-OLD", correct_pid_v2="V2-NEW" + ) + + found = FixPidV2.get_or_create(self.user, self.ppx, correct_pid_v2="V2-OTHER") + + self.assertEqual(found.pk, existing.pk) + self.assertEqual(FixPidV2.objects.count(), 1) + + def test_creates_using_pid_provider_xml_v2_as_incorrect(self): + obj = FixPidV2.get_or_create(self.user, self.ppx, correct_pid_v2="V2-NEW") + + self.assertEqual(obj.incorrect_pid_v2, "V2-OLD") + self.assertEqual(obj.correct_pid_v2, "V2-NEW") diff --git a/pid_provider/tests/test_get_best_match.py b/pid_provider/tests/test_get_best_match.py index 1d83e94e..f84908ec 100644 --- a/pid_provider/tests/test_get_best_match.py +++ b/pid_provider/tests/test_get_best_match.py @@ -45,7 +45,7 @@ def side_effect_compare(item_data, xml_adapter_data): # "unmatched" sempre é exposto self.assertEqual(len(result["unmatched"]), 1) - self.assertEqual(result["unmatched"][0]["id"], 102) + self.assertEqual(result["unmatched"][0]["data"]["id"], 102) @patch("pid_provider.models.compare") def test_get_best_match_no_candidates_approved(self, mock_compare): @@ -64,7 +64,7 @@ def test_get_best_match_no_candidates_approved(self, mock_compare): self.assertNotIn("registered", result) self.assertNotIn("matched", result) self.assertEqual(len(result["unmatched"]), 1) - self.assertEqual(result["unmatched"][0]["id"], 201) + self.assertEqual(result["unmatched"][0]["data"]["id"], 201) @patch("pid_provider.models.compare") def test_get_best_match_two_matches_excludes_registered_from_matched(self, mock_compare): @@ -94,7 +94,7 @@ def test_get_best_match_two_matches_excludes_registered_from_matched(self, mock_ # "matched" agora é matched[1:] -> exclui o item que virou "registered" self.assertIn("matched", result) self.assertEqual(len(result["matched"]), 1) - self.assertEqual(result["matched"][0]["id"], 301) + self.assertEqual(result["matched"][0]["data"]["id"], 301) self.assertNotIn("unmatched", result) @@ -133,7 +133,7 @@ def side_effect_compare(item_data, xml_adapter_data): # "matched" deve conter apenas item_2 (0.85) e item_3 (0.75), nessa ordem self.assertEqual(len(result["matched"]), 2) - self.assertEqual(result["matched"][0]["id"], 402) - self.assertEqual(result["matched"][1]["id"], 403) + self.assertEqual(result["matched"][0]["data"]["id"], 402) + self.assertEqual(result["matched"][1]["data"]["id"], 403) self.assertNotIn("unmatched", result) \ No newline at end of file diff --git a/pid_provider/tests/test_models.py b/pid_provider/tests/test_models.py new file mode 100644 index 00000000..0a355472 --- /dev/null +++ b/pid_provider/tests/test_models.py @@ -0,0 +1,26 @@ +from unittest.mock import patch + +from django.test import SimpleTestCase + +from pid_provider.models import PidProviderXMLRegistration, XMLVersion + + +class XMLVersionTests(SimpleTestCase): + def test_string_representation_without_pid_provider_xml(self): + xml_version = XMLVersion(pid_provider_xml=None) + + self.assertEqual(str(xml_version), "- None") + + +class PidProviderXMLRegistrationTests(SimpleTestCase): + @patch.object(PidProviderXMLRegistration, "save") + def test_skipped_event_does_not_store_detail(self, save): + registration = PidProviderXMLRegistration.record( + user=None, + event_status=PidProviderXMLRegistration.EVENT_SKIPPED, + detail={"large": "payload"}, + ) + + save.assert_called_once_with() + self.assertIsNotNone(registration) + self.assertIsNone(registration.detail) diff --git a/pid_provider/tests/test_other_pid.py b/pid_provider/tests/test_other_pid.py new file mode 100644 index 00000000..75e4ea46 --- /dev/null +++ b/pid_provider/tests/test_other_pid.py @@ -0,0 +1,82 @@ +"""Testes para OtherPid (pid_provider/models.py).""" +from django.contrib.auth import get_user_model +from django.test import TestCase + +from pid_provider.models import OtherPid, PidProviderXML, XMLVersion + +User = get_user_model() + + +class OtherPidGetOrCreateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="otherpid-user", password="x") + self.ppx = PidProviderXML.objects.create(creator=self.user, v3="ABC") + # version é obrigatório (truthy) para get_or_create -- um FK vazio + # (None) faz a validação falhar e levantar ValueError. + self.version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=self.ppx) + + def test_raises_value_error_when_any_required_argument_is_missing(self): + with self.assertRaises(ValueError): + OtherPid.get_or_create( + pid_type=None, + pid_in_xml="OLD-V3", + version=self.version, + user=self.user, + pid_provider_xml=self.ppx, + ) + + def test_raises_value_error_when_version_is_missing(self): + with self.assertRaises(ValueError): + OtherPid.get_or_create( + pid_type="pid_v3", + pid_in_xml="OLD-V3", + version=None, + user=self.user, + pid_provider_xml=self.ppx, + ) + + def test_creates_new_when_none_exists(self): + obj = OtherPid.get_or_create( + pid_type="pid_v3", + pid_in_xml="OLD-V3", + version=self.version, + user=self.user, + pid_provider_xml=self.ppx, + ) + + self.assertIsNotNone(obj.pk) + self.assertEqual(obj.pid_type, "pid_v3") + self.assertEqual(obj.pid_in_xml, "OLD-V3") + self.assertEqual(obj.creator, self.user) + self.assertEqual(obj.pid_provider_xml, self.ppx) + + def test_returns_existing_without_duplicating(self): + first = OtherPid.get_or_create( + pid_type="pid_v2", + pid_in_xml="OLD-V2", + version=self.version, + user=self.user, + pid_provider_xml=self.ppx, + ) + + second = OtherPid.get_or_create( + pid_type="pid_v2", + pid_in_xml="OLD-V2", + version=self.version, + user=self.user, + pid_provider_xml=self.ppx, + ) + + self.assertEqual(first.pk, second.pk) + self.assertEqual(OtherPid.objects.filter(pid_provider_xml=self.ppx).count(), 1) + + def test_created_updated_prefers_updated_when_present(self): + obj = OtherPid.get_or_create( + pid_type="pid_v3", + pid_in_xml="OLD-V3", + version=self.version, + user=self.user, + pid_provider_xml=self.ppx, + ) + obj.updated = None + self.assertEqual(obj.created_updated, obj.created) diff --git a/pid_provider/tests/test_pid_provider_config.py b/pid_provider/tests/test_pid_provider_config.py new file mode 100644 index 00000000..d5c1c1d1 --- /dev/null +++ b/pid_provider/tests/test_pid_provider_config.py @@ -0,0 +1,52 @@ +"""Testes para PidProviderConfig (pid_provider/models.py).""" +from django.contrib.auth import get_user_model +from django.test import TestCase + +from pid_provider.models import PidProviderConfig + +User = get_user_model() + + +class PidProviderConfigGetOrCreateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="config-user", password="x") + + def test_creates_new_when_none_exists(self): + obj = PidProviderConfig.get_or_create( + creator=self.user, + pid_provider_api_post_xml="https://example.org/post", + pid_provider_api_get_token="https://example.org/token", + api_username="user", + api_password="pass", + timeout=30, + ) + + self.assertIsNotNone(obj.pk) + self.assertEqual(obj.pid_provider_api_post_xml, "https://example.org/post") + self.assertEqual(obj.timeout, 30) + + def test_returns_existing_singleton_ignoring_new_arguments(self): + """ + get_or_create usa cls.objects.first() -- se já existe QUALQUER + registro, ele é retornado tal como está, ignorando os parâmetros + passados na segunda chamada (não atualiza os campos). + """ + first = PidProviderConfig.get_or_create( + creator=self.user, + pid_provider_api_post_xml="https://example.org/post", + ) + + second = PidProviderConfig.get_or_create( + creator=self.user, + pid_provider_api_post_xml="https://other.example.org/post", + ) + + self.assertEqual(first.pk, second.pk) + self.assertEqual(second.pid_provider_api_post_xml, "https://example.org/post") + self.assertEqual(PidProviderConfig.objects.count(), 1) + + def test_str_representation(self): + obj = PidProviderConfig.get_or_create( + creator=self.user, pid_provider_api_post_xml="https://example.org/post" + ) + self.assertEqual(str(obj), "https://example.org/post") diff --git a/pid_provider/tests/test_pid_provider_xml_helpers.py b/pid_provider/tests/test_pid_provider_xml_helpers.py new file mode 100644 index 00000000..d8e9f872 --- /dev/null +++ b/pid_provider/tests/test_pid_provider_xml_helpers.py @@ -0,0 +1,473 @@ +""" +Testes para métodos auxiliares "puros"/pouco acoplados de PidProviderXML +(pid_provider/models.py): validação de PID, comparação de PIDs registrados, +propriedades de leitura (is_aop, record_status, created_updated, +get_readable_data, data_to_compare, data) e utilitários de consulta +(get_by_pid_v3, get_queryset, delete_queryset, mark_as_waiting/done). + +Fluxos mais pesados de negócio (register, select_record, get_best_match) +já têm cobertura própria em outros arquivos deste diretório. +""" +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from django.contrib.auth import get_user_model +from django.test import SimpleTestCase, TestCase + +from pid_provider import exceptions +from pid_provider.models import ( + OtherPid, + PidProviderXML, + PidProviderXMLPidV3ConflictError, +) + +User = get_user_model() + + +class IsValidPidLenTests(SimpleTestCase): + def test_returns_true_for_23_char_value(self): + self.assertTrue(PidProviderXML.is_valid_pid_len("a" * 23, "pid_v3")) + + def test_raises_value_error_for_wrong_length(self): + with self.assertRaises(ValueError) as ctx: + PidProviderXML.is_valid_pid_len("short", "pid_v3") + self.assertIn("pid_v3", str(ctx.exception)) + + def test_raises_value_error_for_falsy_value(self): + with self.assertRaises(ValueError): + PidProviderXML.is_valid_pid_len(None, "aop_pid") + with self.assertRaises(ValueError): + PidProviderXML.is_valid_pid_len("", "aop_pid") + + +class IsUpdatedTests(SimpleTestCase): + """is_updated é @staticmethod: nunca retorna valor -- sinaliza via exceção ou None.""" + + def test_force_update_returns_none_regardless_of_other_params(self): + result = PidProviderXML.is_updated( + xml_with_pre=MagicMock(), registered=None, force_update=True, + origin_date=None, registered_in_core=None, + ) + self.assertIsNone(result) + + def test_not_registered_returns_none(self): + result = PidProviderXML.is_updated( + xml_with_pre=MagicMock(), registered=None, force_update=False, + origin_date=None, registered_in_core=None, + ) + self.assertIsNone(result) + + def test_needs_registered_in_core_update_returns_none(self): + registered = MagicMock(registered_in_core=False) + result = PidProviderXML.is_updated( + xml_with_pre=MagicMock(), registered=registered, force_update=False, + origin_date=None, registered_in_core=True, + ) + self.assertIsNone(result) + + def test_missing_readable_data_returns_none(self): + registered = MagicMock(registered_in_core=True, readable_data=None) + result = PidProviderXML.is_updated( + xml_with_pre=MagicMock(), registered=registered, force_update=False, + origin_date=None, registered_in_core=None, + ) + self.assertIsNone(result) + + def test_equal_xml_raises_skip(self): + registered = MagicMock(registered_in_core=True, readable_data={"x": 1}) + registered.is_equal_to.return_value = True + with self.assertRaises(exceptions.SkipSavePidProviderXML): + PidProviderXML.is_updated( + xml_with_pre=MagicMock(), registered=registered, force_update=False, + origin_date=None, registered_in_core=None, + ) + + def test_aop_over_vor_raises_forbidden(self): + registered = MagicMock(registered_in_core=True, readable_data={"x": 1}, is_aop=False) + registered.is_equal_to.return_value = False + xml_with_pre = MagicMock(is_aop=True) + with self.assertRaises(exceptions.ForbiddenPidProviderXMLRegistrationError): + PidProviderXML.is_updated( + xml_with_pre=xml_with_pre, registered=registered, force_update=False, + origin_date=None, registered_in_core=None, + ) + + def test_older_origin_date_raises_skip(self): + registered = MagicMock( + registered_in_core=True, readable_data={"x": 1}, is_aop=False, origin_date="2026-02-01" + ) + registered.is_equal_to.return_value = False + xml_with_pre = MagicMock(is_aop=False) + with self.assertRaises(exceptions.SkipSavePidProviderXML): + PidProviderXML.is_updated( + xml_with_pre=xml_with_pre, registered=registered, force_update=False, + origin_date="2026-01-01", registered_in_core=None, + ) + + def test_newer_origin_date_returns_none(self): + registered = MagicMock( + registered_in_core=True, readable_data={"x": 1}, is_aop=False, origin_date="2026-01-01" + ) + registered.is_equal_to.return_value = False + xml_with_pre = MagicMock(is_aop=False) + result = PidProviderXML.is_updated( + xml_with_pre=xml_with_pre, registered=registered, force_update=False, + origin_date="2026-02-01", registered_in_core=None, + ) + self.assertIsNone(result) + + +class CheckRegisteredPidsChangedTests(SimpleTestCase): + def test_reports_only_fields_that_differ(self): + ppx = PidProviderXML(v3="V3-OLD", v2="V2-SAME", aop_pid=None) + xml_with_pre = SimpleNamespace(v3="V3-NEW", v2="V2-SAME", aop_pid="AOP-NEW") + + changed = ppx.check_registered_pids_changed(xml_with_pre) + + by_type = {c["pid_type"]: c for c in changed} + self.assertEqual(set(by_type), {"pid_v3", "aop_pid"}) + self.assertEqual(by_type["pid_v3"]["pid_in_xml"], "V3-NEW") + self.assertEqual(by_type["pid_v3"]["registered"], "V3-OLD") + self.assertEqual(by_type["aop_pid"]["pid_in_xml"], "AOP-NEW") + self.assertEqual(by_type["aop_pid"]["registered"], None) + + def test_returns_empty_list_when_nothing_changed(self): + ppx = PidProviderXML(v3="V3", v2="V2", aop_pid="AOP") + xml_with_pre = SimpleNamespace(v3="V3", v2="V2", aop_pid="AOP") + + self.assertEqual(ppx.check_registered_pids_changed(xml_with_pre), []) + + +class GetValidPidV3Tests(SimpleTestCase): + def test_no_xml_pid_returns_registered_pid(self): + xml_adapter = MagicMock(v3=None) + result = PidProviderXML.get_valid_pid_v3(xml_adapter, registered_pid="REG-PID") + self.assertEqual(result, "REG-PID") + + def test_no_xml_pid_and_no_registered_generates_unique(self): + xml_adapter = MagicMock(v3=None) + with patch.object(PidProviderXML, "_get_unique_v3", return_value="GENERATED"): + result = PidProviderXML.get_valid_pid_v3(xml_adapter, registered_pid=None) + self.assertEqual(result, "GENERATED") + + def test_xml_pid_equal_to_registered_skips_lookup(self): + xml_adapter = MagicMock(v3="SAME-PID") + with patch.object(PidProviderXML, "get_record_by_pid_v3") as mock_get: + result = PidProviderXML.get_valid_pid_v3(xml_adapter, registered_pid="SAME-PID") + mock_get.assert_not_called() + self.assertEqual(result, "SAME-PID") + + def test_xml_pid_belongs_to_same_document_is_accepted(self): + xml_adapter = MagicMock(v3="NEW-PID") + with patch.object(PidProviderXML, "get_record_by_pid_v3", return_value=MagicMock()): + result = PidProviderXML.get_valid_pid_v3(xml_adapter, registered_pid="OLD-PID") + self.assertEqual(result, "NEW-PID") + + def test_xml_pid_unused_is_accepted(self): + xml_adapter = MagicMock(v3="UNUSED-PID") + with patch.object( + PidProviderXML, "get_record_by_pid_v3", side_effect=PidProviderXML.DoesNotExist + ): + result = PidProviderXML.get_valid_pid_v3(xml_adapter, registered_pid="OLD-PID") + self.assertEqual(result, "UNUSED-PID") + + def test_conflict_raises_when_auto_solve_disabled(self): + xml_adapter = MagicMock(v3="CONFLICTING-PID") + with patch.object( + PidProviderXML, + "get_record_by_pid_v3", + side_effect=PidProviderXMLPidV3ConflictError("conflict"), + ): + with self.assertRaises(PidProviderXMLPidV3ConflictError): + PidProviderXML.get_valid_pid_v3( + xml_adapter, registered_pid="OLD-PID", auto_solve_pid_conflict=False + ) + + def test_conflict_ignored_when_auto_solve_enabled_falls_back_to_registered(self): + xml_adapter = MagicMock(v3="CONFLICTING-PID") + with patch.object( + PidProviderXML, + "get_record_by_pid_v3", + side_effect=PidProviderXMLPidV3ConflictError("conflict"), + ): + result = PidProviderXML.get_valid_pid_v3( + xml_adapter, registered_pid="OLD-PID", auto_solve_pid_conflict=True + ) + self.assertEqual(result, "OLD-PID") + + +class CompleteMissingXmlPidsTests(SimpleTestCase): + def _adapter(self, v3=None, v2=None, aop_pid=None): + xml_with_pre = SimpleNamespace(v3=v3, v2=v2, aop_pid=aop_pid) + return MagicMock(xml_with_pre=xml_with_pre) + + def test_validates_pid_lengths_before_resolving(self): + xml_adapter = self._adapter(v3="too-short") + with self.assertRaises(ValueError): + PidProviderXML.complete_missing_xml_pids(xml_adapter, registered=None, auto_solve_pid_conflict=True) + + def test_records_pid_v3_change_when_resolved_value_differs(self): + xml_adapter = self._adapter(v3="a" * 23) + with patch.object(PidProviderXML, "get_valid_pid_v3", return_value="b" * 23): + xml_changed = PidProviderXML.complete_missing_xml_pids( + xml_adapter, registered=None, auto_solve_pid_conflict=True + ) + self.assertEqual(xml_changed, {"pid_v3": "b" * 23}) + self.assertEqual(xml_adapter.xml_with_pre.v3, "b" * 23) + + def test_no_change_when_resolved_pid_v3_matches_incoming(self): + pid = "a" * 23 + xml_adapter = self._adapter(v3=pid) + with patch.object(PidProviderXML, "get_valid_pid_v3", return_value=pid): + xml_changed = PidProviderXML.complete_missing_xml_pids( + xml_adapter, registered=None, auto_solve_pid_conflict=True + ) + self.assertEqual(xml_changed, {}) + + def test_fills_missing_v2_and_aop_pid_from_registered(self): + pid = "a" * 23 + xml_adapter = self._adapter(v3=pid, v2=None, aop_pid=None) + registered = MagicMock(v3=pid, v2="REG-V2", aop_pid="REG-AOP") + with patch.object(PidProviderXML, "get_valid_pid_v3", return_value=pid): + xml_changed = PidProviderXML.complete_missing_xml_pids( + xml_adapter, registered=registered, auto_solve_pid_conflict=True + ) + self.assertEqual(xml_changed, {"pid_v2": "REG-V2", "aop_pid": "REG-AOP"}) + self.assertEqual(xml_adapter.xml_with_pre.v2, "REG-V2") + self.assertEqual(xml_adapter.xml_with_pre.aop_pid, "REG-AOP") + + def test_does_not_overwrite_existing_v2_or_aop_pid(self): + pid = "a" * 23 + xml_v2 = "b" * 23 + xml_aop = "c" * 23 + xml_adapter = self._adapter(v3=pid, v2=xml_v2, aop_pid=xml_aop) + registered = MagicMock(v3=pid, v2="REG-V2", aop_pid="REG-AOP") + with patch.object(PidProviderXML, "get_valid_pid_v3", return_value=pid): + xml_changed = PidProviderXML.complete_missing_xml_pids( + xml_adapter, registered=registered, auto_solve_pid_conflict=True + ) + self.assertEqual(xml_changed, {}) + self.assertEqual(xml_adapter.xml_with_pre.v2, xml_v2) + self.assertEqual(xml_adapter.xml_with_pre.aop_pid, xml_aop) + + +class IsAopTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="isaop-user", password="x") + + def test_true_when_no_volume_and_no_number(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", volume=None, number=None) + self.assertTrue(ppx.is_aop) + + def test_false_when_volume_present(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="B", volume="10") + self.assertFalse(ppx.is_aop) + + def test_false_when_number_present(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="C", number="2") + self.assertFalse(ppx.is_aop) + + +class RecordStatusAndCreatedUpdatedTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="recstatus-user", password="x") + + def test_freshly_created_object_reports_created_status(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + status = ppx.record_status + self.assertEqual(status["record_status"], "created") + self.assertIn("created", status) + self.assertIn("updated", status) + + def test_updated_long_after_creation_reports_updated_status(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + ppx.updated = ppx.created + timedelta(seconds=10) + self.assertEqual(ppx.record_status["record_status"], "updated") + + def test_unsaved_instance_has_empty_record_status(self): + ppx = PidProviderXML(v3="A") + self.assertEqual(ppx.record_status, {}) + + def test_created_updated_prefers_updated(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + self.assertEqual(ppx.created_updated, ppx.updated) + + def test_created_updated_falls_back_to_created_when_no_updated(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + ppx.updated = None + self.assertEqual(ppx.created_updated, ppx.created) + + +class GetReadableDataTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="readable-user", password="x") + + def test_returns_stored_readable_data_without_partial_body(self): + ppx = PidProviderXML.objects.create( + creator=self.user, + v3="A", + readable_data={"article_titles": ["T"], "partial_body": "legacy"}, + ) + result = ppx.get_readable_data() + self.assertNotIn("partial_body", result) + self.assertEqual(result["article_titles"], ["T"]) + + def test_returns_empty_dict_when_no_readable_data_and_no_current_version(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", readable_data=None) + self.assertEqual(ppx.get_readable_data(), {}) + # xml_with_pre falhou (sem current_version) -> marca como inválido + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, "NVALID") + + +class DataToCompareTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="datacompare-user", password="x") + + def test_includes_titles_and_body_fragment_when_present(self): + ppx = PidProviderXML.objects.create( + creator=self.user, + v3="A", + z_surnames="Silva", + z_partial_body="hash-1", + readable_data={"article_titles": ["T1"], "body_fragment": "frag"}, + ) + data = ppx.data_to_compare + self.assertEqual(data["article_titles"], ["T1"]) + self.assertEqual(data["body_fragment"], "frag") + self.assertEqual(data["z_surnames"], "Silva") + self.assertEqual(data["body_fragment_fingerprint"], "hash-1") + + def test_omits_titles_and_body_fragment_when_absent(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", readable_data=None) + data = ppx.data_to_compare + self.assertNotIn("article_titles", data) + self.assertNotIn("body_fragment", data) + + +class DataPropertyTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="dataprop-user", password="x") + + def test_assembles_expected_keys(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="V3X", v2="V2X", aop_pid="AOPX", pkg_name="pkg", + registered_in_core=True, + ) + data = ppx.data + self.assertEqual(data["v3"], "V3X") + self.assertEqual(data["v2"], "V2X") + self.assertEqual(data["aop_pid"], "AOPX") + self.assertEqual(data["pkg_name"], "pkg") + self.assertTrue(data["registered_in_core"]) + self.assertEqual(data["ppx_id"], ppx.id) + self.assertIn("registered_data", data) + self.assertEqual(data["record_status"], "created") + + +class GetByPidV3Tests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="getbypid-user", password="x") + + def test_returns_single_match(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-UNIQUE") + found = PidProviderXML.get_by_pid_v3("V3-UNIQUE") + self.assertEqual(found.pk, ppx.pk) + + def test_raises_does_not_exist_when_no_match(self): + with self.assertRaises(PidProviderXML.DoesNotExist): + PidProviderXML.get_by_pid_v3("NAO-EXISTE") + + def test_multiple_matches_falls_back_to_most_recently_updated(self): + older = PidProviderXML.objects.create(creator=self.user, v3="V3-DUP", pkg_name="older") + newer = PidProviderXML.objects.create(creator=self.user, v3="V3-DUP", pkg_name="newer") + newer.save() # garante 'updated' mais recente que 'older' + + found = PidProviderXML.get_by_pid_v3("V3-DUP") + + self.assertEqual(found.pk, newer.pk) + + +class MarkAsWaitingDoneTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="markstatus-user", password="x") + + def test_mark_as_waiting_updates_status_and_persists(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", proc_status="TODO") + ppx.mark_as_waiting() + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, "WAIT") + + def test_mark_as_waiting_is_noop_when_already_waiting(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", proc_status="WAIT") + with patch.object(PidProviderXML, "save") as mock_save: + ppx.mark_as_waiting() + mock_save.assert_not_called() + + def test_mark_as_done_updates_status_and_persists(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", proc_status="TODO") + ppx.mark_as_done() + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, "DONE") + + def test_mark_as_done_is_noop_when_already_done(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", proc_status="DONE") + with patch.object(PidProviderXML, "save") as mock_save: + ppx.mark_as_done() + mock_save.assert_not_called() + + +class GetQuerysetTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="getqueryset-user", password="x") + self.match = PidProviderXML.objects.create( + creator=self.user, v3="A", issn_electronic="0000-1111", pub_year="2026", + proc_status="TODO", + ) + self.other_issn = PidProviderXML.objects.create( + creator=self.user, v3="B", issn_electronic="9999-9999", pub_year="2026", + proc_status="DONE", + ) + self.other_year = PidProviderXML.objects.create( + creator=self.user, v3="C", issn_electronic="0000-1111", pub_year="1999", + ) + + def test_filters_by_issn_list(self): + qs = PidProviderXML.get_queryset(issn_list=["0000-1111"]) + ids = set(qs.values_list("id", flat=True)) + self.assertIn(self.match.id, ids) + self.assertNotIn(self.other_issn.id, ids) + + def test_filters_by_pub_year_range(self): + qs = PidProviderXML.get_queryset( + issn_list=["0000-1111"], from_pub_year="2020", until_pub_year="2030" + ) + ids = set(qs.values_list("id", flat=True)) + self.assertIn(self.match.id, ids) + self.assertNotIn(self.other_year.id, ids) + + def test_filters_by_proc_status_list(self): + qs = PidProviderXML.get_queryset(proc_status_list=["TODO"]) + ids = set(qs.values_list("id", flat=True)) + self.assertIn(self.match.id, ids) + self.assertNotIn(self.other_issn.id, ids) + + +class DeleteQuerysetTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="delqueryset-user", password="x") + + def test_deletes_items_and_their_other_pid_records(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + version = ppx.current_version + OtherPid.objects.create( + creator=self.user, pid_provider_xml=ppx, pid_type="pid_v3", + pid_in_xml="OLD", version=version, + ) + + qs = PidProviderXML.objects.filter(pk=ppx.pk) + PidProviderXML.delete_queryset(qs) + + self.assertFalse(PidProviderXML.objects.filter(pk=ppx.pk).exists()) + self.assertFalse(OtherPid.objects.filter(pid_provider_xml_id=ppx.pk).exists()) diff --git a/pid_provider/tests/test_pid_provider_xml_maintenance.py b/pid_provider/tests/test_pid_provider_xml_maintenance.py new file mode 100644 index 00000000..d379141b --- /dev/null +++ b/pid_provider/tests/test_pid_provider_xml_maintenance.py @@ -0,0 +1,482 @@ +""" +Testes para operações de manutenção/consulta de PidProviderXML +(pid_provider/models.py): _is_registered_pid, get_record_by_pid_v3, +add_collections, mark_items_as_invalid, find_duplicated_pkg_names, +mark_items_as_duplicated, deduplicate_items, fix_duplicated_pkg_name, +fix_pkg_name, is_registered, public_items, xml_with_pre/get_xml_with_pre, +fix_pid_v2. +""" +from datetime import timedelta +from unittest.mock import MagicMock, patch + +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.utils import timezone + +from collection.models import Collection +from journal.models import Journal, OfficialJournal +from pid_provider import choices, exceptions +from pid_provider.models import OtherPid, PidProviderXML, XMLVersion + +User = get_user_model() + + +class IsRegisteredPidTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="isregpid-user", password="x") + + def test_returns_none_when_no_pid_given(self): + self.assertIsNone(PidProviderXML._is_registered_pid()) + + def test_true_when_v3_matches_directly(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-DIRECT") + self.assertTrue(PidProviderXML._is_registered_pid(v3="V3-DIRECT")) + + def test_true_when_v3_matches_other_pid(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-CURRENT") + OtherPid.objects.create( + creator=self.user, pid_provider_xml=ppx, pid_type="pid_v3", pid_in_xml="V3-LEGACY" + ) + self.assertTrue(PidProviderXML._is_registered_pid(v3="V3-LEGACY")) + + def test_false_when_v3_unused(self): + self.assertFalse(PidProviderXML._is_registered_pid(v3="NAO-EXISTE")) + + def test_true_when_v2_matches_directly(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-X", v2="V2-DIRECT") + self.assertTrue(PidProviderXML._is_registered_pid(v2="V2-DIRECT")) + + def test_true_when_aop_pid_matches_v2_or_aop_pid_field(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-Y", aop_pid="AOP-1") + self.assertTrue(PidProviderXML._is_registered_pid(aop_pid="AOP-1")) + PidProviderXML.objects.create(creator=self.user, v3="V3-Z", v2="AOP-2") + self.assertTrue(PidProviderXML._is_registered_pid(aop_pid="AOP-2")) + + +class GetRecordByPidV3Tests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="getrecord-user", password="x") + + def test_raises_value_error_when_adapter_has_no_v3(self): + xml_adapter = MagicMock(v3=None) + with self.assertRaises(ValueError): + PidProviderXML.get_record_by_pid_v3(xml_adapter) + + def test_raises_does_not_exist_when_pid_is_unused(self): + xml_adapter = MagicMock(v3="UNUSED-PID") + xml_adapter.get_data_to_compare.return_value = {} + xml_adapter.xml_with_pre.body_fragment_fingerprint = None + with self.assertRaises(PidProviderXML.DoesNotExist): + PidProviderXML.get_record_by_pid_v3(xml_adapter) + + def test_returns_registered_when_best_match_approves(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-MATCH") + xml_adapter = MagicMock(v3="V3-MATCH") + xml_adapter.get_data_to_compare.return_value = {} + xml_adapter.xml_with_pre.body_fragment_fingerprint = None + + with patch.object( + PidProviderXML, "get_best_match", return_value={"registered": ppx} + ): + result = PidProviderXML.get_record_by_pid_v3(xml_adapter) + + self.assertEqual(result, ppx) + + def test_raises_conflict_when_best_match_does_not_approve(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-CONFLICT") + xml_adapter = MagicMock(v3="V3-CONFLICT") + xml_adapter.get_data_to_compare.return_value = {} + xml_adapter.xml_with_pre.body_fragment_fingerprint = None + + with patch.object(PidProviderXML, "get_best_match", return_value={}): + with self.assertRaises(Exception) as ctx: + PidProviderXML.get_record_by_pid_v3(xml_adapter) + from pid_provider.models import PidProviderXMLPidV3ConflictError + self.assertIsInstance(ctx.exception, PidProviderXMLPidV3ConflictError) + + +class MarkItemsAsInvalidTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="markinvalid-user", password="x") + + def test_marks_items_without_valid_xml_as_invalid(self): + # sem current_version -> xml_with_pre falha -> valid=False -> marca NVALID + ppx = PidProviderXML.objects.create( + creator=self.user, v3="A", issn_print="1111-1111", proc_status="TODO" + ) + + PidProviderXML.mark_items_as_invalid(["1111-1111"]) + + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, choices.PPXML_STATUS_INVALID) + + def test_does_not_touch_items_with_other_issns(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="B", issn_print="2222-2222", proc_status="TODO" + ) + + PidProviderXML.mark_items_as_invalid(["1111-1111"]) + + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, "TODO") + + def test_valid_item_is_left_untouched(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="C", issn_print="3333-3333", proc_status="TODO" + ) + with patch( + "pid_provider.models.PidProviderXML.xml_with_pre", + new_callable=lambda: property(lambda self: "
"), + ): + PidProviderXML.mark_items_as_invalid(["3333-3333"]) + + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, "TODO") + + +class FindAndMarkDuplicatedTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="dup-user", password="x") + + def test_find_duplicated_pkg_names_detects_repeated_names(self): + PidProviderXML.objects.create( + creator=self.user, v3="A", issn_print="4444-4444", pkg_name="pkg-dup" + ) + PidProviderXML.objects.create( + creator=self.user, v3="B", issn_print="4444-4444", pkg_name="pkg-dup" + ) + PidProviderXML.objects.create( + creator=self.user, v3="C", issn_print="4444-4444", pkg_name="pkg-unique" + ) + + result = PidProviderXML.find_duplicated_pkg_names(["4444-4444"]) + + self.assertEqual(result, ["pkg-dup"]) + + def test_find_duplicated_pkg_names_excludes_already_flagged_items(self): + PidProviderXML.objects.create( + creator=self.user, v3="D", issn_print="5555-5555", pkg_name="pkg-x", + proc_status=choices.PPXML_STATUS_DUPLICATED, + ) + PidProviderXML.objects.create( + creator=self.user, v3="E", issn_print="5555-5555", pkg_name="pkg-x", + proc_status=choices.PPXML_STATUS_DUPLICATED, + ) + + result = PidProviderXML.find_duplicated_pkg_names(["5555-5555"]) + + self.assertEqual(result, []) + + def test_mark_items_as_duplicated_flags_all_matching_pkg_name(self): + a = PidProviderXML.objects.create( + creator=self.user, v3="F", issn_print="6666-6666", pkg_name="pkg-y" + ) + b = PidProviderXML.objects.create( + creator=self.user, v3="G", issn_print="6666-6666", pkg_name="pkg-y" + ) + + result = PidProviderXML.mark_items_as_duplicated(["6666-6666"]) + + self.assertEqual(result, ["pkg-y"]) + a.refresh_from_db() + b.refresh_from_db() + self.assertEqual(a.proc_status, choices.PPXML_STATUS_DUPLICATED) + self.assertEqual(b.proc_status, choices.PPXML_STATUS_DUPLICATED) + + def test_mark_items_as_duplicated_returns_none_when_nothing_duplicated(self): + result = PidProviderXML.mark_items_as_duplicated(["7777-7777"]) + self.assertIsNone(result) + + +class FixDuplicatedPkgNameTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixdup-user", password="x") + + def test_keeps_most_recent_and_records_other_pid_for_the_rest(self): + """ + OtherPid.get_or_create EXIGE version truthy -- por isso `older` + precisa de um current_version real, senão a chamada dentro de + fix_duplicated_pkg_name levanta ValueError, que é silenciosamente + capturado pelo `except Exception` do próprio método (vira + UnexpectedEvent, sem propagar), deixando o OtherPid por criar. + """ + older = PidProviderXML.objects.create( + creator=self.user, v3="V3-OLDER", pkg_name="pkg-fix" + ) + older_version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=older) + older.current_version = older_version + older.save() + + newer = PidProviderXML.objects.create( + creator=self.user, v3="V3-NEWER", pkg_name="pkg-fix" + ) + newer.save() # garante updated mais recente + + PidProviderXML.fix_duplicated_pkg_name("pkg-fix", self.user) + + newer.refresh_from_db() + self.assertEqual(newer.proc_status, choices.PPXML_STATUS_DEDUPLICATED) + self.assertTrue( + OtherPid.objects.filter( + pid_provider_xml=newer, pid_type="pid_v3", pid_in_xml="V3-OLDER" + ).exists() + ) + + def test_swallows_error_when_older_item_has_no_current_version(self): + """ + Documenta o comportamento real: sem current_version, o OtherPid do + item mais antigo não pode ser criado (version obrigatório) -- + fix_duplicated_pkg_name captura a exceção internamente (via + UnexpectedEvent.create) e não propaga, mas também não cria o + OtherPid nem levanta erro para o chamador. + """ + PidProviderXML.objects.create( + creator=self.user, v3="V3-OLDER-NOVERSION", pkg_name="pkg-fix-noversion" + ) + newer = PidProviderXML.objects.create( + creator=self.user, v3="V3-NEWER-NOVERSION", pkg_name="pkg-fix-noversion" + ) + newer.save() + + PidProviderXML.fix_duplicated_pkg_name("pkg-fix-noversion", self.user) # não deve propagar + + newer.refresh_from_db() + self.assertEqual(newer.proc_status, choices.PPXML_STATUS_DEDUPLICATED) + self.assertFalse( + OtherPid.objects.filter( + pid_provider_xml=newer, pid_in_xml="V3-OLDER-NOVERSION" + ).exists() + ) + + def test_noop_when_less_than_two_items_share_pkg_name(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-SOLO", pkg_name="pkg-solo") + result = PidProviderXML.fix_duplicated_pkg_name("pkg-solo", self.user) + self.assertEqual(result, 0) + + +class FixPkgNameTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpkgname-user", password="x") + + def test_updates_when_pkg_name_differs(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", pkg_name="old-name") + result = ppx.fix_pkg_name("new-name") + self.assertTrue(result) + ppx.refresh_from_db() + self.assertEqual(ppx.pkg_name, "new-name") + + def test_returns_false_when_pkg_name_unchanged(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", pkg_name="same-name") + result = ppx.fix_pkg_name("same-name") + self.assertFalse(result) + + def test_falls_back_to_xml_with_pre_sps_pkg_name_when_no_pkg_name_given(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", pkg_name="old-name") + fake_xml_with_pre = MagicMock(sps_pkg_name="derived-name") + with patch.object( + PidProviderXML, "xml_with_pre", new_callable=lambda: property(lambda self: fake_xml_with_pre) + ): + result = ppx.fix_pkg_name(None) + self.assertTrue(result) + self.assertEqual(ppx.pkg_name, "derived-name") + + +class IsRegisteredTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="isregistered-user", password="x") + + def test_returns_registered_false_when_not_found(self): + xml_with_pre = MagicMock() + xml_with_pre.data = {} + xml_with_pre.filename = "file.xml" + + with patch("packtools.sps.pid_provider.xml_sps_adapter.PidProviderXMLAdapter") as MockAdapter: + mock_adapter = MockAdapter.return_value + mock_adapter.data = {} + with patch.object(PidProviderXML, "select_records", return_value=iter([])): + response = PidProviderXML.is_registered(xml_with_pre) + + self.assertFalse(response["registered"]) + self.assertEqual(response["filename"], "file.xml") + + def test_returns_registered_true_with_is_equal_flag(self): + xml_with_pre = MagicMock() + xml_with_pre.data = {} + + registered = MagicMock() + registered.data = {"v3": "V3-X"} + registered.readable_data = {"x": 1} + registered.is_equal_to.return_value = True + + with patch("packtools.sps.pid_provider.xml_sps_adapter.PidProviderXMLAdapter") as MockAdapter: + MockAdapter.return_value.data = {} + with patch.object(PidProviderXML, "select_records", return_value=iter([])), \ + patch.object( + PidProviderXML, "select_record", + return_value={"registered": registered}, + ): + response = PidProviderXML.is_registered(xml_with_pre) + + self.assertTrue(response["registered"]) + self.assertTrue(response["is_equal"]) + + def test_is_equal_forced_false_when_no_readable_data(self): + xml_with_pre = MagicMock() + xml_with_pre.data = {} + + registered = MagicMock() + registered.data = {"v3": "V3-X"} + registered.readable_data = None + registered.is_equal_to.return_value = True + + with patch("packtools.sps.pid_provider.xml_sps_adapter.PidProviderXMLAdapter") as MockAdapter: + MockAdapter.return_value.data = {} + with patch.object(PidProviderXML, "select_records", return_value=iter([])), \ + patch.object( + PidProviderXML, "select_record", + return_value={"registered": registered}, + ): + response = PidProviderXML.is_registered(xml_with_pre) + + self.assertFalse(response["is_equal"]) + + def test_returns_error_keys_on_unexpected_exception(self): + xml_with_pre = MagicMock() + xml_with_pre.data = {} + + with patch( + "packtools.sps.pid_provider.xml_sps_adapter.PidProviderXMLAdapter", + side_effect=ValueError("boom"), + ): + response = PidProviderXML.is_registered(xml_with_pre) + + self.assertIn("error_msg", response) + self.assertIn("error_type", response) + + +class PublicItemsTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="public-user", password="x") + + def test_excludes_items_without_current_version_pid_v3(self): + PidProviderXML.objects.create(creator=self.user, v3="A") + from_date = (timezone.now() - timedelta(days=1)).isoformat() + + items = list(PidProviderXML.public_items(from_date)) + + self.assertEqual(items, []) + + def test_includes_available_and_recently_updated_item(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + + from_date = (timezone.now() - timedelta(days=1)).isoformat() + items = list(PidProviderXML.public_items(from_date)) + + self.assertIn(ppx.id, [item.id for item in items]) + + def test_excludes_item_not_yet_available(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="A", + available_since=(timezone.now() + timedelta(days=30)).isoformat()[:10], + ) + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + + from_date = (timezone.now() - timedelta(days=1)).isoformat() + items = list(PidProviderXML.public_items(from_date)) + + self.assertNotIn(ppx.id, [item.id for item in items]) + + +class XmlWithPreInstancePropertyTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="xmlwithpre-user", password="x") + + def test_returns_none_and_marks_invalid_when_current_version_missing(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A", proc_status="TODO") + + result = ppx.xml_with_pre + + self.assertIsNone(result) + ppx.refresh_from_db() + self.assertEqual(ppx.proc_status, choices.PPXML_STATUS_INVALID) + + def test_returns_current_version_xml_with_pre_when_available(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="A") + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + + fake_xml_with_pre = MagicMock() + with patch.object( + XMLVersion, "xml_with_pre", new_callable=lambda: property(lambda self: fake_xml_with_pre) + ): + self.assertIs(ppx.xml_with_pre, fake_xml_with_pre) + + def test_get_xml_with_pre_returns_none_when_v3_missing(self): + self.assertIsNone(PidProviderXML.get_xml_with_pre("NAO-EXISTE")) + + def test_get_xml_with_pre_delegates_to_instance_property(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-GET") + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + + fake_xml_with_pre = MagicMock() + with patch.object( + XMLVersion, "xml_with_pre", new_callable=lambda: property(lambda self: fake_xml_with_pre) + ): + self.assertIs(PidProviderXML.get_xml_with_pre("V3-GET"), fake_xml_with_pre) + + +class FixPidV2MethodTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="fixpidv2method-user", password="x") + + def test_raises_does_not_exist_with_context_when_pid_missing(self): + with self.assertRaises(PidProviderXML.DoesNotExist): + PidProviderXML.fix_pid_v2(self.user, "NAO-EXISTE", "V2-NEW") + + def test_returns_data_unchanged_when_correct_equals_current(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-SAME", v2="V2-SAME") + + result = PidProviderXML.fix_pid_v2(self.user, "V3-SAME", "V2-SAME") + + self.assertEqual(result["v2"], "V2-SAME") + + def test_updates_v2_and_current_version_xml(self): + ppx = PidProviderXML.objects.create(creator=self.user, v3="V3-FIX", v2="V2-OLD") + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + + fake_xml_with_pre = MagicMock() + with patch.object( + XMLVersion, "xml_with_pre", new_callable=lambda: property(lambda self: fake_xml_with_pre) + ), patch.object(PidProviderXML, "_add_current_version") as mock_add_version: + result = PidProviderXML.fix_pid_v2(self.user, "V3-FIX", "V2-NEW") + + self.assertEqual(fake_xml_with_pre.v2, "V2-NEW") + mock_add_version.assert_called_once() + ppx.refresh_from_db() + self.assertEqual(ppx.v2, "V2-NEW") + self.assertEqual(result["v2"], "V2-NEW") + + def test_wraps_unexpected_error_in_fix_pid_v2_error(self): + PidProviderXML.objects.create(creator=self.user, v3="V3-ERR", v2="V2-OLD") + + with patch.object( + PidProviderXML, "_add_current_version", side_effect=ValueError("boom") + ), patch.object( + XMLVersion, "xml_with_pre", new_callable=lambda: property(lambda self: MagicMock()) + ), self.assertRaises(exceptions.PidProviderXMLFixPidV2Error): + # precisa de current_version setado para chegar em xml_with_pre + ppx = PidProviderXML.objects.get(v3="V3-ERR") + version = XMLVersion.objects.create(creator=self.user, pid_provider_xml=ppx) + ppx.current_version = version + ppx.save() + PidProviderXML.fix_pid_v2(self.user, "V3-ERR", "V2-NEW") diff --git a/pid_provider/tests/test_pid_provider_xml_registration.py b/pid_provider/tests/test_pid_provider_xml_registration.py new file mode 100644 index 00000000..56296bd9 --- /dev/null +++ b/pid_provider/tests/test_pid_provider_xml_registration.py @@ -0,0 +1,101 @@ +""" +Testes para PidProviderXMLRegistration (pid_provider/models.py). + +O caso "skipped não grava detail" já está coberto em test_models.py; aqui +cobrimos _serialize_detail isoladamente e os demais status (que DEVEM +gravar detail), incluindo a substituição de detail["registered"] (um +objeto PidProviderXML) por {"id", "v3"} para permitir serialização JSON. +""" +from django.contrib.auth import get_user_model +from django.test import TestCase + +from pid_provider.models import PidProviderXML, PidProviderXMLRegistration + +User = get_user_model() + + +class SerializeDetailTests(TestCase): + def test_returns_none_for_falsy_detail(self): + self.assertIsNone(PidProviderXMLRegistration._serialize_detail(None)) + self.assertIsNone(PidProviderXMLRegistration._serialize_detail({})) + + def test_replaces_registered_object_with_id_and_v3(self): + user = User.objects.create_user(username="serialize-user", password="x") + ppx = PidProviderXML.objects.create(creator=user, v3="V3-123") + + result = PidProviderXMLRegistration._serialize_detail( + {"registered": ppx, "other": "value"} + ) + + self.assertEqual(result["registered"], {"id": ppx.id, "v3": "V3-123"}) + self.assertEqual(result["other"], "value") + + def test_keeps_registered_untouched_when_not_a_model_instance(self): + result = PidProviderXMLRegistration._serialize_detail({"registered": None}) + self.assertIsNone(result["registered"]) + + def test_does_not_mutate_original_dict(self): + user = User.objects.create_user(username="serialize-user2", password="x") + ppx = PidProviderXML.objects.create(creator=user, v3="V3-456") + original = {"registered": ppx} + + PidProviderXMLRegistration._serialize_detail(original) + + self.assertIs(original["registered"], ppx) + + +class RecordTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="record-user", password="x") + + def test_lightweight_statuses_do_not_store_detail(self): + for status in PidProviderXMLRegistration.LIGHTWEIGHT_STATUSES: + registration = PidProviderXMLRegistration.record( + user=self.user, event_status=status, detail={"payload": "x"} + ) + self.assertIsNone(registration.detail) + self.assertEqual(registration.event_status, status) + + def test_non_lightweight_status_stores_serialized_detail(self): + registration = PidProviderXMLRegistration.record( + user=self.user, + event_status=PidProviderXMLRegistration.EVENT_CONFLICT, + detail={"error_msg": "conflict!"}, + ) + self.assertEqual(registration.detail, {"error_msg": "conflict!"}) + + def test_pkg_name_falls_back_to_pid_provider_xml_pkg_name(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="V3-789", pkg_name="pkg-from-ppx" + ) + + registration = PidProviderXMLRegistration.record( + user=self.user, + event_status=PidProviderXMLRegistration.EVENT_UNMATCHED, + pid_provider_xml=ppx, + ) + + self.assertEqual(registration.pkg_name, "pkg-from-ppx") + + def test_explicit_pkg_name_overrides_pid_provider_xml_pkg_name(self): + ppx = PidProviderXML.objects.create( + creator=self.user, v3="V3-999", pkg_name="pkg-from-ppx" + ) + + registration = PidProviderXMLRegistration.record( + user=self.user, + event_status=PidProviderXMLRegistration.EVENT_ERROR, + pid_provider_xml=ppx, + pkg_name="pkg-explicit", + ) + + self.assertEqual(registration.pkg_name, "pkg-explicit") + + def test_str_representation(self): + registration = PidProviderXMLRegistration.record( + user=self.user, + event_status=PidProviderXMLRegistration.EVENT_ERROR, + pkg_name="pkg-str", + ) + self.assertIn("pkg-str", str(registration)) + self.assertIn("error", str(registration)) diff --git a/pid_provider/tests/test_query_params.py b/pid_provider/tests/test_query_params.py index 6262bb11..98f168f7 100644 --- a/pid_provider/tests/test_query_params.py +++ b/pid_provider/tests/test_query_params.py @@ -4,15 +4,20 @@ Atualizado para cobrir a correção do falso-match na branch journal-article: - QueryBuilderPidProviderXML agora também lê - `xml_adapter.xml_with_pre.body_fragment_fingerprint` (fingerprint parcial do body do artigo) - e também `xml_adapter.xml_with_pre.body_fingerprint` (fingerprint corpo INTEIRO do artigo) - diretamente do xml_with_pre. + `xml_adapter.xml_with_pre.body_fragment_fingerprint` (fingerprint de um + fragmento estável do corpo) diretamente do xml_with_pre — sem depender de + mudança no PidProviderXMLAdapter/packtools. - `article_data_query` não usa mais z_partial_body isolado: delega ao novo `partial_body_query`, que monta `z_partial_body__in=[...]` com os hashes disponíveis (legado + novo fingerprint) ou, quando nenhum dos dois existe no XML de entrada, `z_partial_body__isnull=True` — nunca `z_partial_body__in=(None, None)`, que em SQL jamais casaria com candidatos NULL (NULL = NULL é UNKNOWN, não True). +- QueryBuilderPidProviderXML agora lê xml_adapter.xml_with_pre.readable_data + (não mais get_article_data(300)), que expõe "body_fragment" no lugar de + "partial_body". +- compare() trata labels ausentes em input_data como None via .get(label) + (não os pula) — comportamento coberto em CompareTests. ATENÇÃO: ajuste o caminho de import abaixo (`pid_provider.query_params`) para o módulo real onde essas classes/funções estão definidas no projeto, @@ -47,9 +52,9 @@ def make_xml_adapter( surnames=None, collab=None, links=None, - partial_body=None, - body_fingerprint=None, + body_fragment=None, body_fragment_fingerprint=None, + body_fingerprint=None, ): """ Monta um mock de xml_adapter com a forma esperada por @@ -57,26 +62,40 @@ def make_xml_adapter( body_fragment_fingerprint: valor de xml_adapter.xml_with_pre.body_fragment_fingerprint, o novo sinal - (hash do corpo inteiro do artigo) usado em partial_body_query. + (hash de um fragmento estável do corpo) usado em partial_body_query. + + body_fragment: valor de xml_adapter.xml_with_pre.readable_data["body_fragment"], + usado por validate_input_data. Substitui o antigo "partial_body", que + não existe mais em readable_data. """ adapter = MagicMock() adapter.data = data or {} - adapter.get_data_to_compare.return_value = {} adapter.v3 = v3 adapter.v2 = v2 adapter.aop_pid = aop_pid adapter.pkg_name = pkg_name adapter.sps_pkg_name = sps_pkg_name adapter.order = order + # QueryBuilderPidProviderXML lê z_partial_body como ATRIBUTO direto do + # adapter (self.z_partial_body = xml_adapter.z_partial_body), não via + # xml_adapter.data.get("z_partial_body") -- por isso precisa ser + # configurado explicitamente aqui, senão vira um MagicMock não + # configurado (nunca None nem o valor esperado). + adapter.z_partial_body = (data or {}).get("z_partial_body") adapter.xml_with_pre.deprecated_sps_pkg_name_list = deprecated_sps_pkg_name_list or [] - adapter.xml_with_pre.body_fingerprint = body_fingerprint adapter.xml_with_pre.body_fragment_fingerprint = body_fragment_fingerprint - adapter.xml_with_pre.get_article_data.return_value = { + adapter.xml_with_pre.body_fingerprint = body_fingerprint + # QueryBuilderPidProviderXML.__init__ lê xml_with_pre.readable_data + # como ATRIBUTO (não mais xml_with_pre.get_article_data(...) chamado + # como método) — precisa ser um dict de verdade, não um MagicMock + # não configurado, senão validate_input_data quebra ao tentar iterar + # sobre um MagicMock. + adapter.xml_with_pre.readable_data = { "article_titles": article_titles or [], "surnames": surnames, "collab": collab, "links": links, - "partial_body": partial_body, + "body_fragment": body_fragment, } return adapter @@ -127,6 +146,18 @@ def test_passes_when_only_surnames_present(self): qbuilder = QueryBuilderPidProviderXML(adapter) qbuilder.validate_input_data() # não deve levantar + def test_passes_when_only_body_fragment_present(self): + """ + Cobre especificamente a chave nova "body_fragment" (antes + "partial_body"), que validate_input_data passou a checar. + """ + adapter = make_xml_adapter( + data={"pub_year": "2026", "issn_electronic": "0000-1111"}, + body_fragment="um fragmento de corpo qualquer", + ) + qbuilder = QueryBuilderPidProviderXML(adapter) + qbuilder.validate_input_data() # não deve levantar + def test_raises_not_enough_parameters_when_all_empty(self): adapter = make_xml_adapter( data={"pub_year": "2026", "issn_electronic": "0000-1111"}, @@ -145,6 +176,15 @@ def test_raises_not_enough_parameters_when_titles_are_blank(self): with self.assertRaises(exceptions.NotEnoughParametersToGetPidProviderXMLError): qbuilder.validate_input_data() + def test_raises_not_enough_parameters_when_body_fragment_is_blank(self): + adapter = make_xml_adapter( + data={"pub_year": "2026", "issn_electronic": "0000-1111"}, + body_fragment="", + ) + qbuilder = QueryBuilderPidProviderXML(adapter) + with self.assertRaises(exceptions.NotEnoughParametersToGetPidProviderXMLError): + qbuilder.validate_input_data() + class PkgNameListTests(SimpleTestCase): @@ -270,7 +310,7 @@ def test_with_order_adds_v2_endswith(self): class PartialBodyQueryTests(SimpleTestCase): """ Cobre especificamente o fix do incidente: z_partial_body agora aceita - dois formatos de hash (legado e fingerprint do corpo inteiro), e o + dois formatos de hash (legado e fingerprint de fragmento do corpo), e o caso "nenhum dos dois presente" precisa cair em isnull=True, nunca em __in=(None, None). """ @@ -288,23 +328,26 @@ def test_uses_in_with_only_legacy_partial_body(self): def test_uses_in_with_only_body_fragment_fingerprint(self): adapter = make_xml_adapter( data={}, - body_fragment_fingerprint="hash-fragmento-do-corpo", + body_fragment_fingerprint="hash-fragmento-corpo", ) qbuilder = QueryBuilderPidProviderXML(adapter) self.assertEqual( qbuilder.partial_body_query, - Q(z_partial_body__in=["hash-fragmento-do-corpo"]), + Q(z_partial_body__in=["hash-fragmento-corpo"]), ) def test_uses_in_with_both_hashes_when_both_present_and_different(self): adapter = make_xml_adapter( data={"z_partial_body": "hash-legado"}, - body_fragment_fingerprint="hash-fragmento-do-corpo", + body_fragment_fingerprint="hash-fragmento-corpo", ) qbuilder = QueryBuilderPidProviderXML(adapter) - self.assertEqual( - qbuilder.partial_body_query, - Q(z_partial_body__in={"hash-legado", "hash-fragmento-do-corpo"}), + expected = Q( + z_partial_body__in={"hash-legado", "hash-fragmento-corpo"} + ) + self.assertDictEqual( + dict(qbuilder.partial_body_query.children), + dict(expected.children), ) def test_deduplicates_when_both_hashes_are_equal(self): @@ -332,56 +375,20 @@ def test_uses_isnull_when_neither_hash_is_present(self): qbuilder.partial_body_query, Q(z_partial_body__in=(None, None)) ) - def test_combines_textual_fields_with_all_three_body_hashes(self): - """ - article_data_query propaga corretamente o candidato adicional - body_fingerprint (corpo inteiro) através de partial_body_query, - junto com z_partial_body legado e body_fragment_fingerprint. - """ + def test_ignores_body_fingerprint(self): adapter = make_xml_adapter( - data={ - "z_surnames": "Silva", - "z_collab": None, - "z_links": None, - "z_partial_body": "hash-legado", - }, - body_fragment_fingerprint="hash-fragmento-do-corpo", - body_fingerprint="hash-corpo-inteiro", + data={"z_partial_body": "hash-legado"}, + body_fragment_fingerprint="hash-fragmento-corpo", + body_fingerprint="hash-corpo-inteiro-que-nao-deve-ser-usado", ) qbuilder = QueryBuilderPidProviderXML(adapter) - expected = Q(z_surnames="Silva", z_collab=None, z_links=None) & Q( - z_partial_body__in={ - "hash-legado", - "hash-fragmento-do-corpo", - "hash-corpo-inteiro", - } - ) - self.assertDictEqual( - dict(qbuilder.article_data_query.children), - dict(expected.children) - ) - def test_two_articles_with_same_legacy_hash_but_different_body_fingerprint_differ(self): - """ - Regressão direta do incidente original: dois artigos com o mesmo - rótulo genérico em z_partial_body legado (ex. "ARTIGO DE REVISÃO") - mas com body_fingerprint (corpo inteiro) diferentes devem produzir - queries distintas — o falso-positivo de match não deve mais ocorrer. - """ - adapter_a = make_xml_adapter( - data={"z_partial_body": "rotulo-generico-artigo-revisao"}, - body_fragment_fingerprint=None, - body_fingerprint="hash-corpo-artigo-a", - ) - adapter_b = make_xml_adapter( - data={"z_partial_body": "rotulo-generico-artigo-revisao"}, - body_fragment_fingerprint=None, - body_fingerprint="hash-corpo-artigo-b", + expected = Q( + z_partial_body__in={"hash-legado", "hash-fragmento-corpo"} ) - qbuilder_a = QueryBuilderPidProviderXML(adapter_a) - qbuilder_b = QueryBuilderPidProviderXML(adapter_b) - self.assertNotEqual( - qbuilder_a.article_data_query, qbuilder_b.article_data_query + self.assertDictEqual( + dict(qbuilder.partial_body_query.children), + dict(expected.children), ) @@ -400,13 +407,16 @@ def test_combines_textual_fields_with_partial_body_query_both_hashes(self): "z_links": None, "z_partial_body": "hash-legado", }, - body_fragment_fingerprint="hash-fragmento-do-corpo", + body_fragment_fingerprint="hash-fragmento-corpo", ) qbuilder = QueryBuilderPidProviderXML(adapter) expected = Q(z_surnames="Silva", z_collab=None, z_links=None) & Q( - z_partial_body__in={"hash-legado", "hash-fragmento-do-corpo"} + z_partial_body__in={"hash-legado", "hash-fragmento-corpo"} + ) + self.assertDictEqual( + dict(qbuilder.article_data_query.children), + dict(expected.children), ) - self.assertEqual(qbuilder.article_data_query, expected) def test_falls_back_to_isnull_when_no_body_hash_available(self): adapter = make_xml_adapter(data={}, body_fragment_fingerprint=None) @@ -421,7 +431,7 @@ def test_two_different_articles_produce_different_queries(self): Regressão conceitual do incidente: dois artigos com hashes de corpo diferentes (mesmo que ambos tenham, no passado, colidido via z_partial_body legado genérico) agora produzem queries IN - distintas, pois o fingerprint do corpo inteiro entra na + distintas, pois o fingerprint do fragmento do corpo entra na composição. """ adapter_a = make_xml_adapter( @@ -558,10 +568,21 @@ def test_none_and_falsy_are_treated_as_equal(self): def test_different_scalars_uses_how_similar_and_includes_registered( self, mock_how_similar ): + """ + Quando score != 1, o response inclui "registered" E "input_data" + (não apenas "registered") -- ambos úteis para inspecionar a + divergência. + """ mock_how_similar.return_value = 0.4 result = compare_items("z_surnames", "Silva", "Souza") self.assertEqual( - result, {"label": "z_surnames", "score": 0.4, "registered": "Silva"} + result, + { + "label": "z_surnames", + "score": 0.4, + "registered": "Silva", + "input_data": "Souza", + }, ) mock_how_similar.assert_called_once_with("Souza", "Silva") @@ -572,7 +593,13 @@ def test_none_input_data_falls_back_to_empty_string_for_how_similar( mock_how_similar.return_value = 0.2 result = compare_items("z_links", "algum-link", None) self.assertEqual( - result, {"label": "z_links", "score": 0.2, "registered": "algum-link"} + result, + { + "label": "z_links", + "score": 0.2, + "registered": "algum-link", + "input_data": None, + }, ) mock_how_similar.assert_called_once_with("", "algum-link") @@ -582,11 +609,25 @@ def test_none_registered_falls_back_to_empty_string_for_how_similar( ): mock_how_similar.return_value = 0.3 result = compare_items("z_links", None, "algum-link") - self.assertEqual(result, {"label": "z_links", "score": 0.3, "registered": None}) + self.assertEqual( + result, + { + "label": "z_links", + "score": 0.3, + "registered": None, + "input_data": "algum-link", + }, + ) mock_how_similar.assert_called_once_with("algum-link", "") class CompareTests(SimpleTestCase): + """ + compare() usa input_data.get(label) para cada label de + registered_items -- um label ausente em input_data é tratado como + None (não é pulado): sempre gera uma entrada em "items", e conta no + cálculo de total_score/percentual_score via compare_items(None, ...). + """ @patch("pid_provider.query_params.how_similar") def test_aggregates_scores_from_all_items(self, mock_how_similar): @@ -600,11 +641,44 @@ def test_aggregates_scores_from_all_items(self, mock_how_similar): self.assertEqual(result["total_score"], 1.5) # 1 (match) + 0.5 (mocked) self.assertEqual(result["percentual_score"], 0.75) - def test_missing_input_key_is_treated_as_none(self): - registered_items = {"z_collab": None} - input_data = {} + def test_missing_input_key_is_treated_as_none_not_skipped(self): + """ + Um label ausente em input_data vira None via .get(label) -- se o + valor registrado também é falsy (None), compare_items considera + os dois "iguais" (score 1), então o label ausente ENTRA em items + e contribui com score 1, não é descartado. + """ + registered_items = {"z_collab": None, "z_surnames": "Silva"} + input_data = {"z_surnames": "Silva"} # z_collab ausente -> None + + result = compare(registered_items, input_data) + + self.assertEqual(len(result["items"]), 2) + labels = {item["label"] for item in result["items"]} + self.assertEqual(labels, {"z_collab", "z_surnames"}) + self.assertEqual(result["total_score"], 2) + self.assertEqual(result["percentual_score"], 1) + + def test_missing_input_key_with_truthy_registered_value_lowers_score(self): + """ + Se o label ausente em input_data tem um valor registrado truthy, + o None resultante de .get(label) NÃO é igual ao registrado -- + cai no ramo how_similar (não é match automático). + """ + registered_items = {"z_surnames": "Silva"} + input_data = {} # z_surnames ausente -> None result = compare(registered_items, input_data) - self.assertEqual(result["total_score"], 1) - self.assertEqual(result["percentual_score"], 1) \ No newline at end of file + self.assertEqual(len(result["items"]), 1) + self.assertEqual(result["items"][0]["label"], "z_surnames") + self.assertLess(result["items"][0]["score"], 1) + + def test_empty_registered_items_raises_zero_division_error(self): + """ + Único caso em que items fica vazio: registered_items já vem + vazio -- não há nada para iterar, então + total_score / len(items) levanta ZeroDivisionError. + """ + with self.assertRaises(ZeroDivisionError): + compare({}, {"z_surnames": "Silva"}) diff --git a/pid_provider/tests/test_register.py b/pid_provider/tests/test_register.py index 2d076456..263d8834 100644 --- a/pid_provider/tests/test_register.py +++ b/pid_provider/tests/test_register.py @@ -24,7 +24,7 @@ `PidProviderXMLRegistration.record` só acontece se: error_type (uma exceção foi capturada) OR select_record_response.get("matched_items") (havia ambiguidade) OR - PidProviderSetting.record_all_registration_events (configuração no Wagtail Admin ativada) + PidProviderSetting.record_all_registration_events (configuração ativa) Ou seja: um "created"/"updated"/"skipped" limpo, sem matches concorrentes e sem erro, NÃO gera registro de auditoria por padrão. @@ -61,12 +61,22 @@ 6. Os caminhos "conflict" e "unmatched" continuam setando `event_status` explicitamente antes de re-levantar a exceção. +7. **`register()` agora faz `input_data.update(xml_with_pre.readable_data)`** + em vez de `input_data.update(xml_with_pre.get_article_data())`, alinhado + à migração do packtools (readable_data substitui get_article_data como + fonte de dados legíveis, sem a chave partial_body). Por isso, + `make_xml_with_pre()` precisa configurar `readable_data` como um dict de + verdade — um MagicMock não configurado quebra dict.update() com + TypeError, silenciosamente capturado pelo except Exception externo de + register() e mascarado como event_status="error" em qualquer teste que + dependa do fluxo normal de sucesso. + Ajuste os caminhos de import (PATCH_BASE) conforme a estrutura do seu projeto. """ from unittest.mock import patch, MagicMock -from django.test import TestCase, override_settings +from django.test import TestCase from pid_provider import exceptions from pid_provider.models import ( @@ -81,15 +91,24 @@ def make_xml_with_pre(**overrides): """ XMLWithPre falso, com os atributos que register/build_readable_data tocam. + + IMPORTANTE: register() faz input_data.update(xml_with_pre.readable_data), + então readable_data precisa ser um dict de verdade — um MagicMock não + configurado quebra dict.update() com TypeError, silenciosamente + capturado pelo except Exception externo de register() e mascarado como + event_status="error" em qualquer teste que dependa do fluxo normal. """ m = MagicMock(name="xml_with_pre") m.data = {"pid_v3": overrides.get("v3"), "sps_pkg_name": "pkg-fake"} m.sps_pkg_name = overrides.get("sps_pkg_name", "pkg-fake") - m.authors = {"person": [{"surname": "SILVA"}]} - m.collab = None - m.links = [] - m.article_titles_texts = ["Some title"] - m.partial_body = "corpo parcial" + m.readable_data = { + "surnames": ["SILVA"], + "collab": None, + "links": [], + "article_titles": ["Some title"], + "body_fragment": "corpo parcial", + } + m.body_fragment_fingerprint = "fingerprint-fake" return m @@ -128,6 +147,12 @@ def setUp(self): self.m_record = self.p_record.start() self.addCleanup(self.p_record.stop) + # Mantém os testes independentes da configuração persistida. + self.p_setting = patch(f"{PATCH_BASE}.PidProviderSetting.load") + self.m_setting_load = self.p_setting.start() + self.m_setting_load.return_value.record_all_registration_events = False + self.addCleanup(self.p_setting.stop) + # -- helpers de asserção -------------------------------------------- def assert_recorded_status(self, expected_status): self.assertTrue( @@ -512,22 +537,159 @@ def test_record_called_exactly_once_on_skip(self): # --------------------------------------------------------------------------- -# Testes da setting PidProviderSetting no Wagtail Admin +# Configuração opcional: fluxos limpos também podem gerar auditoria # --------------------------------------------------------------------------- class RecordAllEventsSettingTest(RegisterTestBase): - @patch(f"{PATCH_BASE}.PidProviderSetting.load") - def test_record_called_on_clean_success_when_setting_enabled(self, m_setting_load): - m_setting_load.return_value.record_all_registration_events = True + def test_record_called_on_clean_success_when_setting_enabled(self): + self.m_setting_load.return_value.record_all_registration_events = True + with patch(f"{PATCH_BASE}.PidProviderXML.select_record") as m_select, \ patch(f"{PATCH_BASE}.PidProviderXML.complete_missing_xml_pids") as m_cmp, \ patch(f"{PATCH_BASE}.PidProviderXML.is_updated") as m_upd, \ patch(f"{PATCH_BASE}.PidProviderXML._save") as m_save: - m_select.side_effect = PidProviderXML.DoesNotExist + m_select.return_value = {} m_cmp.return_value = {} m_upd.return_value = None - m_save.return_value = MagicMock(data={"v3": "ABC"}) + saved = MagicMock(name="saved_ppx") + saved.data = {"v3": "ABC"} + m_save.return_value = saved - PidProviderXML.register(self.xml, "file.xml", self.user) + response = PidProviderXML.register(self.xml, "file.xml", self.user) + self.assertEqual(response.get("event_status"), "created") self.assert_recorded_status("created") + + +class RegisterResponseSchemaTest(RegisterTestBase): + """ + Teste de contrato: não valida os VALORES do response, só as CHAVES + presentes em cada nível. Serve como sentinela contra mudanças + silenciosas de schema (ex.: trocar get_article_data() por + readable_data alterou as chaves aninhadas em response["input_data"] + sem que nenhum teste anterior percebesse, pois testavam apenas + event_status/v3/etc., não a forma do dict completo). + + Se este teste quebrar, NÃO conserte só o teste: confirme que a + mudança de schema foi intencional e que os consumidores de + register() (endpoints da API, logs de auditoria) foram atualizados + de acordo. + """ + + def _register_clean_created(self): + with patch(f"{PATCH_BASE}.PidProviderXML.select_record") as m_select, \ + patch(f"{PATCH_BASE}.PidProviderXML.complete_missing_xml_pids") as m_cmp, \ + patch(f"{PATCH_BASE}.PidProviderXML.is_updated") as m_upd, \ + patch(f"{PATCH_BASE}.PidProviderXML._save") as m_save: + + m_select.return_value = {} + m_cmp.return_value = {} + m_upd.return_value = None + saved = MagicMock(name="saved_ppx") + saved.data = { + "v3": "ABC", + "record_status": "created", + "created": "2026-01-01T00:00:00", + } + m_save.return_value = saved + + return PidProviderXML.register(self.xml, "file.xml", self.user) + + def test_top_level_keys_on_clean_success(self): + response = self._register_clean_created() + # chaves esperadas no nível superior de um fluxo "created" limpo + expected_keys = { + "input_data", + "xml_adapter_data", + "xml_changed", + "v3", + "record_status", + "created", + "event_status", + } + self.assertEqual(set(response.keys()), expected_keys) + + def test_input_data_keys_reflect_readable_data_not_get_article_data(self): + """ + Sentinela específico para a migração get_article_data() -> + readable_data: garante que response["input_data"] contém as + chaves de readable_data (article_titles, surnames, collab, + links, body_fragment) e NÃO contém "partial_body", que só + existia no formato antigo (get_article_data()). + """ + response = self._register_clean_created() + input_data = response["input_data"] + + # chaves vindas de xml_with_pre.data + xml_with_pre.readable_data + # + "origin", conforme montado em register() + self.assertIn("article_titles", input_data) + self.assertIn("surnames", input_data) + self.assertIn("collab", input_data) + self.assertIn("links", input_data) + self.assertIn("body_fragment", input_data) + self.assertIn("origin", input_data) + + # a chave antiga não deve mais aparecer + self.assertNotIn("partial_body", input_data) + + def test_error_path_keys(self): + with patch(f"{PATCH_BASE}.PidProviderXML.select_record") as m_select: + m_select.side_effect = ValueError("falha inesperada") + response = PidProviderXML.register(self.xml, "file.xml", self.user) + + expected_keys = { + "input_data", + "xml_adapter_data", + "error_msg", + "error_type", + "traceback", + "event_status", + } + self.assertEqual(set(response.keys()), expected_keys) + + def test_conflict_path_keys(self): + existing = MagicMock(name="existing_ppx") + with patch(f"{PATCH_BASE}.PidProviderXML.select_record") as m_select, \ + patch(f"{PATCH_BASE}.PidProviderXML.complete_missing_xml_pids") as m_cmp: + + m_select.return_value = {"registered": existing} + m_cmp.side_effect = PidProviderXMLPidV3ConflictError("conflict!") + + response = PidProviderXML.register(self.xml, "file.xml", self.user) + + expected_keys = { + "input_data", + "xml_adapter_data", + "error_msg", + "error_type", + "traceback", + "event_status", + } + self.assertEqual(set(response.keys()), expected_keys) + + def test_skipped_path_keys(self): + existing = MagicMock(name="existing_ppx") + existing.data = {"v3": "ABC", "record_status": "updated"} + with patch(f"{PATCH_BASE}.PidProviderXML.select_record") as m_select, \ + patch(f"{PATCH_BASE}.PidProviderXML.complete_missing_xml_pids") as m_cmp, \ + patch(f"{PATCH_BASE}.PidProviderXML.is_updated") as m_upd, \ + patch(f"{PATCH_BASE}.PidProviderXML._save") as m_save: + + m_select.return_value = {"registered": existing} + m_cmp.return_value = {} + m_upd.side_effect = exceptions.SkipSavePidProviderXML + + response = PidProviderXML.register(self.xml, "file.xml", self.user) + + m_save.assert_not_called() + + expected_keys = { + "input_data", + "xml_adapter_data", + "v3", + "record_status", + "skipped", + "event_status", + "xml_changed", + } + self.assertEqual(set(response.keys()), expected_keys) diff --git a/pid_provider/tests/test_select_record.py b/pid_provider/tests/test_select_record.py index 52ed0e9d..7ab5323b 100644 --- a/pid_provider/tests/test_select_record.py +++ b/pid_provider/tests/test_select_record.py @@ -19,9 +19,19 @@ def _make_results(self, count): """Simula a lista de candidatos já materializada para um label.""" return [MagicMock(name=f"candidate_{i}") for i in range(count)] - def _make_xml_adapter(self, data_to_compare=None): + def _make_xml_adapter(self, data_to_compare=None, body_fragment_fingerprint=None): + """ + select_record não usa xml_adapter.get_data_to_compare() diretamente: + chama fix_get_data_to_compare(xml_adapter), que pega o retorno de + get_data_to_compare() e ACRESCENTA a chave + "body_fragment_fingerprint" (lida de + xml_adapter.xml_with_pre.body_fragment_fingerprint). Por isso esse + atributo precisa ser configurado explicitamente aqui -- senão vira + um MagicMock não configurado, tornando o dict final imprevisível. + """ xml_adapter = MagicMock() xml_adapter.get_data_to_compare.return_value = data_to_compare or {} + xml_adapter.xml_with_pre.body_fragment_fingerprint = body_fragment_fingerprint return xml_adapter @patch("pid_provider.models.PidProviderXML.get_best_match") @@ -204,13 +214,24 @@ def test_select_record_accumulates_actual_unmatched_lists_across_labels_when_non @patch("pid_provider.models.PidProviderXML.get_best_match") def test_select_record_passes_candidates_and_comparison_data_to_get_best_match(self, mock_get_best_match): - """get_best_match deve ser chamado com a lista de candidatos do label e os dados já processados do xml_adapter.""" + """ + get_best_match deve ser chamado com a lista de candidatos do label + e os dados já processados via fix_get_data_to_compare -- que é + get_data_to_compare() ACRESCIDO de "body_fragment_fingerprint" + (não o retorno cru de get_data_to_compare()). + """ candidates = self._make_results(1) - xml_adapter = self._make_xml_adapter(data_to_compare={"title": "Foo"}) + xml_adapter = self._make_xml_adapter( + data_to_compare={"title": "Foo"}, + body_fragment_fingerprint="fingerprint-fake", + ) mock_get_best_match.return_value = {"unmatched": ["ITEM_DATA"]} PidProviderXML.select_record(xml_adapter, [("journal", candidates)]) - mock_get_best_match.assert_called_once_with(candidates, {"title": "Foo"}) + mock_get_best_match.assert_called_once_with( + candidates, + {"title": "Foo", "body_fragment_fingerprint": "fingerprint-fake"}, + ) diff --git a/pid_provider/tests/test_xml_url.py b/pid_provider/tests/test_xml_url.py new file mode 100644 index 00000000..cfedb982 --- /dev/null +++ b/pid_provider/tests/test_xml_url.py @@ -0,0 +1,180 @@ +"""Testes para XMLURL (pid_provider/models.py).""" +import shutil +import tempfile +import zipfile +from unittest.mock import MagicMock, patch + +from django.contrib.auth import get_user_model +from django.db import IntegrityError +from django.test import TestCase, override_settings + +from pid_provider.models import XMLURL + +User = get_user_model() + +MEDIA_ROOT = tempfile.mkdtemp() + + +class XMLURLGetTests(TestCase): + def test_raises_value_error_without_url(self): + with self.assertRaises(ValueError): + XMLURL.get(url=None) + + def test_raises_does_not_exist_when_no_match(self): + with self.assertRaises(XMLURL.DoesNotExist): + XMLURL.get(url="https://example.org/missing") + + +class XMLURLCreateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="xmlurl-create", password="x") + + def test_creates_new_record(self): + obj = XMLURL.create( + self.user, + url="https://example.org/a.xml", + status="pending", + pid="PID-1", + detail={"foo": "bar"}, + is_public=True, + ) + self.assertIsNotNone(obj.pk) + self.assertEqual(obj.status, "pending") + self.assertTrue(obj.is_public) + + def test_falls_back_to_get_on_integrity_error(self): + existing = XMLURL.create(self.user, url="https://example.org/dup.xml") + + with patch.object(XMLURL, "save", side_effect=IntegrityError): + result = XMLURL.create(self.user, url="https://example.org/dup.xml") + + self.assertEqual(result.pk, existing.pk) + + +class XMLURLCreateOrUpdateTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="xmlurl-cou", password="x") + + def test_creates_when_none_exists(self): + obj = XMLURL.create_or_update( + self.user, url="https://example.org/new.xml", status="success" + ) + self.assertIsNotNone(obj.pk) + self.assertEqual(obj.status, "success") + + def test_updates_only_fields_explicitly_passed(self): + existing = XMLURL.create( + self.user, + url="https://example.org/upd.xml", + status="pending", + pid="OLD-PID", + is_public=False, + ) + + updated = XMLURL.create_or_update( + self.user, + url="https://example.org/upd.xml", + status="success", + # pid, detail, is_public não passados (None) -> preservados + ) + + self.assertEqual(updated.pk, existing.pk) + self.assertEqual(updated.status, "success") + self.assertEqual(updated.pid, "OLD-PID") + self.assertFalse(updated.is_public) + self.assertEqual(updated.updated_by, self.user) + + +@override_settings(MEDIA_ROOT=MEDIA_ROOT) +class XMLURLSaveFileTests(TestCase): + @classmethod + def tearDownClass(cls): + shutil.rmtree(MEDIA_ROOT, ignore_errors=True) + super().tearDownClass() + + def setUp(self): + self.user = User.objects.create_user(username="xmlurl-savefile", password="x") + self.obj = XMLURL.create(self.user, url="https://example.org/savefile.xml", pid="PID-9") + + def test_save_file_success_writes_readable_zip(self): + result = self.obj.save_file("
", filename="doc.xml") + + self.assertTrue(result) + self.obj.refresh_from_db() + self.assertTrue(self.obj.zipfile.name) + with zipfile.ZipFile(self.obj.zipfile.path) as zf: + self.assertEqual(zf.namelist(), ["doc.xml"]) + self.assertEqual(zf.read("doc.xml"), b"
") + + def test_save_file_accepts_bytes_content(self): + result = self.obj.save_file(b"
") + self.assertTrue(result) + + def test_save_file_returns_false_on_failure(self): + with patch("pid_provider.models.zipfile.ZipFile", side_effect=OSError("boom")): + result = self.obj.save_file("
") + self.assertFalse(result) + + +class XMLURLRecordTests(TestCase): + def setUp(self): + self.user = User.objects.create_user(username="xmlurl-record", password="x") + + def test_record_creates_new_with_pid_from_response(self): + obj = XMLURL.record( + user=self.user, + url="https://example.org/rec.xml", + status="success", + document_item={"status": "true"}, + response={"v3": "PID-FROM-RESPONSE"}, + ) + + self.assertEqual(obj.pid, "PID-FROM-RESPONSE") + self.assertEqual(obj.detail["response"], {"v3": "PID-FROM-RESPONSE"}) + self.assertTrue(obj.is_public) + + def test_record_is_public_false_when_status_is_the_string_false(self): + obj = XMLURL.record( + user=self.user, + url="https://example.org/rec2.xml", + status="success", + document_item={"status": "false"}, + ) + self.assertFalse(obj.is_public) + + def test_record_is_public_none_when_no_document_item(self): + obj = XMLURL.record( + user=self.user, + url="https://example.org/rec3.xml", + status="success", + document_item=None, + ) + self.assertIsNone(obj.is_public) + + def test_record_stores_exception_traceback_in_detail(self): + obj = XMLURL.record( + user=self.user, + url="https://example.org/rec4.xml", + status="xml_fetch_failed", + document_item={"info": "x"}, + exception=True, + ) + self.assertIn("exceptions", obj.detail) + + @override_settings(MEDIA_ROOT=MEDIA_ROOT) + def test_record_saves_file_when_xml_with_pre_given(self): + fake_xml_with_pre = MagicMock() + fake_xml_with_pre.tostring.return_value = "
" + + with patch.object(XMLURL, "save_file") as mock_save_file: + obj = XMLURL.record( + user=self.user, + url="https://example.org/rec5.xml", + status="success", + document_item={"status": "true"}, + response={"v3": "PID-5"}, + xml_with_pre=fake_xml_with_pre, + ) + + mock_save_file.assert_called_once_with("
", filename="PID-5") + self.assertEqual(obj.pid, "PID-5") diff --git a/pid_provider/tests/test_xml_version.py b/pid_provider/tests/test_xml_version.py new file mode 100644 index 00000000..b9d8e801 --- /dev/null +++ b/pid_provider/tests/test_xml_version.py @@ -0,0 +1,211 @@ +""" +Testes para XMLVersion (pid_provider/models.py). + +XMLVersion.create/get_or_create/save_file lidam com um FileField real, então +os testes que gravam arquivo usam override_settings(MEDIA_ROOT=) para +não sujar o MEDIA_ROOT do projeto. + +A property xml_with_pre (que depende de packtools.XMLWithPre.create) e a +cached_property xml são testadas com XMLWithPre.create / xml_with_pre +mockados -- não é objetivo destes testes validar o parsing real de XML, e +sim o contrato de XMLVersion ao redor dele. +""" +import shutil +import tempfile +from unittest.mock import MagicMock, PropertyMock, patch + +from django.contrib.auth import get_user_model +from django.db import IntegrityError +from django.test import TestCase, override_settings + +from pid_provider.models import ( + PidProviderXML, + XMLVersion, + XMLVersionGetError, + XMLVersionXmlWithPreError, +) + +User = get_user_model() + +MEDIA_ROOT = tempfile.mkdtemp() + + +def make_xml_with_pre(finger_print="fp-1", v3="ABCDEFGHIJKLMNOPQRSTUVW"): + xml_with_pre = MagicMock(name="xml_with_pre") + xml_with_pre.finger_print = finger_print + xml_with_pre.tostring.return_value = "
" + xml_with_pre.v3 = v3 + return xml_with_pre + + +@override_settings(MEDIA_ROOT=MEDIA_ROOT) +class XMLVersionMediaTestCase(TestCase): + """Base com limpeza do MEDIA_ROOT temporário usado pelos testes que gravam arquivo.""" + + @classmethod + def tearDownClass(cls): + shutil.rmtree(MEDIA_ROOT, ignore_errors=True) + super().tearDownClass() + + def setUp(self): + self.user = User.objects.create_user(username=f"user-{id(self)}", password="x") + # pkg_name é obrigatório para xml_directory_path (upload_to do FileField) + self.ppx = PidProviderXML.objects.create( + creator=self.user, v3="ABC", pkg_name="pkg-fake" + ) + + +class XMLVersionCreateTests(XMLVersionMediaTestCase): + def test_create_sets_finger_print_creator_and_saves_file(self): + xml_with_pre = make_xml_with_pre(finger_print="fp-abc") + + version = XMLVersion.create(self.user, self.ppx, xml_with_pre) + + self.assertEqual(version.finger_print, "fp-abc") + self.assertEqual(version.creator, self.user) + self.assertEqual(version.pid_provider_xml, self.ppx) + self.assertTrue(version.file.name.endswith(".xml")) + + def test_create_falls_back_to_get_on_integrity_error(self): + """ + IntegrityError ao salvar (ex.: corrida entre requisições concorrentes + criando a mesma versão) faz create() recuperar o registro já + existente via get(pid_provider_xml, finger_print), em vez de + propagar o erro. + """ + xml_with_pre = make_xml_with_pre(finger_print="fp-dup") + existing = XMLVersion.create(self.user, self.ppx, xml_with_pre) + + with patch.object(XMLVersion, "save", side_effect=IntegrityError): + result = XMLVersion.create(self.user, self.ppx, xml_with_pre) + + self.assertEqual(result.pk, existing.pk) + + +class XMLVersionIsEqualToTests(XMLVersionMediaTestCase): + def test_is_equal_to_true_when_finger_print_matches_and_file_exists(self): + version = XMLVersion.create(self.user, self.ppx, make_xml_with_pre(finger_print="same-fp")) + + self.assertTrue(version.is_equal_to(make_xml_with_pre(finger_print="same-fp"))) + + def test_is_equal_to_false_when_finger_print_differs(self): + version = XMLVersion.create(self.user, self.ppx, make_xml_with_pre(finger_print="fp-a")) + + self.assertFalse(version.is_equal_to(make_xml_with_pre(finger_print="fp-b"))) + + +class XMLVersionXmlWithPreTests(TestCase): + def test_raises_xml_version_error_when_file_has_no_path(self): + """ + Sem arquivo associado, self.file.path levanta ValueError -- capturado + pelo `except Exception` genérico da property e relançado como + XMLVersionXmlWithPreError (mensagem inclui o v3 do pid_provider_xml). + """ + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC-V3")) + + with self.assertRaises(XMLVersionXmlWithPreError) as ctx: + version.xml_with_pre + self.assertIn("ABC-V3", str(ctx.exception)) + + def test_raises_xml_version_error_when_xml_with_pre_create_fails(self): + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC")) + version.file.name = "some/path.xml" + + with patch("pid_provider.models.XMLWithPre.create", side_effect=Exception("boom")): + with self.assertRaises(XMLVersionXmlWithPreError): + version.xml_with_pre + + def test_returns_first_item_from_xml_with_pre_create(self): + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC")) + version.file.name = "some/path.xml" + item1, item2 = MagicMock(name="item1"), MagicMock(name="item2") + + with patch("pid_provider.models.XMLWithPre.create", return_value=[item1, item2]): + self.assertIs(version.xml_with_pre, item1) + + +class XMLVersionXmlCachedPropertyTests(TestCase): + """ + `xml` é uma cached_property que delega para a property `xml_with_pre` + (não cacheada), então cada teste usa uma instância nova para evitar + reaproveitar o cache entre casos. + """ + + def test_returns_string_on_success(self): + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC")) + fake_xml_with_pre = MagicMock() + fake_xml_with_pre.tostring.return_value = "
" + + with patch.object( + XMLVersion, "xml_with_pre", new_callable=PropertyMock + ) as mock_prop: + mock_prop.return_value = fake_xml_with_pre + self.assertEqual(version.xml, "
") + + def test_returns_error_string_when_xml_with_pre_raises(self): + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC")) + + with patch.object( + XMLVersion, "xml_with_pre", new_callable=PropertyMock + ) as mock_prop: + mock_prop.side_effect = XMLVersionXmlWithPreError("deu erro") + self.assertEqual(version.xml, "deu erro") + + def test_returns_none_when_file_not_found(self): + version = XMLVersion(pid_provider_xml=PidProviderXML(v3="ABC")) + + with patch.object( + XMLVersion, "xml_with_pre", new_callable=PropertyMock + ) as mock_prop: + mock_prop.side_effect = FileNotFoundError + self.assertIsNone(version.xml) + + +class XMLVersionGetTests(XMLVersionMediaTestCase): + def test_get_raises_when_missing_pid_provider_xml_or_finger_print(self): + with self.assertRaises(XMLVersionGetError): + XMLVersion.get(None, "fp") + with self.assertRaises(XMLVersionGetError): + XMLVersion.get(self.ppx, None) + + def test_get_returns_latest_matching_finger_print(self): + created = XMLVersion.create(self.user, self.ppx, make_xml_with_pre(finger_print="fp-x")) + + found = XMLVersion.get(self.ppx, "fp-x") + + self.assertEqual(found.pk, created.pk) + + def test_get_raises_does_not_exist_when_no_match(self): + with self.assertRaises(XMLVersion.DoesNotExist): + XMLVersion.get(self.ppx, "nao-existe") + + +class XMLVersionGetOrCreateTests(XMLVersionMediaTestCase): + def test_creates_new_when_none_exists(self): + xml_with_pre = make_xml_with_pre(finger_print="fp-new") + + version = XMLVersion.get_or_create(self.user, self.ppx, xml_with_pre) + + self.assertEqual(version.finger_print, "fp-new") + self.assertTrue(XMLVersion.objects.filter(pk=version.pk).exists()) + + def test_returns_latest_when_file_still_exists(self): + xml_with_pre = make_xml_with_pre(finger_print="fp-keep") + created = XMLVersion.create(self.user, self.ppx, xml_with_pre) + + with patch.object(XMLVersion, "save_file") as mock_save_file: + found = XMLVersion.get_or_create(self.user, self.ppx, xml_with_pre) + + self.assertEqual(found.pk, created.pk) + mock_save_file.assert_not_called() + + def test_resaves_file_when_latest_exists_but_file_missing(self): + xml_with_pre = make_xml_with_pre(finger_print="fp-missing-file") + created = XMLVersion.create(self.user, self.ppx, xml_with_pre) + # simula arquivo removido do storage sem remover o registro + created.file.delete(save=True) + + found = XMLVersion.get_or_create(self.user, self.ppx, xml_with_pre) + + self.assertEqual(found.pk, created.pk) + self.assertTrue(found.file.name)