diff --git a/packit_service/worker/handlers/distgit.py b/packit_service/worker/handlers/distgit.py index b56669279..6c87918e0 100644 --- a/packit_service/worker/handlers/distgit.py +++ b/packit_service/worker/handlers/distgit.py @@ -238,6 +238,8 @@ def __init__( event: dict, celery_task: Task, sync_release_run_id: Optional[int] = None, + retry_tag: Optional[str] = None, + retry_version: Optional[str] = None, ): super().__init__( package_config=package_config, @@ -247,6 +249,8 @@ def __init__( ) self._project_url = self.data.project_url self._sync_release_run_id = sync_release_run_id + self._retry_tag = retry_tag + self._retry_version = retry_version self.helper: Optional[SyncReleaseHelper] = None @property @@ -316,6 +320,8 @@ def sync_branch( # is not retried also automatically kargs = self.celery_task.task.request.kwargs.copy() kargs["sync_release_run_id"] = model.id + kargs["retry_tag"] = tag + kargs["retry_version"] = version # https://docs.celeryq.dev/en/stable/userguide/tasks.html#retrying # https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task.retry self.celery_task.task.retry( @@ -344,8 +350,30 @@ def sync_branch( return downstream_pr, additional_prs - def _get_or_create_sync_release_run(self, project_event_model=None) -> SyncReleaseModel: - if self._sync_release_run_id is not None: + def _get_or_create_sync_release_run( + self, + project_event_model: Optional[ProjectEventModel] = None, + tag: Optional[str] = None, + version: Optional[str] = None, + ) -> SyncReleaseModel: + """Get the existing sync release run model if retrying, or create a new one. + + On retry, the existing model is reused only when the tag and version + match the retry parameters, so that other versions in a multi-version + run get their own models. + + Args: + project_event_model: The project event model associated with the run. + tag: The tag of the release being synced. + version: The version of the release being synced. + + Returns: + The SyncReleaseModel representing the run. + """ + if self._sync_release_run_id is not None and ( + (self._retry_tag is None and self._retry_version is None) + or (tag == self._retry_tag and version == self._retry_version) + ): return SyncReleaseModel.get_by_id(self._sync_release_run_id) sync_release_model, _ = SyncReleaseModel.create_with_new_run( @@ -605,7 +633,9 @@ def _run_for_release( if tag and self.data.event_type == anitya.NewHotness.event_type() else None ) - sync_release_run_model = self._get_or_create_sync_release_run(release_event) + sync_release_run_model = self._get_or_create_sync_release_run( + release_event, tag=tag, version=version + ) branches_to_run = [target.branch for target in sync_release_run_model.sync_release_targets] logger.debug( f"Branches to run {self.job_config.type} " @@ -720,6 +750,8 @@ def __init__( event: dict, celery_task: Task, sync_release_run_id: Optional[int] = None, + retry_tag: Optional[str] = None, + retry_version: Optional[str] = None, ): super().__init__( package_config=package_config, @@ -727,6 +759,8 @@ def __init__( event=event, celery_task=celery_task, sync_release_run_id=sync_release_run_id, + retry_tag=retry_tag, + retry_version=retry_version, ) @staticmethod @@ -782,6 +816,8 @@ def __init__( event: dict, celery_task: Task, sync_release_run_id: Optional[int] = None, + retry_tag: Optional[str] = None, + retry_version: Optional[str] = None, ): super().__init__( package_config=package_config, @@ -789,6 +825,8 @@ def __init__( event=event, celery_task=celery_task, sync_release_run_id=sync_release_run_id, + retry_tag=retry_tag, + retry_version=retry_version, ) if self.data.event_type in (pagure.pr.Comment.event_type(),): # use upstream project URL when retriggering from dist-git PR diff --git a/packit_service/worker/tasks.py b/packit_service/worker/tasks.py index f04302eec..58d486286 100644 --- a/packit_service/worker/tasks.py +++ b/packit_service/worker/tasks.py @@ -469,12 +469,16 @@ def run_propose_downstream_handler( package_config: dict, job_config: dict, sync_release_run_id: Optional[int] = None, + retry_tag: Optional[str] = None, + retry_version: Optional[str] = None, ): handler = ProposeDownstreamHandler( package_config=load_package_config(package_config), job_config=load_job_config(job_config), event=event, sync_release_run_id=sync_release_run_id, + retry_tag=retry_tag, + retry_version=retry_version, celery_task=self, ) return get_handlers_task_results(handler.run_job(), event) @@ -494,12 +498,16 @@ def run_pull_from_upstream_handler( package_config: dict, job_config: dict, sync_release_run_id: Optional[int] = None, + retry_tag: Optional[str] = None, + retry_version: Optional[str] = None, ): handler = PullFromUpstreamHandler( package_config=load_package_config(package_config), job_config=load_job_config(job_config), event=event, sync_release_run_id=sync_release_run_id, + retry_tag=retry_tag, + retry_version=retry_version, celery_task=self, ) return get_handlers_task_results(handler.run_job(), event) diff --git a/tests/integration/test_new_hotness_update.py b/tests/integration/test_new_hotness_update.py index c08e5d598..74e50c78a 100644 --- a/tests/integration/test_new_hotness_update.py +++ b/tests/integration/test_new_hotness_update.py @@ -579,6 +579,209 @@ class AnityaTestProjectModel(AnityaProjectModel): assert first_dict_value(results["job"])["success"] +def test_retry_pull_from_upstream_multi_version(new_hotness_update): + """On retry with multiple versions, only the failed version should reuse its + SyncReleaseModel. Other versions must create their own models.""" + new_hotness_update["trigger"]["msg"]["message"]["upstream_versions"] = ["7.0.3", "7.0.4"] + + class AnityaTestProjectModel(AnityaProjectModel): + pass + + db_project_object = flexmock( + id=12, + project_event_model_type=ProjectEventModelType.anitya_multiple_versions, + job_config_trigger_type=JobConfigTriggerType.release, + project=AnityaTestProjectModel(), + ) + project_event = ( + flexmock(type=ProjectEventModelType.anitya_multiple_versions, event_id=12) + .should_receive("get_project_event_object") + .and_return(db_project_object) + .mock() + ) + run_model = flexmock(PipelineModel) + flexmock(ProjectEventModel).should_receive("get_or_create").with_args( + type=ProjectEventModelType.anitya_multiple_versions, + event_id=12, + commit_sha=None, + ).and_return(project_event) + flexmock(AnityaMultipleVersionsModel).should_receive("get_or_create").with_args( + versions=["7.0.3", "7.0.4"], + project_name="redis", + project_id=4181, + package="redis", + ).and_return(db_project_object) + + # The retried model (for version 7.0.4 which failed download previously) + retried_target = flexmock(status="retry", id=1234, branch="main") + retried_model = flexmock(id=123, sync_release_targets=[retried_target]) + flexmock(SyncReleaseModel).should_receive("get_by_id").with_args(123).and_return( + retried_model, + ).once() + + # A new model should be created for version 7.0.3 (never attempted before) + new_target = flexmock(status="queued", id=1235, branch="main") + new_model = flexmock(id=124, sync_release_targets=[]) + flexmock(SyncReleaseModel).should_receive("create_with_new_run").with_args( + status=SyncReleaseStatus.running, + project_event_model=project_event, + job_type=SyncReleaseJobType.pull_from_upstream, + package_name="redis", + ).and_return(new_model, run_model).once() + flexmock(SyncReleaseTargetModel).should_receive("create").with_args( + status=SyncReleaseTargetStatus.queued, + branch="main", + ).and_return(new_target).once() + + flexmock(SyncReleasePullRequestModel).should_receive("get_or_create").with_args( + pr_id=21, + namespace="downstream-namespace", + repo_name="downstream-repo", + project_url="https://src.fedoraproject.org/rpms/downstream-repo", + target_branch=str, + url=str, + ).and_return(flexmock(sync_release_targets=[flexmock()])) + + packit_yaml = ( + "{'specfile_path': 'hello-world.spec', " + "jobs: [{trigger: release, job: pull_from_upstream, metadata: {targets:[]}}]}" + ) + flexmock(Github, get_repo=lambda full_name_or_id: None) + distgit_project = flexmock( + get_files=lambda ref, recursive: [".packit.yaml"], + get_file_content=lambda path, ref, headers: packit_yaml, + full_repo_name="rpms/redis", + repo="redis", + namespace="rpms", + is_private=lambda: False, + default_branch="main", + ) + + lp = flexmock(LocalProject, refresh_the_arguments=lambda: None) + flexmock(LocalProjectBuilder, _refresh_the_state=lambda *args: lp) + lp.working_dir = "" + lp.git_project = distgit_project + flexmock(DistGit).should_receive("local_project").and_return(lp) + + flexmock(Allowlist, check_and_report=True) + + flexmock( + packit_service.worker.handlers.distgit, + ).should_receive("get_monitoring_metadata").and_return( + flexmock(all_versions=True), + ) + + service_config = ServiceConfig().get_service_config() + flexmock(service_config).should_receive("get_project").with_args( + "https://src.fedoraproject.org/rpms/redis", + required=False, + ).and_return(distgit_project) + flexmock(service_config).should_receive("get_project").with_args( + "https://src.fedoraproject.org/rpms/redis", + ).and_return(distgit_project) + + target_project = ( + flexmock(namespace="downstream-namespace", repo="downstream-repo") + .should_receive("get_web_url") + .and_return("https://src.fedoraproject.org/rpms/downstream-repo") + .mock() + ) + pr = ( + flexmock( + id=21, + url="some_url", + target_project=target_project, + description="some-title", + ) + .should_receive("comment") + .mock() + ) + # 7.0.4 (retried version) succeeds this time + flexmock(PackitAPI).should_receive("sync_release").with_args( + dist_git_branch="main", + versions=["7.0.4"], + create_pr=True, + local_pr_branch_suffix="update-pull_from_upstream", + use_downstream_specfile=True, + add_pr_instructions=True, + resolved_bugs=["rhbz#2106196"], + release_monitoring_project_id=4181, + sync_acls=True, + pr_description_footer=DistgitAnnouncement.get_announcement(), + add_new_sources=True, + fast_forward_merge_branches=set(), + ).and_return((pr, {})).once().ordered() + # 7.0.3 (new version) also succeeds + flexmock(PackitAPI).should_receive("sync_release").with_args( + dist_git_branch="main", + versions=["7.0.3"], + create_pr=True, + local_pr_branch_suffix="update-pull_from_upstream", + use_downstream_specfile=True, + add_pr_instructions=True, + resolved_bugs=["rhbz#2106196"], + release_monitoring_project_id=4181, + sync_acls=True, + pr_description_footer=DistgitAnnouncement.get_announcement(), + add_new_sources=True, + fast_forward_merge_branches=set(), + ).and_return((pr, {})).once().ordered() + flexmock(PackitAPI).should_receive("clean") + + for target, model in [ + (retried_target, retried_model), + (new_target, new_model), + ]: + flexmock(target).should_receive("set_status").with_args( + status=SyncReleaseTargetStatus.running, + ).once() + flexmock(target).should_receive("set_downstream_pr_url").with_args( + downstream_pr_url="some_url", + ) + flexmock(target).should_receive("set_downstream_prs").with_args( + downstream_prs=list, + ).once() + flexmock(target).should_receive("set_status").with_args( + status=SyncReleaseTargetStatus.submitted, + ).once() + flexmock(target).should_receive("set_start_time").once() + flexmock(target).should_receive("set_finished_time").once() + flexmock(target).should_receive("set_logs").once() + flexmock(model).should_receive("set_status").with_args( + status=SyncReleaseStatus.finished, + ).once() + model.should_receive("get_package_name").and_return(None) + + flexmock(IsRunConditionSatisfied).should_receive("pre_check").and_return(True) + + flexmock(AddReleaseEventToDb).should_receive("db_project_object").and_return( + flexmock( + job_config_trigger_type=JobConfigTriggerType.release, + id=123, + project_event_model_type=ProjectEventModelType.release, + ), + ) + flexmock(group).should_receive("apply_async").once() + flexmock(Pushgateway).should_receive("push").times(2).and_return() + flexmock(shutil).should_receive("rmtree").with_args("") + + processing_results = SteveJobs().process_message(new_hotness_update) + event_dict, _, job_config, package_config = get_parameters_from_results( + processing_results, + ) + assert json.dumps(event_dict) + + # Simulate retry: pass sync_release_run_id and retry_version for 7.0.4 + results = run_pull_from_upstream_handler( + package_config=package_config, + event=event_dict, + job_config=job_config, + sync_release_run_id=123, + retry_version="7.0.4", + ) + assert first_dict_value(results["job"])["success"] + + def test_new_hotness_update_non_git(new_hotness_update, sync_release_model_non_git): model = flexmock(status="queued", id=1234, branch="main") flexmock(SyncReleaseTargetModel).should_receive("create").with_args(