def get_valid_stage()

in fbpcs/pl_coordinator/pl_instance_runner.py [0:0]

• 17 lines of code
• 10 McCabe index (conditional complexity)

    def get_valid_stage(self) -> Optional[PrivateComputationBaseStageFlow]:
        if not self.is_finished():
            publisher_stage = self.publisher.get_valid_stage(self.stage_flow)
            partner_stage = self.partner.get_valid_stage(self.stage_flow)

            # expected for all joint stages
            if publisher_stage is partner_stage:
                return publisher_stage

            elif publisher_stage is None:
                return partner_stage
            elif partner_stage is None:
                return publisher_stage

            elif publisher_stage is partner_stage.previous_stage:
                if not publisher_stage.is_joint_stage:
                    # Example: publisher is PREPARE_DATA_FAILED, partner is PREPARE_DATA_COMPLETED
                    return publisher_stage
            elif partner_stage is publisher_stage.previous_stage:
                if not partner_stage.is_joint_stage:
                    # Example: publisher is PREPARE_DATA_COMPLETED, partner is PREPARE_DATA_FAILED
                    return partner_stage
        return None