[IMP] queue jobs visibility on records using the queue

spp_base_common/CHANGELOG.md

@@ -0,0 +1,21 @@
+# Changelog
+
+## 2025-11-20
+
+### 2025-11-20 16:30:00 - [FIX] queue job test creation with proper sentinel context
+
+- Fixed test job creation to use `_job_edit_sentinel` context for protected fields
+- Added `_create_test_job` helper method to simplify job creation in tests
+- Updated all test methods to use helper instead of direct create() calls
+- Tests now properly respect queue.job model's protected fields constraint
+
+### 2025-11-20 16:00:00 - [ADD] comprehensive tests for queue job tracking
+
+- Added `test_queue_job_tracking.py` with 10 comprehensive test cases
+- Tests verify `res_id` and `res_model` field functionality
+- Tests verify `_compute_job_ids` correctly filters jobs by record
+- Tests verify `_compute_has_ongoing_jobs` detects all job states (pending, enqueued, started)
+- Tests verify completed jobs (done/failed) don't trigger ongoing flag
+- Tests verify model isolation - jobs from different models don't interfere
+- Tests cover edge cases including empty results and mixed job states
+- Updated `__init__.py` to import new test module

spp_base_common/__manifest__.py

@@ -13,6 +13,7 @@
     "maintainers": ["jeremi", "gonzalesedwin1123", "emjay0921"],
     "depends": [
         "base",
+        "queue_job",
         "theme_openspp_muk",
         "spp_user_roles",
         "spp_area_base",
@@ -29,6 +30,8 @@
         "security/ir.model.access.csv",
         "views/main_view.xml",
         "views/phone_validation_view.xml",
+        "views/queue_job_views.xml",
+        "views/area_import_view.xml",
     ],
     "assets": {
         "web.assets_backend": [

spp_base_common/models/__init__.py

@@ -1,4 +1,7 @@
+from . import base
 from . import ir_module_module
 from . import phone_validation
 from . import phone_number
+from . import queue_job
 from . import res_partner
+from . import area_import

spp_base_common/models/area_import.py

@@ -0,0 +1,50 @@
+from odoo import api, fields, models
+
+
+class OpenSPPAreaImport(models.Model):
+    _inherit = "spp.area.import"
+
+    job_ids = fields.One2many(
+        "queue.job",
+        compute="_compute_job_ids",
+        string="Related Jobs",
+        help="Queue jobs related to this area import",
+    )
+
+    has_ongoing_jobs = fields.Boolean(
+        compute="_compute_has_ongoing_jobs",
+        string="Has Ongoing Jobs",
+        help="True if there are any ongoing queue jobs for this model",
+    )
+
+    def _compute_job_ids(self):
+        """
+        Compute related queue jobs based on res_id and res_model fields.
+        """
+        for rec in self:
+            jobs = self.env["queue.job"].search(
+                [
+                    ("res_model", "=", "spp.area.import"),
+                    ("res_id", "=", rec.id),
+                ]
+            )
+            rec.job_ids = jobs
+
+    @api.depends("job_ids", "job_ids.state")
+    def _compute_has_ongoing_jobs(self):
+        """
+        Check if there are any ongoing jobs for the spp.area.import model.
+        This checks across ALL area import records to prevent concurrent operations.
+        """
+        # Check for any ongoing jobs for the entire model
+        ongoing_jobs_count = self.env["queue.job"].search_count(
+            [
+                ("res_model", "=", "spp.area.import"),
+                ("state", "in", ["pending", "enqueued", "started"]),
+            ]
+        )
+        has_ongoing = ongoing_jobs_count > 0
+
+        # Set the same value for all records
+        for rec in self:
+            rec.has_ongoing_jobs = has_ongoing

spp_base_common/models/base.py

@@ -0,0 +1,37 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+from odoo import api, models
+
+
+class Base(models.AbstractModel):
+    """Extend base model to automatically populate res_id in queue jobs."""
+
+    _inherit = "base"
+
+    @api.model
+    def _job_store_values(self, job):
+        """
+        Override to automatically populate res_id and res_model in queue jobs.
+
+        This method is called when a job is being stored in the database.
+        It extracts the record ID from the recordset that created the job
+        and stores it in res_id field for later monitoring.
+
+        :param job: current queue_job.job.Job instance.
+        :return: dictionary for setting job values.
+        """
+        vals = super()._job_store_values(job)
+
+        # Get the recordset that is creating the job
+        recordset = job.recordset
+
+        # If the recordset has a single record, store its ID
+        if recordset and len(recordset) == 1:
+            vals["res_id"] = recordset.id
+            vals["res_model"] = recordset._name
+        # If the recordset has multiple records, store the first ID
+        elif recordset and len(recordset) > 1:
+            vals["res_id"] = recordset[0].id
+            vals["res_model"] = recordset._name
+
+        return vals

spp_base_common/models/queue_job.py

@@ -0,0 +1,20 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+from odoo import fields, models
+
+
+class QueueJob(models.Model):
+    """Extend queue.job to add res_id field for record tracking."""
+
+    _inherit = "queue.job"
+
+    res_id = fields.Integer(
+        string="Record ID",
+        help="ID of the record that created this job",
+        index=True,
+    )
+    res_model = fields.Char(
+        string="Record Model",
+        help="Model name of the record that created this job",
+        index=True,
+    )

spp_base_common/tests/__init__.py

@@ -1,2 +1,3 @@
 from . import test_ir_module_module
 from . import test_phone_number_validation
+from . import test_queue_job_tracking

spp_base_common/tests/test_queue_job_tracking.py

@@ -0,0 +1,186 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+import uuid
+
+from odoo.tests.common import TransactionCase
+
+
+class TestQueueJobTracking(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.queue_job_model = cls.env["queue.job"]
+        cls.area_import_model = cls.env["spp.area.import"]
+
+        # Get the EDIT_SENTINEL for protected fields
+        cls.job_edit_sentinel = cls.queue_job_model.EDIT_SENTINEL
+
+        # Create test area import records
+        cls.area_import_1 = cls.area_import_model.create(
+            {
+                "name": "Test Area Import 1",
+            }
+        )
+        cls.area_import_2 = cls.area_import_model.create(
+            {
+                "name": "Test Area Import 2",
+            }
+        )
+
+    def _create_test_job(self, name, res_model, res_id, state="done", method_name="test_method"):
+        """Helper method to create test queue jobs with proper sentinel context"""
+        return self.queue_job_model.with_context(_job_edit_sentinel=self.job_edit_sentinel).create(
+            {
+                "uuid": str(uuid.uuid4()),
+                "name": name,
+                "model_name": res_model,
+                "method_name": method_name,
+                "res_model": res_model,
+                "res_id": res_id,
+                "state": state,
+            }
+        )
+
+    def test_01_queue_job_res_fields(self):
+        """Test that queue.job model has res_id and res_model fields"""
+        job = self._create_test_job("Test Job", "spp.area.import", self.area_import_1.id, state="done")
+
+        self.assertEqual(job.res_model, "spp.area.import")
+        self.assertEqual(job.res_id, self.area_import_1.id)
+
+    def test_02_compute_job_ids(self):
+        """Test that area import correctly computes related jobs"""
+        # Create jobs for area_import_1
+        job1 = self._create_test_job("Job 1 for Import 1", "spp.area.import", self.area_import_1.id, state="done")
+        job2 = self._create_test_job("Job 2 for Import 1", "spp.area.import", self.area_import_1.id, state="pending")
+
+        # Create job for area_import_2
+        job3 = self._create_test_job("Job 1 for Import 2", "spp.area.import", self.area_import_2.id, state="done")
+
+        # Trigger compute
+        self.area_import_1._compute_job_ids()
+        self.area_import_2._compute_job_ids()
+
+        # Assert area_import_1 has 2 jobs
+        self.assertEqual(len(self.area_import_1.job_ids), 2)
+        self.assertIn(job1, self.area_import_1.job_ids)
+        self.assertIn(job2, self.area_import_1.job_ids)
+        self.assertNotIn(job3, self.area_import_1.job_ids)
+
+        # Assert area_import_2 has 1 job
+        self.assertEqual(len(self.area_import_2.job_ids), 1)
+        self.assertIn(job3, self.area_import_2.job_ids)
+        self.assertNotIn(job1, self.area_import_2.job_ids)
+        self.assertNotIn(job2, self.area_import_2.job_ids)
+
+    def test_03_has_ongoing_jobs_no_jobs(self):
+        """Test has_ongoing_jobs when there are no jobs"""
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        self.assertFalse(self.area_import_1.has_ongoing_jobs)
+        self.assertFalse(self.area_import_2.has_ongoing_jobs)
+
+    def test_04_has_ongoing_jobs_with_pending_job(self):
+        """Test has_ongoing_jobs when there is a pending job"""
+        # Create a pending job
+        self._create_test_job("Pending Job", "spp.area.import", self.area_import_1.id, state="pending")
+
+        # Trigger compute on both records
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        # Both should show has_ongoing_jobs = True because it checks the entire model
+        self.assertTrue(self.area_import_1.has_ongoing_jobs)
+        self.assertTrue(self.area_import_2.has_ongoing_jobs)
+
+    def test_05_has_ongoing_jobs_with_enqueued_job(self):
+        """Test has_ongoing_jobs when there is an enqueued job"""
+        # Clean up previous jobs
+        self.queue_job_model.search([("res_model", "=", "spp.area.import")]).unlink()
+
+        # Create an enqueued job
+        self._create_test_job("Enqueued Job", "spp.area.import", self.area_import_2.id, state="enqueued")
+
+        # Trigger compute
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        # Both should show has_ongoing_jobs = True
+        self.assertTrue(self.area_import_1.has_ongoing_jobs)
+        self.assertTrue(self.area_import_2.has_ongoing_jobs)
+
+    def test_06_has_ongoing_jobs_with_started_job(self):
+        """Test has_ongoing_jobs when there is a started job"""
+        # Clean up previous jobs
+        self.queue_job_model.search([("res_model", "=", "spp.area.import")]).unlink()
+
+        # Create a started job
+        self._create_test_job("Started Job", "spp.area.import", self.area_import_1.id, state="started")
+
+        # Trigger compute
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        # Both should show has_ongoing_jobs = True
+        self.assertTrue(self.area_import_1.has_ongoing_jobs)
+        self.assertTrue(self.area_import_2.has_ongoing_jobs)
+
+    def test_07_has_ongoing_jobs_with_done_job(self):
+        """Test has_ongoing_jobs when all jobs are done"""
+        # Clean up previous jobs
+        self.queue_job_model.search([("res_model", "=", "spp.area.import")]).unlink()
+
+        # Create only done/failed jobs
+        self._create_test_job("Done Job", "spp.area.import", self.area_import_1.id, state="done")
+        self._create_test_job("Failed Job", "spp.area.import", self.area_import_2.id, state="failed")
+
+        # Trigger compute
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        # Both should show has_ongoing_jobs = False
+        self.assertFalse(self.area_import_1.has_ongoing_jobs)
+        self.assertFalse(self.area_import_2.has_ongoing_jobs)
+
+    def test_08_has_ongoing_jobs_mixed_states(self):
+        """Test has_ongoing_jobs with mixed job states"""
+        # Clean up previous jobs
+        self.queue_job_model.search([("res_model", "=", "spp.area.import")]).unlink()
+
+        # Create jobs with different states
+        self._create_test_job("Done Job", "spp.area.import", self.area_import_1.id, state="done")
+        self._create_test_job("Pending Job", "spp.area.import", self.area_import_2.id, state="pending")
+
+        # Trigger compute
+        self.area_import_1._compute_has_ongoing_jobs()
+        self.area_import_2._compute_has_ongoing_jobs()
+
+        # Both should show has_ongoing_jobs = True because there's one pending job
+        self.assertTrue(self.area_import_1.has_ongoing_jobs)
+        self.assertTrue(self.area_import_2.has_ongoing_jobs)
+
+    def test_09_job_ids_empty_for_different_model(self):
+        """Test that job_ids is empty when jobs belong to different model"""
+        # Create a job with different res_model
+        self._create_test_job("Job for different model", "res.partner", 1, state="done")
+
+        # Trigger compute
+        self.area_import_1._compute_job_ids()
+
+        # Should not include jobs from other models
+        self.assertEqual(len(self.area_import_1.job_ids), 0)
+
+    def test_10_has_ongoing_jobs_different_model(self):
+        """Test that has_ongoing_jobs is not affected by jobs from different models"""
+        # Clean up previous jobs
+        self.queue_job_model.search([("res_model", "=", "spp.area.import")]).unlink()
+
+        # Create a pending job for a different model
+        self._create_test_job("Job for different model", "res.partner", 1, state="pending")
+
+        # Trigger compute
+        self.area_import_1._compute_has_ongoing_jobs()
+
+        # Should not be affected by jobs from other models
+        self.assertFalse(self.area_import_1.has_ongoing_jobs)