fix(stage): Handle SKIPPED tasks and synthetic stages

orca-queue/src/main/kotlin/com/netflix/spinnaker/orca/q/handler/CompleteStageHandler.kt

@@ -271,7 +271,7 @@ class CompleteStageHandler(
       allStatuses.contains(STOPPED) -> STOPPED
       allStatuses.contains(CANCELED) -> CANCELED
       allStatuses.contains(FAILED_CONTINUE) -> FAILED_CONTINUE
-      allStatuses.all { it == SUCCEEDED } -> SUCCEEDED
+      allStatuses.all { it == SUCCEEDED || it == SKIPPED } -> SUCCEEDED
       afterStageStatuses.contains(NOT_STARTED) -> RUNNING // after stages were planned but not run yet
       else -> {
         log.error("Unhandled condition for stage $id of ${execution.id}, marking as TERMINAL. syntheticStatuses=$syntheticStatuses, taskStatuses=$taskStatuses, planningStatus=$planningStatus, afterStageStatuses=$afterStageStatuses")

orca-queue/src/test/kotlin/com/netflix/spinnaker/orca/q/handler/CompleteStageHandlerTest.kt

@@ -715,6 +715,79 @@ object CompleteStageHandlerTest : SubjectSpek<CompleteStageHandler>({
       }
     }
 
+    describe("when a stage had skipped synthetics or tasks") {
+      given("some tasks were skipped") {
+        val pipeline = pipeline {
+          stage {
+            refId = "1"
+            type = multiTaskStage.type
+            multiTaskStage.plan(this)
+            tasks[0].status = SUCCEEDED
+            tasks[1].status = SKIPPED
+            tasks[2].status = SUCCEEDED
+            status = RUNNING
+          }
+        }
+        val message = CompleteStage(pipeline.stageByRef("1"))
+
+        beforeGroup {
+          whenever(repository.retrieve(PIPELINE, message.executionId)) doReturn pipeline
+        }
+
+        afterGroup(::resetMocks)
+
+        on("receiving the message") {
+          subject.handle(message)
+        }
+
+        it("updates the stage state") {
+          verify(repository).storeStage(
+            check {
+              assertThat(it.status).isEqualTo(SUCCEEDED)
+              assertThat(it.endTime).isEqualTo(clock.millis())
+            }
+          )
+        }
+      }
+
+      given("some synthetic stages were skipped") {
+        val pipeline = pipeline {
+          stage {
+            refId = "1"
+            type = stageWithSyntheticBefore.type
+            stageWithSyntheticBefore.buildBeforeStages(this)
+            stageWithSyntheticBefore.buildTasks(this)
+          }
+        }
+
+        val message = CompleteStage(pipeline.stageByRef("1"))
+
+        beforeGroup {
+          pipeline
+            .stages
+            .filter { it.syntheticStageOwner == STAGE_BEFORE }
+            .forEach { it.status = SKIPPED }
+          pipeline.stageById(message.stageId).tasks.forEach { it.status = SUCCEEDED }
+          whenever(repository.retrieve(PIPELINE, message.executionId)) doReturn pipeline
+        }
+
+        afterGroup(::resetMocks)
+
+        on("receiving the message") {
+          subject.handle(message)
+        }
+
+        it("updates the stage state") {
+          verify(repository).storeStage(
+            check {
+              assertThat(it.status).isEqualTo(SUCCEEDED)
+              assertThat(it.endTime).isEqualTo(clock.millis())
+            }
+          )
+        }
+      }
+    }
+
     given("a stage had no tasks or before stages") {
       but("does have after stages") {
         val pipeline = pipeline {