[17.0] [IMP] fieldservice[_*]: populate person_id and warehouse_id from the related territory (if set)

fieldservice/models/fsm_location.py

@@ -69,6 +69,11 @@ class FSMLocation(models.Model):
     complete_name = fields.Char(
         compute="_compute_complete_name", recursive=True, store=True
     )
+    complete_direction = fields.Char(
+        compute="_compute_complete_direction",
+        store=True,
+        recursive=True,
+    )
 
     @api.model_create_multi
     def create(self, vals):
@@ -94,6 +99,13 @@ def _compute_complete_name(self):
                 else:
                     loc.complete_name = loc.partner_id.name
 
+    @api.depends("direction", "fsm_parent_id.complete_direction")
+    def _compute_complete_direction(self):
+        for rec in self:
+            parent_direction = rec.fsm_parent_id.complete_direction
+            complete_direction = (parent_direction or "") + (rec.direction or "")
+            rec.complete_direction = complete_direction or False
+
     @api.onchange("fsm_parent_id")
     def _onchange_fsm_parent_id(self):
         self.owner_id = self.fsm_parent_id.owner_id or False

fieldservice/models/fsm_order.py

@@ -1,9 +1,10 @@
 # Copyright (C) 2018 Open Source Integrators
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
+import warnings
 from datetime import datetime, timedelta
 
-from odoo import _, api, fields, models
+from odoo import Command, _, api, fields, models
 from odoo.exceptions import UserError, ValidationError
 
 from . import fsm_stage
@@ -112,7 +113,12 @@
     location_id = fields.Many2one(
         "fsm.location", string="Location", index=True, required=True
     )
-    location_directions = fields.Char()
+    location_directions = fields.Char(
+        compute="_compute_location_directions",
+        precompute=True,
+        store=True,
+        readonly=False,
+    )
     request_early = fields.Datetime(
         string="Earliest Request Date", default=datetime.now()
     )
@@ -151,32 +157,37 @@
         return vals
 
     request_late = fields.Datetime(string="Latest Request Date")
-    description = fields.Text()
+    description = fields.Text(
+        compute="_compute_description",
+        precompute=True,
+        store=True,
+        readonly=False,
+    )
 
     person_ids = fields.Many2many("fsm.person", string="Field Service Workers")
 
-    @api.onchange("location_id")
-    def _onchange_location_id_customer(self):
-        if self.location_id:
-            self.territory_id = self.location_id.territory_id or False
-            self.branch_id = self.location_id.branch_id or False
-            self.district_id = self.location_id.district_id or False
-            self.region_id = self.location_id.region_id or False
-            self.copy_notes()
-        if self.company_id.auto_populate_equipments_on_order:
-            fsm_equipment_rec = self.env["fsm.equipment"].search(
-                [("current_location_id", "=", self.location_id.id)]
-            )
-            self.equipment_ids = [(6, 0, fsm_equipment_rec.ids)]
-
     # Planning
-    person_id = fields.Many2one("fsm.person", string="Assigned To", index=True)
+    person_id = fields.Many2one(
+        "fsm.person",
+        string="Assigned To",
+        compute="_compute_person_id",
+        precompute=True,
+        store=True,
+        readonly=False,
+        index=True,
+    )
     person_phone = fields.Char(related="person_id.phone", string="Worker Phone")
     scheduled_date_start = fields.Datetime(string="Scheduled Start (ETA)")
     scheduled_duration = fields.Float(help="Scheduled duration of the work in" " hours")
     scheduled_date_end = fields.Datetime(string="Scheduled End")
     sequence = fields.Integer(default=10)
-    todo = fields.Text(string="Instructions")
+    todo = fields.Text(
+        string="Instructions",
+        compute="_compute_todo",
+        precompute=True,
+        store=True,
+        readonly=False,
+    )
 
     # Execution
     resolution = fields.Text()
@@ -191,19 +202,24 @@
 
     # Location
     territory_id = fields.Many2one(
-        "res.territory",
-        string="Territory",
         related="location_id.territory_id",
+        precompute=True,
         store=True,
     )
     branch_id = fields.Many2one(
-        "res.branch", string="Branch", related="location_id.branch_id", store=True
+        related="location_id.branch_id",
+        precompute=True,
+        store=True,
     )
     district_id = fields.Many2one(
-        "res.district", string="District", related="location_id.district_id", store=True
+        related="location_id.district_id",
+        precompute=True,
+        store=True,
     )
     region_id = fields.Many2one(
-        "res.region", string="Region", related="location_id.region_id", store=True
+        related="location_id.region_id",
+        precompute=True,
+        store=True,
     )
 
     # Fields for Geoengine Identify
@@ -229,11 +245,77 @@
     equipment_id = fields.Many2one("fsm.equipment", string="Equipment")
 
     # Equipment used for all other Service Orders
-    equipment_ids = fields.Many2many("fsm.equipment", string="Equipments")
+    equipment_ids = fields.Many2many(
+        "fsm.equipment",
+        string="Equipments",
+        compute="_compute_equipment_ids",
+        precompute=True,
+        store=True,
+        readonly=False,
+    )
     type = fields.Many2one("fsm.order.type")
 
     internal_type = fields.Selection(related="type.internal_type")
 
+    @api.depends("company_id")
+    def _compute_equipment_ids(self):
+        for rec in self:
+            # Clear equipments that no longer match the order company
+            to_remove = rec.equipment_ids.filtered(
+                lambda equipment, rec=rec: equipment.company_id != rec.company_id
+            )
+            if to_remove:
+                rec.equipment_ids = [
+                    Command.unlink(equipment.id) for equipment in to_remove
+                ]
+            # If we have no equipments, auto populate if needed
+            if (
+                rec.company_id.auto_populate_equipments_on_order
+                and not rec.equipment_ids
+            ):
+                rec.equipment_ids = self.env["fsm.equipment"].search(
+                    [
+                        ("current_location_id", "=", rec.location_id.id),
+                        ("company_id", "=", rec.company_id.id),
+                    ]
+                )
+
+    @api.depends("location_id")
+    def _compute_location_directions(self):
+        for rec in self:
+            rec.location_directions = rec.location_id.complete_direction
+
+    @api.depends("template_id")
+    def _compute_todo(self):
+        for rec in self:
+            if rec.template_id:
+                rec.todo = rec.template_id.instructions
+
+    @api.depends("equipment_ids", "equipment_id", "type")
+    def _compute_description(self):
+        for rec in self:
+            if rec.description:
+                continue
+            equipments = (
+                rec.equipment_ids
+                if rec.type and rec.internal_type not in ("repair", "maintenance")
+                else rec.equipment_id
+            )
+            rec.description = "\n".join(
+                equipment.notes for equipment in equipments if equipment.notes
+            )
+
+    @api.depends("territory_id")
+    def _compute_person_id(self):
+        """Compute the person from the territory"""
+        for rec in self:
+            # If the person is one of the territory's workers, keep it.
+            if rec.person_id in rec.territory_id.person_ids:
+                continue
+            # If the territory has a primary assignment, use it.
+            if rec.territory_id.person_id:
+                rec.person_id = rec.territory_id.person_id
+
     @api.model
     def _read_group_stage_ids(self, stages, domain, order):
         search_domain = [("stage_type", "=", "order")]
@@ -358,51 +440,25 @@
         else:
             self.scheduled_date_end = self.scheduled_date_start
 
-    def copy_notes(self):
-        old_desc = self.description
-        self.location_directions = ""
-        if self.type and self.type.name not in ["repair", "maintenance"]:
-            for equipment_id in self.equipment_ids.filtered(lambda eq: eq.notes):
-                desc = self.description or ""
-                self.description = desc + equipment_id.notes + "\n "
-        else:
-            if self.equipment_id.notes:
-                desc = self.description if self.description else ""
-                self.description = desc + self.equipment_id.notes + "\n "
-        if self.location_id:
-            self.location_directions = self._get_location_directions(self.location_id)
-        if self.template_id:
-            self.todo = self.template_id.instructions
-        if old_desc:
-            self.description = old_desc
-
-    @api.onchange("equipment_ids")
-    def onchange_equipment_ids(self):
-        self.copy_notes()
-
     @api.onchange("template_id")
     def _onchange_template_id(self):
         if self.template_id:
             self.category_ids = self.template_id.category_ids
             self.scheduled_duration = self.template_id.duration
-            self.copy_notes()
             if self.template_id.type_id:
                 self.type = self.template_id.type_id
             if self.template_id.team_id:
                 self.team_id = self.template_id.team_id
 
-    def _get_location_directions(self, location_id):
-        self.location_directions = ""
-        s = self.location_id.direction or ""
-        parent_location = self.location_id.fsm_parent_id
-        # ps => Parent Location Directions
-        # s => String to Return
-        while parent_location.id is not False:
-            ps = parent_location.direction
-            if ps:
-                s += parent_location.direction
-            parent_location = parent_location.fsm_parent_id
-        return s
+    def _get_location_directions(self, location_id):  # pragma: no cover
+        # TODO(migration): Remove this method
+        warnings.warn(
+            "Deprecated fsm.order._get_location_directions(), "
+            "use location.complete_direction instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return location_id.complete_direction
 
     @api.constrains("scheduled_date_start")
     def check_day(self):

fieldservice/tests/test_fsm_order.py

@@ -13,7 +13,11 @@ class TestFSMOrder(TransactionCase):
     def setUpClass(cls):
         super().setUpClass()
         cls.Order = cls.env["fsm.order"]
+        cls.person_1 = cls.env.ref("fieldservice.person_1")
+        cls.person_2 = cls.env.ref("fieldservice.person_2")
+        cls.person_3 = cls.env.ref("fieldservice.person_3")
         cls.test_location = cls.env.ref("fieldservice.test_location")
+        cls.test_territory = cls.env.ref("base_territory.test_territory")
         cls.stage1 = cls.env.ref("fieldservice.fsm_stage_completed")
         cls.stage2 = cls.env.ref("fieldservice.fsm_stage_cancelled")
         cls.init_values = {
@@ -172,7 +176,6 @@ def test_fsm_order(self):
         order4.action_complete()
         order3.action_cancel()
         self.env.user.company_id.auto_populate_equipments_on_order = True
-        order._onchange_location_id_customer()
         self.assertEqual(order.custom_color, order.stage_id.custom_color)
         # Test _compute_duration
         self.assertEqual(order.duration, hours_diff)
@@ -239,17 +242,20 @@ def test_fsm_order(self):
                 }
             )
         order.description = "description"
-        order.copy_notes()
+        order.equipment_ids = equipment
+        self.assertEqual(order.description, "description", "Shouldn't have changed")
         order.description = False
-        order.copy_notes()
-        order.type = False
-        order.equipment_id = equipment.id
-        order.onchange_equipment_ids()
+        equipment.notes = "equipment notes"
+        order.equipment_ids = equipment
+        self.assertEqual(
+            order.description,
+            equipment.notes,
+            "Description should be set from equipment",
+        )
         order.type = False
         order.description = False
         self.location_1.direction = "Test Direction"
         order2.location_id.fsm_parent_id = self.location_1.id
-        order.copy_notes()
         data = (
             self.env["fsm.order"]
             .with_context(**{"default_team_id": self.test_team.id})
@@ -287,3 +293,53 @@ def test_order_unlink(self):
             order.stage_id.stage_type = "location"
             order.can_unlink()
             order.unlink()
+
+    def test_order_person_from_territory(self):
+        self.test_territory.person_ids = self.person_1 | self.person_2
+        self.test_territory.person_id = self.person_1
+        order = self.env["fsm.order"].create(
+            {
+                "location_id": self.test_location.id,
+                "stage_id": self.stage1.id,
+            }
+        )
+        self.assertEqual(
+            order.person_id,
+            self.person_1,
+            "Person should be assigned from territory",
+        )
+        # Other location with no territory, person should be kept
+        order.location_id = self.env.ref("fieldservice.location_1")
+        self.assertEqual(
+            order.person_id,
+            self.person_1,
+            "Person should be kept, because no territory is set",
+        )
+        # Other location with territory, matching person
+        other_location = self.env.ref("fieldservice.location_2")
+        other_location.territory_id = self.test_territory.copy(
+            {
+                "person_ids": (self.person_1 + self.person_2 + self.person_3).ids,
+                "person_id": self.person_3.id,
+            }
+        )
+        order.location_id = other_location
+        self.assertEqual(
+            order.person_id,
+            self.person_1,
+            "Person should be kept, because it's a territory worker",
+        )
+        # Other location with territory, unmatching person
+        other_location = self.env.ref("fieldservice.location_3")
+        other_location.territory_id = self.test_territory.copy(
+            {
+                "person_ids": (self.person_2 + self.person_3).ids,
+                "person_id": self.person_3.id,
+            }
+        )
+        order.location_id = other_location
+        self.assertEqual(
+            order.person_id,
+            self.person_3,
+            "Person should be assigned from territory",
+        )

fieldservice/tests/test_fsm_order_template_onchange.py

@@ -17,7 +17,7 @@ def setUpClass(cls):
     def test_fsm_order_onchange_template(self):
         """Test the onchange function for FSM Template
         - Category IDs, Scheduled Duration,and Type should update
-        - The copy_notes() method should be called and instructions copied
+        - The instructions should be copied
         """
         categories = []
         categories.append(self.fsm_category_a.id)