[REF] hr_payroll_document: In tests, extract common setup

hr_payroll_document/__manifest__.py

@@ -8,7 +8,11 @@
     "version": "18.0.1.0.0",
     "depends": ["hr", "base_vat"],
     "maintainers": ["peluko00"],
-    "external_dependencies": {"python": ["pypdf"]},
+    "external_dependencies": {
+        "python": [
+            "pypdf",
+        ],
+    },
     "data": [
         "wizard/payroll_management_wizard.xml",
         "security/ir.model.access.csv",

hr_payroll_document/models/hr_employee.py

@@ -1,4 +1,7 @@
-from odoo import fields, models
+# Copyright 2025 Simone Rubino - PyTech
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+
+from odoo import api, fields, models
 from odoo.exceptions import ValidationError
 
 
@@ -44,12 +47,21 @@ def action_get_payroll_tree_view(self):
         )
         return action
 
-    def write(self, vals):
-        res = super().write(vals)
-        if "identification_id" in vals and not self.env["res.partner"].simple_vat_check(
-            self.env.company.country_id.code, vals["identification_id"]
-        ):
-            raise ValidationError(
-                self.env._("The field identification ID is not valid")
-            )
-        return res
+    def _validate_payroll_identification(self, code=None):
+        # Override if the identification should be validated in another way
+        if code is None and len(self) == 1:
+            code = self.identification_id
+        if country_code := self.env.company.country_id.code:
+            is_valid = self.env["res.partner"].simple_vat_check(country_code, code)
+        else:
+            is_valid = True
+        return is_valid
+
+    @api.constrains("identification_id")
+    def _constrain_payroll_identification(self):
+        # Only check the employees that have an `identification_id`
+        for employee in self.filtered("identification_id"):
+            if not employee._validate_payroll_identification():
+                raise ValidationError(
+                    employee.env._("The field identification ID is not valid")
+                )

hr_payroll_document/tests/common.py

@@ -0,0 +1,75 @@
+import base64
+import contextlib
+from unittest import mock
+
+from odoo.tests import common
+from odoo.tools.misc import file_path as open_file_path
+
+from odoo.addons.mail.tests.common import mail_new_test_user
+
+
+class TestHrPayrollDocument(common.TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.env.user.tz = "Europe/Brussels"
+        cls.user_admin = cls.env.ref("base.user_admin")
+
+        # Fix Company without country
+        cls.env.company.country_id = False
+
+        # Test users to use through the various tests
+        cls.user_employee = mail_new_test_user(
+            cls.env, login="david", groups="base.group_user"
+        )
+        cls.user_employee_id = cls.user_employee.id
+
+        # Hr Data
+        cls.employee_emp = cls.env["hr.employee"].create(
+            {
+                "name": "David Employee",
+                "user_id": cls.user_employee_id,
+                "company_id": 1,
+                "identification_id": "30831011V",
+            }
+        )
+
+        cls.wizard = cls._create_wizard("January", "hr_payroll_document/tests/test.pdf")
+
+    @classmethod
+    def _create_wizard(cls, subject, file_path):
+        with open(open_file_path(file_path), "rb") as pdf_file:
+            encoded_string = base64.b64encode(pdf_file.read())
+        ir_values = {
+            "name": "test",
+            "type": "binary",
+            "datas": encoded_string,
+            "store_fname": encoded_string,
+            "res_model": "payroll.management.wizard",
+            "res_id": 1,
+        }
+        cls.attachment = cls.env["ir.attachment"].create(ir_values)
+        cls.subject = subject
+        return cls.env["payroll.management.wizard"].create(
+            {"payrolls": [cls.attachment.id], "subject": cls.subject}
+        )
+
+    @contextlib.contextmanager
+    def _mock_valid_identification(self, employee, identification_code):
+        def _mocked_validate_payroll_identification(self, code=None):
+            if code is None:
+                code = employee.identification_id
+            return code == identification_code
+
+        with mock.patch.object(
+            type(employee),
+            "_validate_payroll_identification",
+            _mocked_validate_payroll_identification,
+        ) as patch:
+            patch.side_effect = _mocked_validate_payroll_identification
+            yield
+
+    def fill_company_id(self):
+        self.env.company.country_id = self.env["res.country"].search(
+            [("name", "=", "Spain")]
+        )

hr_payroll_document/tests/test_hr_payroll_document.py

@@ -4,74 +4,17 @@
 import pypdf
 
 from odoo.exceptions import UserError, ValidationError
-from odoo.tests import common
-from odoo.tools.misc import file_path
 
-from odoo.addons.mail.tests.common import mail_new_test_user
+from odoo.addons.hr_payroll_document.tests.common import TestHrPayrollDocument
 
 
-class TestHRPayrollDocument(common.TransactionCase):
-    @classmethod
-    def setUpClass(cls):
-        super().setUpClass()
-        cls.env.user.tz = "Europe/Brussels"
-        cls.user_admin = cls.env.ref("base.user_admin")
-
-        # Fix Company without country
-        cls.env.company.country_id = False
-
-        # Test users to use through the various tests
-        cls.user_employee = mail_new_test_user(
-            cls.env, login="david", groups="base.group_user"
-        )
-        cls.user_employee_id = cls.user_employee.id
-
-        # Hr Data
-        cls.employee_emp = cls.env["hr.employee"].create(
-            {
-                "name": "David Employee",
-                "user_id": cls.user_employee_id,
-                "company_id": 1,
-                "identification_id": "30831011V",
-            }
-        )
-
-        with open(file_path("hr_payroll_document/tests/test.pdf"), "rb") as pdf_file:
-            encoded_string = base64.b64encode(pdf_file.read())
-        ir_values = {
-            "name": "test",
-            "type": "binary",
-            "datas": encoded_string,
-            "store_fname": encoded_string,
-            "res_model": "payroll.management.wizard",
-            "res_id": 1,
-        }
-        cls.attachment = cls.env["ir.attachment"].create(ir_values)
-        cls.subject = "January"
-        cls.wizard = cls.env["payroll.management.wizard"].create(
-            {"payrolls": [cls.attachment.id], "subject": cls.subject}
-        )
-
-    def fill_company_id(self):
-        self.env.company.country_id = self.env["res.country"].search(
-            [("name", "=", "Spain")]
-        )
+class TestHRPayrollDocument(TestHrPayrollDocument):
+    def setUp(self, *args, **kwargs):
+        super().setUp(*args, **kwargs)
 
     def test_extension_error(self):
-        with open(file_path("hr_payroll_document/tests/test.docx"), "rb") as pdf_file:
-            encoded_string = base64.b64encode(pdf_file.read())
-        ir_values = {
-            "name": "test",
-            "type": "binary",
-            "datas": encoded_string,
-            "store_fname": encoded_string,
-            "res_model": "payroll.management.wizard",
-            "res_id": 1,
-        }
-        self.attachment = self.env["ir.attachment"].create(ir_values)
-        self.subject = "January"
-        self.wizard = self.env["payroll.management.wizard"].create(
-            {"payrolls": [self.attachment.id], "subject": self.subject}
+        self.wizard = self._create_wizard(
+            "January", "hr_payroll_document/tests/test.docx"
         )
         with self.assertRaises(ValidationError):
             self.wizard.send_payrolls()

hr_payroll_document/wizard/payroll_management_wizard.py

@@ -1,7 +1,7 @@
 import base64
 from base64 import b64decode
 
-from pypdf import PdfReader, PdfWriter
+from pypdf import PdfReader, PdfWriter, errors
 
 from odoo import fields, models
 from odoo.exceptions import UserError, ValidationError
@@ -19,55 +19,110 @@ class PayrollManagamentWizard(models.TransientModel):
         "ir.attachment", "payrol_rel", "doc_id", "attach_id3", copy=False, required=True
     )
 
-    def send_payrolls(self):
-        not_found = set()
-        self.merge_pdfs()
-        reader = PdfReader("/tmp/merged-pdf.pdf")
-        employees = set()
-
-        # Validate if company have country
-        if not self.env.company.country_id:
-            raise UserError(self.env._("You must to filled country field of company"))
+    def _get_fallback_reader(self, pdf_reader):
+        # Override to use another reader
+        pass
+
+    def _read_page_content(self, pdf_reader, page, fallback_reader=None):
+        try:
+            page_content = page.extract_text().split()
+        except errors.PdfReadError:
+            if fallback_reader:
+                # The original page cannot be read:
+                # read the simplified page in the fallback_reader
+                page_number = pdf_reader.get_page_number(page)
+                fallback_page = fallback_reader.get_page(page_number)
+                page_content = fallback_page.extract_text().split()
+            else:
+                raise
+        return page_content
+
+    def _extract_employees(self, pdf_reader, fallback_reader=None):
+        employee_to_pages = dict()
+        not_found_ids = set()
 
         # Find all IDs of the employees
-        for page in reader.pages:
-            for value in page.extract_text().split():
+        for page in pdf_reader.pages:
+            page_content = self._read_page_content(
+                pdf_reader, page, fallback_reader=fallback_reader
+            )
+            for value in page_content:
                 if self.validate_id(value) and value != self.env.company.vat:
                     employee = self.env["hr.employee"].search(
                         [("identification_id", "=", value)]
                     )
                     if employee:
-                        employees.add(employee)
+                        employee_to_pages.setdefault(employee, []).append(page)
                     else:
-                        not_found.add(value)
+                        not_found_ids.add(value)
+                    break
 
-        for employee in list(employees):
-            pdfWriter = PdfWriter()
-            for page in reader.pages:
-                if employee.identification_id in page.extract_text():
-                    # Save pdf with payrolls of employee
-                    pdfWriter.add_page(page)
+        return employee_to_pages, not_found_ids
 
-            path = "/tmp/" + self.env._("Payroll ") + employee.name + ".pdf"
+    def _build_employee_payroll(self, file_name, pdf_pages, encryption_key=None):
+        """Return the path to the created payroll.
 
-            if not employee.no_payroll_encryption:
-                # Encrypt the payroll file
-                # with the identification identifier of the employee
-                pdfWriter.encrypt(employee.identification_id, algorithm="AES-256")
+        Optionally encrypt the payroll file with `encryption_key`.
+        """
+        pdfWriter = PdfWriter()
+        for page in pdf_pages:
+            pdfWriter.add_page(page)
 
-            f = open(path, "wb")
-            pdfWriter.write(f)
-            f.close()
+        path = "/tmp/" + file_name
 
-            # Send payroll to the employee
-            self.send_mail(employee, path)
+        if encryption_key:
+            pdfWriter.encrypt(encryption_key, algorithm="AES-256")
+
+        with open(path, "wb") as f:
+            pdfWriter.write(f)
+        return path
 
+    def _show_employees_action(self):
         action = self.env["ir.actions.actions"]._for_xml_id(
             "hr_payroll_document.payrolls_view_action"
         )
         action["views"] = [
             [self.env.ref("hr_payroll_document.view_payroll_tree").id, "list"]
         ]
+        return action
+
+    def send_payrolls(self):
+        self.merge_pdfs()
+        # Validate if company have country
+        if not self.env.company.country_id:
+            raise UserError(self.env._("You must to filled country field of company"))
+
+        reader = PdfReader("/tmp/merged-pdf.pdf")
+
+        try:
+            employee_to_pages, not_found = self._extract_employees(reader)
+        except errors.PdfReadError:
+            # Couldn't read the file, try again with another reader
+            fallback_reader = self._get_fallback_reader(reader)
+            if fallback_reader:
+                employee_to_pages, not_found = self._extract_employees(
+                    reader, fallback_reader=fallback_reader
+                )
+            else:
+                raise
+
+        for employee, pages in employee_to_pages.items():
+            encryption_key = (
+                None if employee.no_payroll_encryption else employee.identification_id
+            )
+            path = self._build_employee_payroll(
+                self.env._(
+                    "Payroll %(subject)s %(employee)s.pdf",
+                    employee=employee.name,
+                    subject=self.subject,
+                ),
+                pages,
+                encryption_key=encryption_key,
+            )
+            # Send payroll to the employee
+            self.send_mail(employee, path)
+
+        action = self._show_employees_action()
         if not_found:
             return {
                 "type": "ir.actions.client",
@@ -156,7 +211,5 @@ def send_mail(self, employee, path):
             employee.id, force_send=True
         )
 
-    def validate_id(self, number):
-        return self.env["res.partner"].simple_vat_check(
-            self.env.company.country_id.code, number
-        )
+    def validate_id(self, code):
+        return self.env["hr.employee"]._validate_payroll_identification(code=code)