Merge pull request #469 from MTES-MCT/feat/business-types-in-controls

Feat/business types in controls

app/data_access/control_data.py

@@ -22,6 +22,7 @@
     graphene_enum_type,
 )
 from app.helpers.submitter_type import SubmitterType
+from app.models import Business
 from app.models.controller_control import ControllerControl, ControlType
 from app.models.regulation_check import RegulationCheckType
 
@@ -190,6 +191,10 @@ class Meta:
     siren = graphene.String()
     company_address = graphene.String()
     mission_address_begin = graphene.String()
+    business_type_during_control = graphene.Field(
+        lambda: BusinessOutput,
+        description="Type d'activité effectuée par le salarié au moment du contrôle",
+    )
     control_type = graphene.String()
     is_day_page_filled = graphene.Boolean(
         required=False,
@@ -228,6 +233,17 @@ def resolve_employments(
             key=lambda e: e.start_date,
             reverse=True,
         )
+        employment_businesses = {
+            int(k): v
+            for k, v in self.control_bulletin.get(
+                "employments_business_types", {}
+            ).items()
+        }
+        for employment in employments:
+            if not employment.id in employment_businesses:
+                continue
+            employment.business_id = employment_businesses[employment.id]
+
         return employments
 
     def resolve_missions(self, info, mission_id=None):
@@ -261,6 +277,14 @@ def resolve_regulation_computations_by_day(
             for day_, computations_ in regulation_computations_by_day.items()
         ]
 
+    def resolve_business_type_during_control(self, info):
+        business_id = self.control_bulletin.get("business_id")
+        if business_id:
+            return Business.query.filter(
+                Business.id == self.control_bulletin.get("business_id")
+            ).one_or_none()
+        return None
+
     def resolve_observed_infractions(self, info):
         observed_infractions = []
 

app/models/controller_control.py

@@ -229,10 +229,17 @@ def get_or_create_mobilic_control(
             return existing_control
         else:
             controlled_user = User.query.get(user_id)
+            current_employments = controlled_user.active_employments_at(
+                date_=qr_code_generation_time.date()
+            )
             company_name = ""
             vehicle_registration_number = ""
-            control_bulletin = {}
-
+            control_bulletin = {
+                "employments_business_types": {
+                    e.id: e.business.id if e.business else None
+                    for e in current_employments
+                }
+            }
             latest_activity_before = controlled_user.latest_activity_before(
                 qr_code_generation_time
             )
@@ -271,6 +278,16 @@ def get_or_create_mobilic_control(
                             "mission_address_begin"
                         ] = latest_mission.start_location.address.format()
 
+                    current_employments_for_company = [
+                        e
+                        for e in current_employments
+                        if e.company_id == latest_mission.company.id
+                    ]
+                    if len(current_employments_for_company) > 0:
+                        control_bulletin[
+                            "business_id"
+                        ] = current_employments_for_company[0].business_id
+
             work_days = group_user_events_by_day_with_limit(
                 user=controlled_user,
                 from_date=compute_history_start_date(

app/tests/controls/__init__.py

@@ -0,0 +1,139 @@
+from datetime import datetime
+
+from flask.ctx import AppContext
+
+from app import db, app
+from app.models import Vehicle, Business
+from app.models.activity import ActivityType
+from app.models.business import BusinessType
+from app.seed import ControllerUserFactory, UserFactory, CompanyFactory
+from app.seed.factories import ControllerControlFactory
+from app.tests import BaseTest
+from app.tests.helpers import (
+    init_regulation_checks_data,
+    init_businesses_data,
+    make_authenticated_request,
+    ApiRequests,
+)
+
+COMPANY_NAME_1 = "Company 1"
+VEHICLE_ID_1 = "AAA 1"
+
+
+class ControlsTestSimple(BaseTest):
+    def setUp(self):
+        super().setUp()
+        init_regulation_checks_data()
+        init_businesses_data()
+        self.controller_user_1 = ControllerUserFactory.create()
+        self.controller_user_2 = ControllerUserFactory.create()
+        self.controlled_user_1 = UserFactory.create()
+        self.controlled_user_2 = UserFactory.create()
+
+    def _create_control(
+        self,
+        controlled_user,
+        controller_user=None,
+        qr_code_generation_time=None,
+    ):
+        if not controller_user:
+            controller_user = self.controller_user_1
+
+        controller_control = (
+            ControllerControlFactory.create(
+                user_id=controlled_user.id,
+                controller_id=controller_user.id,
+                qr_code_generation_time=qr_code_generation_time,
+                creation_time=qr_code_generation_time,
+            )
+            if qr_code_generation_time
+            else ControllerControlFactory.create(
+                user_id=controlled_user.id, controller_id=controller_user.id
+            )
+        )
+        return controller_control.id
+
+    def _query_controller_info(self, controller, from_date=None):
+        response = make_authenticated_request(
+            time=datetime.now(),
+            submitter_id=controller.id,
+            query=ApiRequests.get_controller_user_info,
+            variables=dict(id=controller.id, from_date=from_date),
+            request_by_controller_user=True,
+            unexposed_query=True,
+        )
+        return response["data"]["controllerUser"]
+
+
+class ControlsTest(ControlsTestSimple):
+    def setUp(self):
+        super().setUp()
+
+        self.company1 = CompanyFactory.create(usual_name=COMPANY_NAME_1)
+        self.employee_1 = UserFactory.create(
+            first_name="Tim", last_name="Leader", post__company=self.company1
+        )
+        self.admin_1 = UserFactory.create(
+            post__company=self.company1, post__has_admin_rights=True
+        )
+        self.vehicle1 = Vehicle(
+            registration_number=VEHICLE_ID_1,
+            alias=f"Vehicule 1",
+            submitter=self.admin_1,
+            company_id=self.company1.id,
+        )
+        db.session.add(self.vehicle1)
+        db.session.commit()
+
+        self._app_context = AppContext(app)
+        self._app_context.__enter__()
+
+    def tearDown(self):
+        self._app_context.__exit__(None, None, None)
+        super().tearDown()
+
+    def _create_mission(self, employee, company, vehicle, time=None):
+        create_mission_response = make_authenticated_request(
+            time=time,
+            submitter_id=employee.id,
+            query=ApiRequests.create_mission,
+            variables={
+                "company_id": company.id,
+                "vehicle_id": vehicle.id,
+            },
+        )
+        return create_mission_response["data"]["activities"]["createMission"][
+            "id"
+        ]
+
+    def _convert_employee_to_trm_short_distance(self):
+        trm_short_distance_business = Business.query.filter(
+            Business.business_type == BusinessType.SHORT_DISTANCE.value
+        ).one_or_none()
+        self.employee_1.employments[0].business = trm_short_distance_business
+        db.session.commit()
+
+    def _convert_employee_to_trm_long_distance(self):
+        trm_long_distance_business = Business.query.filter(
+            Business.business_type == BusinessType.LONG_DISTANCE.value
+        ).one_or_none()
+        self.employee_1.employments[0].business = trm_long_distance_business
+        db.session.commit()
+
+    def _log_drive_in_mission(
+        self, mission_id, employee, start_time, end_time=None
+    ):
+        make_authenticated_request(
+            time=None,
+            submitter_id=employee.id,
+            query=ApiRequests.log_activity,
+            variables=dict(
+                start_time=start_time,
+                end_time=end_time,
+                mission_id=mission_id,
+                type=ActivityType.DRIVE,
+                user_id=employee.id,
+                switch=False,
+            ),
+            request_should_fail_with=None,
+        )

app/tests/controls/test_business_type_in_control.py

@@ -0,0 +1,143 @@
+from datetime import datetime
+
+from app import db
+from app.models import Business
+from app.models.business import BusinessType
+from app.models.controller_control import ControllerControl
+from app.seed import CompanyFactory, EmploymentFactory
+from app.seed.helpers import get_time
+from app.tests.controls import ControlsTest
+from app.tests.helpers import make_authenticated_request, ApiRequests
+
+
+class TestBusinessTypeInControl(ControlsTest):
+    def create_mission(self):
+        return self._create_mission(
+            employee=self.employee_1,
+            company=self.company1,
+            vehicle=self.vehicle1,
+        )
+
+    def _get_control_data_(self, control):
+        return make_authenticated_request(
+            time=datetime.now(),
+            submitter_id=self.controller_user_1.id,
+            query=ApiRequests.read_control_data,
+            variables=dict(
+                control_id=control.id,
+            ),
+            request_by_controller_user=True,
+            unexposed_query=True,
+        )
+
+    def _get_control_data_business_type(self, control):
+        response = self._get_control_data_(control=control)
+        return response["data"]["controlData"]["businessTypeDuringControl"]
+
+    def test_employee_controlled_without_mission_no_business_id(self):
+        # When flashing an employee not currently working
+        control = ControllerControl.get_or_create_mobilic_control(
+            controller_id=self.controller_user_1.id,
+            user_id=self.employee_1.id,
+            qr_code_generation_time=get_time(how_many_days_ago=1, hour=11),
+        )
+
+        # No business_id is stored in the control bulletin
+        self.assertIsNone(control.control_bulletin.get("business_id"))
+
+        control_business_type = self._get_control_data_business_type(
+            control=control
+        )
+        self.assertIsNone(control_business_type)
+
+    def test_employee_controlled_with_mission_correct_business_id(self):
+        # An employee has a current mission for SHORT_DISTANCE
+        self._convert_employee_to_trm_short_distance()
+        mission_id = self.create_mission()
+        start_time = get_time(how_many_days_ago=1, hour=10)
+        end_time = get_time(how_many_days_ago=1, hour=11)
+        self._log_drive_in_mission(
+            mission_id, self.employee_1, start_time, end_time
+        )
+
+        control = ControllerControl.get_or_create_mobilic_control(
+            controller_id=self.controller_user_1.id,
+            user_id=self.employee_1.id,
+            qr_code_generation_time=get_time(how_many_days_ago=1, hour=11),
+        )
+        self.assertIsNotNone(control.control_bulletin.get("business_id"))
+
+        # control data shows business type at control time
+        control_business_type = self._get_control_data_business_type(
+            control=control
+        )
+        self.assertEqual(
+            control_business_type["businessType"],
+            BusinessType.SHORT_DISTANCE.name,
+        )
+
+        # even if employee change business type after that
+        self._convert_employee_to_trm_long_distance()
+        control_business_type = self._get_control_data_business_type(
+            control=control
+        )
+        self.assertEqual(
+            control_business_type["businessType"],
+            BusinessType.SHORT_DISTANCE.name,
+        )
+
+    def test_employee_with_several_employments_correct_business_types(self):
+        company2 = CompanyFactory.create(usual_name="Company 2")
+        EmploymentFactory.create(
+            company=company2,
+            submitter=self.admin_1,
+            user=self.admin_1,
+            has_admin_rights=True,
+        )
+        EmploymentFactory.create(
+            company=company2,
+            submitter=self.admin_1,
+            user=self.employee_1,
+            has_admin_rights=False,
+        )
+        trm_short_distance_business = Business.query.filter(
+            Business.business_type == BusinessType.SHORT_DISTANCE.value
+        ).one_or_none()
+        trm_long_distance_business = Business.query.filter(
+            Business.business_type == BusinessType.LONG_DISTANCE.value
+        ).one_or_none()
+        vtc_business = Business.query.filter(
+            Business.business_type == BusinessType.VTC.value
+        ).one_or_none()
+
+        employments = self.employee_1.employments
+        for e in employments:
+            if e.company.name == self.company1.name:
+                e.business = trm_short_distance_business
+            else:
+                e.business = trm_long_distance_business
+        db.session.commit()
+
+        control = ControllerControl.get_or_create_mobilic_control(
+            controller_id=self.controller_user_1.id,
+            user_id=self.employee_1.id,
+            qr_code_generation_time=datetime.now(),
+        )
+
+        employments[0].business = vtc_business
+        employments[1].business = vtc_business
+        db.session.commit()
+
+        response = self._get_control_data_(control=control)
+        response_employments = response["data"]["controlData"]["employments"]
+        for e in response_employments:
+            if e["company"]["name"] == self.company1.name:
+                self.assertEqual(
+                    e["business"]["businessType"],
+                    BusinessType.SHORT_DISTANCE.name,
+                )
+            else:
+                self.assertEqual(
+                    e["business"]["businessType"],
+                    BusinessType.LONG_DISTANCE.name,
+                )