Valida arquivo antes de gravar no cadastro de acessórios em lote

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

sapl/materia/tests/test_acessorio_em_lote.py

@@ -0,0 +1,64 @@
+import os
+
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.urls import reverse
+from model_bakery import baker
+import pytest
+
+from sapl.materia.models import (DocumentoAcessorio, MateriaLegislativa,
+                                 TipoDocumento)
+
+
+@pytest.fixture
+def dirs(settings, tmp_path, monkeypatch):
+    media = tmp_path / 'media'
+    temp = tmp_path / 'temp'
+    media.mkdir()
+    temp.mkdir()
+    settings.MEDIA_ROOT = str(media)
+    monkeypatch.setattr('sapl.materia.views.MEDIA_ROOT', str(media))
+    monkeypatch.setattr('sapl.materia.views.get_tempfile_dir',
+                        lambda: str(temp))
+    return media, temp
+
+
+def _post(client, nome, conteudo):
+    materias = baker.make(MateriaLegislativa, _quantity=2)
+    tipo = baker.make(TipoDocumento, descricao='Parecer')
+    client.post(reverse('sapl.materia:acessorio_em_lote'), {
+        'materia_id': [m.pk for m in materias],
+        'tipo': tipo.descricao,
+        'nome': 'Parecer',
+        'autor': 'Comissão',
+        'data': '12/10/2026',
+        'ementa': 'Ementa',
+        'arquivo': SimpleUploadedFile(nome, conteudo),
+    })
+    return materias
+
+
+@pytest.mark.django_db(transaction=False)
+def test_arquivo_invalido_nao_grava_nada(admin_client, dirs):
+    media, temp = dirs
+
+    _post(admin_client, 'x.html', b'%PDF-1.4\n<html><script></script>')
+
+    assert not DocumentoAcessorio.objects.exists()
+    assert os.listdir(media) == []
+    assert os.listdir(temp) == []
+
+
+@pytest.mark.django_db(transaction=False)
+def test_arquivo_valido_copiado_para_cada_materia(admin_client, dirs):
+    media, temp = dirs
+
+    materias = _post(admin_client, 'parecer.pdf', b'%PDF-1.4\n%%EOF')
+
+    docs = DocumentoAcessorio.objects.filter(materia__in=materias)
+    assert docs.count() == 2
+    for doc in docs:
+        assert doc.arquivo.name == (
+            'sapl/public/documentoacessorio/2026/%s/parecer.pdf' % doc.pk)
+        assert (media / doc.arquivo.name).read_bytes() == b'%PDF-1.4\n%%EOF'
+    assert os.listdir(media) == ['sapl']
+    assert os.listdir(temp) == []

sapl/materia/views.py

@@ -5,6 +5,7 @@ import logging
 import os
 from random import choice
 import shutil
+import tempfile
 from string import ascii_letters, digits
 import time
 import zipfile
@@ -52,11 +53,12 @@ from sapl.materia.forms import (AnexadaForm, AutoriaForm, AutoriaMultiCreateForm
 from sapl.norma.models import LegislacaoCitada
 from sapl.parlamentares.models import Legislatura
 from sapl.protocoloadm.models import Protocolo
-from sapl.settings import MAX_DOC_UPLOAD_SIZE, MEDIA_ROOT, RATE_LIMITER_RATE
+from sapl.settings import MEDIA_ROOT, RATE_LIMITER_RATE
 from sapl.utils import (autor_label, autor_modal, gerar_hash_arquivo, get_base_url,
                         get_client_ip, get_mime_type_from_file_extension, lista_anexados,
                         mail_service_configured, montar_row_autor, SEPARADOR_HASH_PROPOSICAO,
                         show_results_filter_set, get_tempfile_dir,
+                        restringe_tipos_de_arquivo_txt, validar_arquivo,
                         google_recaptcha_configured, MultiFormatOutputMixin, ratelimit_ip)
 
 from .forms import (AcessorioEmLoteFilterSet, AcompanhamentoMateriaForm,
@@ -2330,17 +2332,15 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
             messages.add_message(request, messages.ERROR, msg)
             return self.get(request, self.kwargs)
 
-        if request.FILES['arquivo'].size > MAX_DOC_UPLOAD_SIZE:
+        arquivo = request.FILES['arquivo']
-            msg = _("O arquivo Anexo de Texto Integral deve ser menor que {0:.1f} MB, \
+        try:
-                o tamanho atual desse arquivo é {1:.1f} MB"
+            validar_arquivo(arquivo, 'Arquivo')
-                    .format((MAX_DOC_UPLOAD_SIZE / 1024) / 1024, (request.FILES['arquivo'].size / 1024) / 1024))
+            restringe_tipos_de_arquivo_txt(arquivo)
-            messages.add_message(request, messages.ERROR, msg)
+        except ValidationError as e:
+            for m in e.messages:
+                messages.add_message(request, messages.ERROR, m)
             return self.get(request, self.kwargs)
 
-        tmp_name = os.path.join(MEDIA_ROOT, request.FILES['arquivo'].name)
-        with open(tmp_name, 'wb') as destination:
-            for chunk in request.FILES['arquivo'].chunks():
-                destination.write(chunk)
         try:
             doc_data = tz.localize(datetime.strptime(
                 request.POST['data'], "%d/%m/%Y"))
@@ -2350,44 +2350,47 @@ class DocumentoAcessorioEmLoteView(PermissionRequiredMixin, FilterView):
             messages.add_message(request, messages.ERROR, msg)
             self.logger.error("User={}. {}. Data inserida: {}".format(
                 username, str(msg), request.POST['data']))
-            os.remove(tmp_name)
             return self.get(request, self.kwargs)
 
-        for materia_id in marcadas:
+        # Fora de MEDIA_ROOT, para não ficar acessível em /media/
-            doc = DocumentoAcessorio()
+        with tempfile.NamedTemporaryFile(dir=get_tempfile_dir(),
-            doc.materia_id = materia_id
+                                         delete=False) as tmp:
-            doc.tipo = tipo
+            for chunk in arquivo.chunks():
-            doc.nome = request.POST['nome']
+                tmp.write(chunk)
-            doc.data = doc_data
+        try:
-            doc.autor = request.POST['autor']
+            for materia_id in marcadas:
-            doc.ementa = request.POST['ementa']
+                doc = DocumentoAcessorio()
-            doc.arquivo.name = tmp_name
+                doc.materia_id = materia_id
-            try:
+                doc.tipo = tipo
-                doc.clean_fields()
+                doc.nome = request.POST['nome']
-            except ValidationError as e:
+                doc.data = doc_data
-                for m in ['%s: %s' % (DocumentoAcessorio()._meta.get_field(k).verbose_name, '</br>'.join(v))
+                doc.autor = request.POST['autor']
-                          for k, v in e.message_dict.items()]:
+                doc.ementa = request.POST['ementa']
-                    # Insere as mensagens de erro no formato:
+                try:
-                    # 'verbose_name do nome do campo': 'mensagem de erro'
+                    doc.clean_fields(exclude=['arquivo'])
-                    messages.add_message(request, messages.ERROR, m)
+                except ValidationError as e:
-                    self.logger.error("User={}. {}. Nome do arquivo: {}.".format(
+                    for m in ['%s: %s' % (DocumentoAcessorio()._meta.get_field(k).verbose_name, '</br>'.join(v))
-                        username, str(msg), request.FILES['arquivo'].name))
+                              for k, v in e.message_dict.items()]:
-                os.remove(tmp_name)
+                        # Insere as mensagens de erro no formato:
-                return self.get(request, self.kwargs)
+                        # 'verbose_name do nome do campo': 'mensagem de erro'
-            doc.save()
+                        messages.add_message(request, messages.ERROR, m)
-            diretorio = os.path.join(MEDIA_ROOT,
+                        self.logger.error("User={}. {}. Nome do arquivo: {}.".format(
-                                     'sapl/public/documentoacessorio',
+                            username, str(m), arquivo.name))
-                                     str(doc_data.year),
+                    return self.get(request, self.kwargs)
-                                     str(doc.id))
+                doc.save()
-            if not os.path.exists(diretorio):
+                diretorio = os.path.join(MEDIA_ROOT,
-                os.makedirs(diretorio)
+                                         'sapl/public/documentoacessorio',
-            file_path = os.path.join(diretorio,
+                                         str(doc_data.year),
-                                     request.FILES['arquivo'].name)
+                                         str(doc.id))
-            shutil.copy2(tmp_name, file_path)
+                if not os.path.exists(diretorio):
-            doc.arquivo.name = file_path.split(
+                    os.makedirs(diretorio)
-                MEDIA_ROOT + "/")[1]  # Retira MEDIA_ROOT do nome
+                file_path = os.path.join(diretorio, arquivo.name)
-            doc.save()
+                shutil.copy2(tmp.name, file_path)
-        os.remove(tmp_name)
+                doc.arquivo.name = file_path.split(
+                    MEDIA_ROOT + "/")[1]  # Retira MEDIA_ROOT do nome
+                doc.save()
+        finally:
+            os.remove(tmp.name)
 
         msg = _('Documento(s) criado(s).')
         messages.add_message(request, messages.SUCCESS, msg)