DBC22-6067: added extra test cases to satisfy coverage

Author: bcgov-brwang

src/backend/apps/webcam/enums.py

@@ -9,6 +9,6 @@
 CAMERA_FIELD_MAPPING = {
     'name': 'cam_internetname',
     'caption': 'cam_internetcaption',
-    'is_on': 'cam_controldisabled',
-    'should_appear': 'cam_controldisappear',
+    'is_on': 'isOn',
+    'should_appear': 'should_appear',
 }

src/backend/apps/webcam/tasks.py

@@ -28,7 +28,7 @@
 from django.contrib.gis.db.models.functions import Distance
 from django.contrib.gis.geos import LineString, MultiLineString, Point
 from django.core.exceptions import ObjectDoesNotExist
-from django.db.models import F
+from django.db.models import F, BooleanField
 from huey.exceptions import CancelExecution
 from PIL import Image, ImageDraw, ImageFile, ImageFont
 from psycopg import IntegrityError
@@ -103,14 +103,14 @@ def update_cam_from_sql_db(id: int, current_time: datetime.datetime):
                 orientation=F('cam_locationsorientation'),
                 elevation=F('cam_locationselevation'),
                 isOn=Case(
-                    When(cam_controldisabled=False, then=Value(1)),
-                    default=Value(0),
-                    output_field=IntegerField()
+                    When(cam_controldisabled=0, then=Value(True)),
+                    default=Value(False),
+                    output_field=BooleanField()
                 ),
                 should_appear=Case(
-                    When(cam_controldisappear=False, then=Value(1)),
-                    default=Value(0),
-                    output_field=IntegerField()
+                    When(cam_controldisappear=0, then=Value(True)),
+                    default=Value(False),
+                    output_field=BooleanField()
                 ),
                 isOnDemand=F('cam_maintenanceis_on_demand'),
                 credit=F('cam_internetcredit'),
@@ -126,6 +126,7 @@ def update_cam_from_sql_db(id: int, current_time: datetime.datetime):
                 'orientation',
                 'elevation',
                 'isOn',
+                'should_appear',
                 'isOnDemand',
                 'credit',
                 'dbc_mark',
@@ -158,31 +159,33 @@ def format_region_name(region_name):
     return ''.join(result)
 
 def update_webcam_db(cam_id: int, cam_data: dict):
+    is_updated = False
     time_now_utc = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d%H%M%S%f")[:-3]
     camera_status = calculate_camera_status(time_now_utc)
     ts_seconds = int(camera_status["timestamp"])
     dt_utc = datetime.datetime.fromtimestamp(ts_seconds, tz=ZoneInfo("UTC"))
 
     existing_webcam = Webcam.objects.filter(id=cam_id).first()
     if not existing_webcam:
-        return False
+        return is_updated
     else:
         for field in CAMERA_DIFF_FIELDS:
             source_field = CAMERA_FIELD_MAPPING.get(field, field)
             webcam_value = getattr(existing_webcam, field)
-            source_value = cam_data.get(source_field)  
+            source_value = cam_data.get(source_field)
             if webcam_value != source_value:
                 updated_count = Webcam.objects.filter(id=cam_id).update(
                     is_on=True if cam_data.get("isOn") == 1 else False,
+                    should_appear=True if cam_data.get("should_appear") == 1 else False,
                     name=cam_data.get("cam_internetname"),
                     caption=cam_data.get("cam_internetcaption"),
                     is_new=cam_data.get("isNew"),
                     is_on_demand=cam_data.get("isOnDemand"),
                     last_update_attempt=dt_utc,
                     last_update_modified=dt_utc
                     )
-                return updated_count
-
+                is_updated = True
+        return is_updated
 
 def create_webcam_db(cam_data: dict):
     cam_id = cam_data.id

src/backend/apps/webcam/tests/test_webcam_populate.py

@@ -3,13 +3,15 @@
 import logging
 from http.client import INTERNAL_SERVER_ERROR
 from pathlib import Path
-from unittest.mock import patch
-
+from unittest.mock import MagicMock, patch
 from apps.shared.tests import BaseTest, MockResponse
 from apps.webcam.models import Webcam
 from apps.webcam.tests.test_data.webcam_parsed_feed import parsed_feed
 from django.contrib.gis.geos import Point
 from apps.webcam.tests.test_webcam_ordering import side_effect_populate
+from django.core.exceptions import ObjectDoesNotExist
+from apps.webcam.tasks import populate_webcam_from_data
+from apps.webcam.tasks import update_webcam_db_stale_delayed, update_cam_from_sql_db
 
 # suppress logged error messages to reduce noise
 logging.getLogger().setLevel(logging.CRITICAL)
@@ -121,3 +123,46 @@ def test_populate_webcam_with_validation_error(self, mock_populate):
         webcam_id_list = sorted(Webcam.objects.all().values_list("id", flat=True))
         assert webcam_id_list == [1, 2]
 
+    @patch("apps.webcam.tasks.populate_all_webcam_data")
+    def test_populate_webcam_from_data_success(self, mock_populate):
+        fake_data = MagicMock()
+        fake_data.id = 1
+
+        with patch("apps.webcam.tasks.create_webcam_db") as mock_create:
+            mock_create.return_value = (None, False)
+
+            populate_webcam_from_data(fake_data)
+
+            mock_create.assert_called_once_with(fake_data)
+
+    def test_populate_webcam_from_data_exception(self):
+        fake_data = MagicMock()
+        fake_data.id = 999
+
+        with patch("apps.webcam.tasks.create_webcam_db") as mock_create, \
+            patch("apps.webcam.tasks.logger") as mock_logger:
+
+            mock_create.side_effect = ObjectDoesNotExist()
+
+            populate_webcam_from_data(fake_data)
+
+            mock_logger.error.assert_called_once()
+
+    @patch("apps.webcam.tasks.populate_all_webcam_data")
+    def test_update_webcam_db_stale_delayed(self, mock_populate):
+        mock_populate.side_effect = lambda *args, **kwargs: side_effect_populate(self.mock_webcam_feed_result_2)
+        from apps.webcam import tasks
+        tasks.populate_all_webcam_data()
+
+        # Call the function
+        webcam = Webcam.objects.first()
+        update_webcam_db_stale_delayed(webcam)
+
+        # Refresh from DB
+        webcam.refresh_from_db()
+
+        # Assertions
+        self.assertFalse(webcam.marked_stale)
+        self.assertFalse(webcam.marked_delayed)
+        self.assertIsInstance(webcam.last_update_attempt, datetime.datetime)
+        self.assertIsInstance(webcam.last_update_modified, datetime.datetime)

src/backend/apps/webcam/tests/test_webcam_update_from_db.py

@@ -0,0 +1,103 @@
+import datetime
+import json
+from django.test import TestCase
+from unittest.mock import MagicMock, patch
+from apps.webcam.tasks import update_cam_from_sql_db, update_webcam_db
+from apps.webcam.tests.test_webcam_ordering import side_effect_populate
+from apps.webcam.models import Webcam
+from pathlib import Path
+
+webcam_feed_data_1 = open(
+            str(Path(__file__).parent) + "/test_data/webcam_feed_list_of_two.json"
+        )
+mock_webcam_feed_result_1 = json.load(webcam_feed_data_1)
+
+def _make_existing_webcam(mock_populate, mock_webcam_feed):
+    mock_populate.side_effect = lambda *args, **kwargs: side_effect_populate(mock_webcam_feed)
+    from apps.webcam import tasks
+    tasks.populate_all_webcam_data()
+
+class TestUpdateCamFromSqlDb(TestCase):
+
+    @patch("apps.webcam.tasks.populate_all_webcam_data")
+    @patch("apps.webcam.tasks.update_webcam_db")
+    @patch("apps.webcam.tasks.CameraSource")
+    def test_cam_found_calls_update_webcam_db_and_returns_true(
+        self, mock_camera_source, mock_update_webcam_db, mock_populate
+    ):
+
+        _make_existing_webcam(mock_populate, mock_webcam_feed_result_1)
+        
+        mock_cam = Webcam.objects.get(id=1)
+
+        # Wire up the full ORM chain so .using("mssql") never hits the real DB
+        (mock_camera_source.objects
+            .using.return_value
+            .filter.return_value
+            .annotate.return_value
+            .values.return_value
+            .first.return_value) = mock_cam
+
+        mock_update_webcam_db.return_value = 1
+
+        result = update_cam_from_sql_db(id=1, current_time=datetime.datetime.now())
+
+        self.assertTrue(result)
+        mock_update_webcam_db.assert_called_once_with(1, mock_cam)
+        mock_camera_source.objects.using.assert_called_once_with("mssql")
+
+    @patch("apps.webcam.tasks.CameraSource")
+    def test_orm_exception_returns_empty_dict(self, mock_camera_source):
+        # Simulates a DB connection failure
+        mock_camera_source.objects.using.side_effect = Exception("mssql unreachable")
+
+        result = update_cam_from_sql_db(id=1, current_time=datetime.datetime.now())
+
+        self.assertEqual(result, {})
+
+
+
+class TestUpdateWebcamDb(TestCase):
+
+    @patch("apps.webcam.tasks.populate_all_webcam_data")
+    @patch("apps.webcam.tasks.calculate_camera_status")
+    def test_webcam_not_found_returns_false(self, mock_calc_status, mock_populate):
+        _make_existing_webcam(mock_populate, mock_webcam_feed_result_1)
+        webcam = Webcam.objects.filter(id=1).first()
+
+        camera_source_data = {
+            "id": 1,
+            "cam_internetname": "Test Cam",
+            "cam_internetcaption": "Test Caption",
+            "isOn": True,
+            "should_appear": True,
+            "isNew": 0,
+            "isOnDemand": 0,
+        }
+
+        result = update_webcam_db(cam_id=1, cam_data=camera_source_data)
+        webcam.refresh_from_db()
+        self.assertEqual(webcam.name, "Test Cam")
+        self.assertEqual(webcam.caption, "Test Caption")
+        self.assertEqual(webcam.is_on, True)
+        self.assertEqual(webcam.should_appear, True)
+        self.assertEqual(result, True)
+
+        result = update_webcam_db(cam_id=1, cam_data=camera_source_data)
+        webcam.refresh_from_db()
+        self.assertEqual(webcam.name, "Test Cam")
+        self.assertEqual(webcam.caption, "Test Caption")
+        self.assertEqual(webcam.is_on, True)
+        self.assertEqual(webcam.should_appear, True)
+        self.assertEqual(result, False)
+
+        camera_source_data["cam_internetcaption"] = "Updated Caption"
+        result = update_webcam_db(cam_id=1, cam_data=camera_source_data)
+        webcam.refresh_from_db()
+        self.assertEqual(webcam.caption, "Updated Caption")
+        self.assertEqual(result, True)
+
+        # Ensure DB has NO webcam
+        Webcam.objects.all().delete()
+        result = update_webcam_db(cam_id=1, cam_data=camera_source_data)
+        self.assertEqual(result, False)