Merge branch 'master' into fix/datetime-deprecation

Author: MrButtCode

exceptions.py

@@ -3,10 +3,13 @@
 
 
 class QueuedSampleNotFoundException(Exception):
-    """Custom exception handler for queued sample not found."""
+    """Raised when a queued sample cannot be found by the given ID."""
 
-    def __init__(self, message: str) -> None:
-        Exception.__init__(self)
+    def __init__(self, message="Queued sample not found."):
+        # By providing a default, all existing raise sites that call
+        # this exception without arguments will still work, and new
+        # calls can optionally provide a custom message.
+        super().__init__(message)
         self.message = message
 
 

mod_ci/controllers.py

@@ -2437,7 +2437,9 @@ def progress_type_request(log, test, test_id, request) -> bool:
         results = g.db.query(count(TestResultFile.got)).filter(
             and_(
                 TestResultFile.test_id == test.id,
-                TestResultFile.regression_test_id.in_(results_zero_rc),
+                TestResultFile.regression_test_id.in_(
+                    results_zero_rc.select()
+                ),
                 TestResultFile.got.isnot(None)
             )
         ).scalar()
@@ -2513,13 +2515,13 @@ def update_final_status():
             finished_tests = select(TestProgress.test_id).filter(
                 and_(
                     TestProgress.status.in_([TestStatus.canceled, TestStatus.completed]),
-                    TestProgress.test_id.in_(platform_tests)
+                    TestProgress.test_id.in_(platform_tests.select())
                 )
             )
             in_progress_statuses = [TestStatus.preparation, TestStatus.completed, TestStatus.canceled]
             finished_tests_progress = g.db.query(TestProgress).filter(
                 and_(
-                    TestProgress.test_id.in_(finished_tests),
+                    TestProgress.test_id.in_(finished_tests.select()),
                     TestProgress.status.in_(in_progress_statuses)
                 )
             ).subquery()

mod_sample/controllers.py

@@ -70,11 +70,11 @@ def display_sample_info(sample) -> Dict[str, Any]:
             sq = select(RegressionTest.id).filter(RegressionTest.sample_id == sample.id)
             exit_code = g.db.query(TestResult.exit_code).filter(and_(
                 TestResult.exit_code != TestResult.expected_rc,
-                and_(TestResult.test_id == test_commit.id, TestResult.regression_test_id.in_(sq))
+                and_(TestResult.test_id == test_commit.id, TestResult.regression_test_id.in_(sq.select()))
             )).first()
             not_null = g.db.query(TestResultFile.got).filter(and_(
                 TestResultFile.got.isnot(None),
-                and_(TestResultFile.test_id == test_commit.id, TestResultFile.regression_test_id.in_(sq))
+                and_(TestResultFile.test_id == test_commit.id, TestResultFile.regression_test_id.in_(sq.select()))
             )).first()
 
             if exit_code is None and not_null is None:
@@ -88,12 +88,12 @@ def display_sample_info(sample) -> Dict[str, Any]:
             exit_code = g.db.query(TestResult.exit_code).filter(
                 and_(
                     TestResult.exit_code != TestResult.expected_rc,
-                    and_(TestResult.test_id == test_release.id, TestResult.regression_test_id.in_(sq))
+                    and_(TestResult.test_id == test_release.id, TestResult.regression_test_id.in_(sq.select()))
                 )
             ).first()
             not_null = g.db.query(TestResultFile.got).filter(and_(
                 TestResultFile.got.isnot(None),
-                and_(TestResultFile.test_id == test_release.id, TestResultFile.regression_test_id.in_(sq))
+                and_(TestResultFile.test_id == test_release.id, TestResultFile.regression_test_id.in_(sq.select()))
             )).first()
 
             if exit_code is None and not_null is None:

tests/test_upload/test_controllers.py

@@ -170,35 +170,34 @@ def test_upload_ftp_forbidden_extension(self, mock_shutil, mock_hash, mock_magic
             QueuedSample.sha == filehash).first()
         self.assertEqual(queued_sample, None)
 
-    # TODO: The methods below are not working due to decorator login_required not being mocked
-    # @mock.patch('mod_upload.controllers.login_required', side_effect=mock_decorator)
-    # def test_link_id_confirm_invalid(self, mock_login):
-    #     """
-    #     Try to confirm link for invalid sample and queue.
-    #     """
-    #     from mod_upload.controllers import link_id_confirm, QueuedSampleNotFoundException
-
-    #     with self.assertRaises(QueuedSampleNotFoundException):
-    #         link_id_confirm(1000, 1000)
-
-    # @mock.patch('mod_upload.controllers.redirect')
-    # @mock.patch('mod_upload.controllers.login_required', side_effect=mock_decorator)
-    # @mock.patch('mod_upload.controllers.QueuedSample')
-    # @mock.patch('mod_upload.controllers.Sample')
-    # def test_link_id_confirm(self, mock_sample, mock_queue, mock_login, mock_redirect):
-    #     """
-    #     Test confirm link for valid sample and queue.
-    #     """
-    #     from mod_upload.controllers import link_id_confirm
-
-    #     mock_queue.query.filter.return_value.first.return_value.user_id = g.user
-    #     mock_sample.query.filter.return_value.first.return_value.upload.user_id = g.user
-
-    #     response = link_id_confirm(1, 1)
-
-    #     self.assertEqual(response, mock_redirect())
-    #     mock_queue.query.filter.assert_called_once_with(mock_queue.id == 1)
-    #     mock_sample.query.filter.assert_called_once_with(mock_sample.id == 1)
+    def test_link_id_confirm_invalid(self):
+        """Try to confirm link for invalid sample and queue."""
+        from mod_upload.controllers import (QueuedSampleNotFoundException,
+                                            link_id_confirm)
+
+        with self.assertRaises(QueuedSampleNotFoundException):
+            link_id_confirm.__wrapped__(1000, 1000)
+
+    @mock.patch('mod_upload.controllers.redirect')
+    @mock.patch('mod_upload.controllers.QueuedSample')
+    @mock.patch('mod_upload.controllers.Sample')
+    @mock.patch('mod_upload.controllers.g')
+    def test_link_id_confirm_valid(self, mock_g, mock_sample, mock_queue, mock_redirect):
+        """Test confirm link for valid sample and queue."""
+        from mod_upload.controllers import link_id_confirm
+
+        # Mock g.user to have an 'id' attribute to satisfy the controller
+        mock_g.user.id = 1
+
+        # Ensure the mocked database queries return an object owned by that user
+        mock_queue.query.filter.return_value.first.return_value.user_id = mock_g.user.id
+        mock_sample.query.filter.return_value.first.return_value.upload.user_id = mock_g.user.id
+
+        response = link_id_confirm.__wrapped__(1, 1)
+
+        self.assertEqual(response, mock_redirect())
+        mock_queue.query.filter.assert_called_once()
+        mock_sample.query.filter.assert_called_once()
 
     def test_create_hash_for_sample(self):
         """Test creating hash for temp file."""