launchpad-reviewers team mailing list archive

[Ines Almeida <mp+492931@xxxxxxxxxxxxxxxxxx>]

Ines Almeida has proposed merging ~ines-almeida/launchpad:update-duplicate-vulnerability-job-logic into launchpad:master.

Commit message:
Update logic around duplicate VulnerabilityJobs
    
Updated the logic so that one can re-schedule a job if the previous one was superseded, or the previous one was scheduled more than 6 hours ago

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~ines-almeida/launchpad/+git/launchpad/+merge/492931
-- 
Your team Launchpad code reviewers is requested to review the proposed merge of ~ines-almeida/launchpad:update-duplicate-vulnerability-job-logic into launchpad:master.

diff --git a/lib/lp/bugs/model/exportvulnerabilityjob.py b/lib/lp/bugs/model/exportvulnerabilityjob.py
index 487ca57..37a71f4 100644
--- a/lib/lp/bugs/model/exportvulnerabilityjob.py
+++ b/lib/lp/bugs/model/exportvulnerabilityjob.py
@@ -10,7 +10,7 @@ import logging
 import os
 import tempfile
 import zipfile
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from typing import List, Optional, Tuple
 
 from zope.component import getUtility
@@ -92,9 +92,10 @@ class ExportVulnerabilityJob(VulnerabilityJobDerived):
             VulnerabilityJob.job_id == Job.id,
             VulnerabilityJob.job_type == cls.class_job_type,
             VulnerabilityJob.handler == handler,
-            Job._status.is_in(
-                (JobStatus.WAITING, JobStatus.RUNNING, JobStatus.SUSPENDED)
-            ),
+            Job._status.is_in((JobStatus.WAITING, JobStatus.RUNNING)),
+            # If the job was scheduled more than 4 hours ago (twice the soft
+            # time limit), we allow a new one to be created.
+            Job.date_created > datetime.now(timezone.utc) - timedelta(hours=4),
         ).one()
 
         if vulnerability_job is not None:
diff --git a/lib/lp/bugs/model/importvulnerabilityjob.py b/lib/lp/bugs/model/importvulnerabilityjob.py
index 951e8fc..ba89aaf 100644
--- a/lib/lp/bugs/model/importvulnerabilityjob.py
+++ b/lib/lp/bugs/model/importvulnerabilityjob.py
@@ -7,7 +7,7 @@ __all__ = [
 
 import logging
 import re
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 
 from zope.component import getUtility
 from zope.interface import implementer, provider
@@ -117,9 +117,10 @@ class ImportVulnerabilityJob(VulnerabilityJobDerived):
             VulnerabilityJob.job_id == Job.id,
             VulnerabilityJob.job_type == cls.class_job_type,
             VulnerabilityJob.handler == handler,
-            Job._status.is_in(
-                (JobStatus.WAITING, JobStatus.RUNNING, JobStatus.SUSPENDED)
-            ),
+            Job._status.is_in((JobStatus.WAITING, JobStatus.RUNNING)),
+            # If the job was scheduled more than 4 hours ago (twice the soft
+            # time limit), we allow a new one to be created.
+            Job.date_created > datetime.now(timezone.utc) - timedelta(hours=4),
         ).one()
 
         if vulnerability_job is not None:
diff --git a/lib/lp/bugs/model/tests/test_vulnerability.py b/lib/lp/bugs/model/tests/test_vulnerability.py
index bae25c1..853194f 100644
--- a/lib/lp/bugs/model/tests/test_vulnerability.py
+++ b/lib/lp/bugs/model/tests/test_vulnerability.py
@@ -2,6 +2,7 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Tests for the vulnerability and related models."""
+from datetime import timedelta
 from pathlib import Path
 
 from fixtures import MockPatch
@@ -982,6 +983,97 @@ class TestVulnerabilitySetImportData(TestCaseWithFactory):
                 import_since_commit_sha1=None,
             )
 
+    def test_importData_duplicated_suspended(self):
+        """Test that we can create a duplicated ImportVulnerabilityJob
+        if the existing job is suspended for the same handler.
+        """
+        self.useContext(feature_flags())
+        set_feature_flag(VULNERABILITY_IMPORT_ENABLED_FEATURE_FLAG, "true")
+
+        self.factory.makeDistribution(name="soss", owner=self.requester)
+
+        repo = self.factory.makeGitRepository(
+            owner=self.team, information_type=InformationType.USERDATA
+        )
+
+        self.factory.makeGitRefs(
+            repository=repo,
+            paths=[self.git_ref],
+        )
+
+        vulnerability_set = getUtility(IVulnerabilitySet)
+        with person_logged_in(self.requester):
+            vulnerability_set.importData(
+                self.requester,
+                self.handler,
+                repo,
+                self.git_ref,
+                self.git_paths,
+                self.information_type,
+                import_since_commit_sha1=None,
+            )
+
+        import_job = getUtility(IImportVulnerabilityJobSource).get(
+            self.handler
+        )
+        import_job.job.suspend()
+
+        with person_logged_in(self.requester):
+            vulnerability_set.importData(
+                self.requester,
+                self.handler,
+                repo,
+                self.git_ref,
+                self.git_paths,
+                self.information_type,
+                import_since_commit_sha1=None,
+            )
+
+    def test_importData_duplicated_old_job(self):
+        """Test that we can create a a new ImportVulnerabilityJob if there is
+        a peding one for the same handler but it is older than 4 hours.
+        """
+        self.useContext(feature_flags())
+        set_feature_flag(VULNERABILITY_IMPORT_ENABLED_FEATURE_FLAG, "true")
+
+        self.factory.makeDistribution(name="soss", owner=self.requester)
+
+        repo = self.factory.makeGitRepository(
+            owner=self.team, information_type=InformationType.USERDATA
+        )
+
+        self.factory.makeGitRefs(
+            repository=repo,
+            paths=[self.git_ref],
+        )
+
+        vulnerability_set = getUtility(IVulnerabilitySet)
+        with person_logged_in(self.requester):
+            vulnerability_set.importData(
+                self.requester,
+                self.handler,
+                repo,
+                self.git_ref,
+                self.git_paths,
+                self.information_type,
+                import_since_commit_sha1=None,
+            )
+        import_job = getUtility(IImportVulnerabilityJobSource).get(
+            self.handler
+        )
+        removeSecurityProxy(import_job.job).date_created -= timedelta(hours=5)
+
+        with person_logged_in(self.requester):
+            vulnerability_set.importData(
+                self.requester,
+                self.handler,
+                repo,
+                self.git_ref,
+                self.git_paths,
+                self.information_type,
+                import_since_commit_sha1=None,
+            )
+
 
 class TestVulnerabilitySetExportData(TestCaseWithFactory):
     layer = DatabaseFunctionalLayer
@@ -1069,6 +1161,64 @@ class TestVulnerabilitySetExportData(TestCaseWithFactory):
                 None,
             )
 
+    def test_exportData_duplicated_suspended(self):
+        """Test that we can create a new ExportVulnerabilityJob if there is
+        a suspended one for the same handler.
+        """
+        self.useContext(feature_flags())
+        set_feature_flag(VULNERABILITY_IMPORT_ENABLED_FEATURE_FLAG, "true")
+
+        self.factory.makeDistribution(name="soss", owner=self.requester)
+
+        vulnerability_set = getUtility(IVulnerabilitySet)
+        with person_logged_in(self.requester):
+            vulnerability_set.exportData(
+                self.requester,
+                self.handler,
+                None,
+            )
+
+        export_job = getUtility(IExportVulnerabilityJobSource).get(
+            self.handler
+        )
+        export_job.job.suspend()
+
+        with person_logged_in(self.requester):
+            vulnerability_set.exportData(
+                self.requester,
+                self.handler,
+                None,
+            )
+
+    def test_exportData_duplicated_old_job(self):
+        """Test that we can create a new ExportVulnerabilityJob if there is
+        a pending job for the same handler that is more than 4 hours old
+        """
+        self.useContext(feature_flags())
+        set_feature_flag(VULNERABILITY_IMPORT_ENABLED_FEATURE_FLAG, "true")
+
+        self.factory.makeDistribution(name="soss", owner=self.requester)
+
+        vulnerability_set = getUtility(IVulnerabilitySet)
+        with person_logged_in(self.requester):
+            vulnerability_set.exportData(
+                self.requester,
+                self.handler,
+                None,
+            )
+
+        export_job = getUtility(IExportVulnerabilityJobSource).get(
+            self.handler
+        )
+        removeSecurityProxy(export_job.job).date_created -= timedelta(hours=5)
+
+        with person_logged_in(self.requester):
+            vulnerability_set.exportData(
+                self.requester,
+                self.handler,
+                None,
+            )
+
 
 class TestVulnerabilitySetWebService(TestCaseWithFactory):
     layer = DatabaseFunctionalLayer
diff --git a/lib/lp/bugs/tests/test_exportvulnerabilityjob.py b/lib/lp/bugs/tests/test_exportvulnerabilityjob.py
index 4a83e6f..b0b6dec 100644
--- a/lib/lp/bugs/tests/test_exportvulnerabilityjob.py
+++ b/lib/lp/bugs/tests/test_exportvulnerabilityjob.py
@@ -141,6 +141,22 @@ class ExportVulnerabilityJobTests(TestCaseWithFactory):
         job_duplicated.complete()
         self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
 
+    def test_create_with_existing_suspended_job(self):
+        """If there's a suspended ExportVulnerabilityJob for the handler the
+        job can be runned again.
+        """
+        handler = VulnerabilityHandlerEnum.SOSS
+
+        job = self.job_source.create(handler)
+        job.start()
+        job.suspend()
+        self.assertEqual(job.status, JobStatus.SUSPENDED)
+
+        job_duplicated = self.job_source.create(handler)
+        job_duplicated.start()
+        job_duplicated.complete()
+        self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
+
     def test_arguments(self):
         """Test that ExportVulnerabilityJob specified with arguments can
         be gotten out again."""
diff --git a/lib/lp/bugs/tests/test_importvulnerabilityjob.py b/lib/lp/bugs/tests/test_importvulnerabilityjob.py
index 6ac9f77..6624f31 100644
--- a/lib/lp/bugs/tests/test_importvulnerabilityjob.py
+++ b/lib/lp/bugs/tests/test_importvulnerabilityjob.py
@@ -204,20 +204,6 @@ class ImportVulnerabilityJobTests(TestCaseWithFactory):
         )
         self.assertEqual(job, running_exception.job)
 
-        # Job status from RUNNING to SUSPENDED
-        job.suspend()
-        running_exception = self.assertRaises(
-            VulnerabilityJobInProgress,
-            self.job_source.create,
-            handler,
-            git_repository,
-            git_ref,
-            git_paths,
-            information_type,
-            import_since_commit_sha1,
-        )
-        self.assertEqual(job, running_exception.job)
-
     def test_create_with_existing_completed_job(self):
         """If there's already a completed ImportVulnerabilityJob for the
         handler the job can be run again.
@@ -302,6 +288,41 @@ class ImportVulnerabilityJobTests(TestCaseWithFactory):
         job_duplicated.complete()
         self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
 
+    def test_create_with_existing_suspended_job(self):
+        """If there's a suspended ImportVulnerabilityJob for the handler the
+        job can be run again.
+        """
+        handler = VulnerabilityHandlerEnum.SOSS
+        git_repository = self.repository.id
+        git_ref = "ref/heads/main"
+        git_paths = ["cves"]
+        information_type = InformationType.PRIVATESECURITY.value
+        import_since_commit_sha1 = None
+
+        job = self.job_source.create(
+            handler,
+            git_repository,
+            git_ref,
+            git_paths,
+            information_type,
+            import_since_commit_sha1,
+        )
+        job.start()
+        job.suspend()
+        self.assertEqual(job.status, JobStatus.SUSPENDED)
+
+        job_duplicated = self.job_source.create(
+            handler,
+            git_repository,
+            git_ref,
+            git_paths,
+            information_type,
+            import_since_commit_sha1,
+        )
+        job_duplicated.start()
+        job_duplicated.complete()
+        self.assertEqual(job_duplicated.status, JobStatus.COMPLETED)
+
     def test_arguments(self):
         """Test that ImportVulnerabilityJob specified with arguments can
         be gotten out again."""