|
| 1 | +package runstream |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "testing" |
| 6 | + "time" |
| 7 | + |
| 8 | + "github.com/nats-io/nats.go" |
| 9 | +) |
| 10 | + |
| 11 | +// Test_PublishTFRunEvent_DedupesSameRunIDAndStatus verifies that two |
| 12 | +// publishes for the same (runID, newStatus) result in a single message on |
| 13 | +// the RUN_EVENTS stream. This is the core fix for duplicate GitLab reply |
| 14 | +// comments — TFC fires a webhook for every notification trigger, and several |
| 15 | +// triggers can resolve to the same RunStatus over the run's lifecycle. |
| 16 | +func Test_PublishTFRunEvent_DedupesSameRunIDAndStatus(t *testing.T) { |
| 17 | + _, url := startTestNATS(t) |
| 18 | + nc := testConnect(t, url) |
| 19 | + defer nc.Close() |
| 20 | + js := testGetJetstreamContext(t, nc) |
| 21 | + |
| 22 | + s := newTestStream(t, js) |
| 23 | + seedRunMetadata(t, s, "run-DEDUP", "gitlab") |
| 24 | + |
| 25 | + publishOnce(t, s, "run-DEDUP", "planning") |
| 26 | + publishOnce(t, s, "run-DEDUP", "planning") |
| 27 | + publishOnce(t, s, "run-DEDUP", "planning") |
| 28 | + |
| 29 | + if got := streamMsgs(t, js); got != 1 { |
| 30 | + t.Fatalf("expected 1 message on %s after 3 same-status publishes, got %d", |
| 31 | + RunEventsStreamName, got) |
| 32 | + } |
| 33 | +} |
| 34 | + |
| 35 | +// Test_PublishTFRunEvent_KeepsDistinctStatuses verifies that the dedup is |
| 36 | +// scoped to (runID, newStatus). Different status values for the same run |
| 37 | +// must all land on the stream — that's how real state transitions reach the |
| 38 | +// GitLab reply pipeline. |
| 39 | +func Test_PublishTFRunEvent_KeepsDistinctStatuses(t *testing.T) { |
| 40 | + _, url := startTestNATS(t) |
| 41 | + nc := testConnect(t, url) |
| 42 | + defer nc.Close() |
| 43 | + js := testGetJetstreamContext(t, nc) |
| 44 | + |
| 45 | + s := newTestStream(t, js) |
| 46 | + seedRunMetadata(t, s, "run-DISTINCT", "gitlab") |
| 47 | + |
| 48 | + for _, status := range []string{"planning", "planned", "applying", "applied"} { |
| 49 | + publishOnce(t, s, "run-DISTINCT", status) |
| 50 | + } |
| 51 | + |
| 52 | + if got := streamMsgs(t, js); got != 4 { |
| 53 | + t.Fatalf("expected 4 messages on %s for 4 distinct statuses, got %d", |
| 54 | + RunEventsStreamName, got) |
| 55 | + } |
| 56 | +} |
| 57 | + |
| 58 | +// Test_PublishTFRunEvent_DedupesAcrossRuns verifies that the dedup is keyed |
| 59 | +// to runID so different runs at the same status are not collapsed. |
| 60 | +func Test_PublishTFRunEvent_DedupesAcrossRuns(t *testing.T) { |
| 61 | + _, url := startTestNATS(t) |
| 62 | + nc := testConnect(t, url) |
| 63 | + defer nc.Close() |
| 64 | + js := testGetJetstreamContext(t, nc) |
| 65 | + |
| 66 | + s := newTestStream(t, js) |
| 67 | + seedRunMetadata(t, s, "run-A", "gitlab") |
| 68 | + seedRunMetadata(t, s, "run-B", "gitlab") |
| 69 | + |
| 70 | + publishOnce(t, s, "run-A", "planning") |
| 71 | + publishOnce(t, s, "run-A", "planning") // dedup |
| 72 | + publishOnce(t, s, "run-B", "planning") // distinct run, must land |
| 73 | + publishOnce(t, s, "run-B", "planning") // dedup |
| 74 | + |
| 75 | + if got := streamMsgs(t, js); got != 2 { |
| 76 | + t.Fatalf("expected 2 messages (one per runID) on %s, got %d", |
| 77 | + RunEventsStreamName, got) |
| 78 | + } |
| 79 | +} |
| 80 | + |
| 81 | +// Test_PublishTFRunEvent_PlanThenApplyFlow simulates the manual |
| 82 | +// plan-then-`tfc apply` workflow. Plan and apply are two separate TFC runs |
| 83 | +// with distinct runIDs, but both pass through the planning -> planned |
| 84 | +// status transitions before the apply run progresses to applying -> applied. |
| 85 | +// Each run must own its own set of dedup messages — none collide. |
| 86 | +func Test_PublishTFRunEvent_PlanThenApplyFlow(t *testing.T) { |
| 87 | + _, url := startTestNATS(t) |
| 88 | + nc := testConnect(t, url) |
| 89 | + defer nc.Close() |
| 90 | + js := testGetJetstreamContext(t, nc) |
| 91 | + |
| 92 | + s := newTestStream(t, js) |
| 93 | + seedRunMetadata(t, s, "run-PLAN", "gitlab") |
| 94 | + seedRunMetadata(t, s, "run-APPLY", "gitlab") |
| 95 | + |
| 96 | + // Plan run: TFC walks pending -> planning -> planned -> planned_and_finished. |
| 97 | + // Each redundant webhook within a status (e.g. plan_queued + planning |
| 98 | + // both surface as `planning`) is collapsed by dedup. |
| 99 | + planStatuses := []string{"pending", "planning", "planning", "planned", "planned_and_finished"} |
| 100 | + for _, status := range planStatuses { |
| 101 | + publishOnce(t, s, "run-PLAN", status) |
| 102 | + } |
| 103 | + |
| 104 | + // Apply run: separate runID. TFC runs always plan first, so this run |
| 105 | + // also passes through `planning` and `planned` before reaching the apply |
| 106 | + // phase. Same status names, different runID = distinct dedup keys. |
| 107 | + applyStatuses := []string{"pending", "planning", "planned", "applying", "applying", "applied", "applied"} |
| 108 | + for _, status := range applyStatuses { |
| 109 | + publishOnce(t, s, "run-APPLY", status) |
| 110 | + } |
| 111 | + |
| 112 | + // Expected unique (runID, status) pairs: |
| 113 | + // run-PLAN: pending, planning, planned, planned_and_finished (4) |
| 114 | + // run-APPLY: pending, planning, planned, applying, applied (5) |
| 115 | + const want = 9 |
| 116 | + if got := streamMsgs(t, js); got != want { |
| 117 | + t.Fatalf("plan+apply flow: expected %d distinct messages, got %d", want, got) |
| 118 | + } |
| 119 | +} |
| 120 | + |
| 121 | +// Test_tfRunEventMsgID locks in the exact dedup-key shape so a refactor |
| 122 | +// can't accidentally widen or narrow the scope. |
| 123 | +func Test_tfRunEventMsgID(t *testing.T) { |
| 124 | + tests := []struct { |
| 125 | + runID, status, want string |
| 126 | + }{ |
| 127 | + {"run-rKys4r9a19W7Dk8Q", "planning", "run-rKys4r9a19W7Dk8Q:planning"}, |
| 128 | + {"run-rKys4r9a19W7Dk8Q", "applied", "run-rKys4r9a19W7Dk8Q:applied"}, |
| 129 | + {"run-OTHER", "planning", "run-OTHER:planning"}, |
| 130 | + } |
| 131 | + for _, tc := range tests { |
| 132 | + if got := tfRunEventMsgID(tc.runID, tc.status); got != tc.want { |
| 133 | + t.Errorf("tfRunEventMsgID(%q, %q) = %q, want %q", |
| 134 | + tc.runID, tc.status, got, tc.want) |
| 135 | + } |
| 136 | + } |
| 137 | +} |
| 138 | + |
| 139 | +// newTestStream wires up a Stream wired to the test JetStream context. We |
| 140 | +// skip NewStream() because it spins up a polling task dispatcher goroutine |
| 141 | +// that we don't need for these tests. |
| 142 | +// |
| 143 | +// JetStream uses on-disk storage by default and natstest reuses os.TempDir() |
| 144 | +// across instances, so the stream can survive a NATS server shutdown. Purge |
| 145 | +// it on entry so each test starts at zero messages and zero dedup history. |
| 146 | +func newTestStream(t *testing.T, js nats.JetStreamContext) *Stream { |
| 147 | + t.Helper() |
| 148 | + configureTFRunEventsStream(js, testDedupWindow) |
| 149 | + if err := js.PurgeStream(RunEventsStreamName); err != nil { |
| 150 | + t.Fatalf("could not purge %s before test: %v", RunEventsStreamName, err) |
| 151 | + } |
| 152 | + kv, err := configureTFRunMetadataKVStore(js) |
| 153 | + if err != nil { |
| 154 | + t.Fatalf("could not configure metadata KV: %v", err) |
| 155 | + } |
| 156 | + return &Stream{js: js, metadataKV: kv} |
| 157 | +} |
| 158 | + |
| 159 | +func seedRunMetadata(t *testing.T, s *Stream, runID, vcs string) { |
| 160 | + t.Helper() |
| 161 | + // KV is persistent across test runs (file storage under os.TempDir). |
| 162 | + // Clear any prior entry so AddRunMeta's Create() doesn't fail with |
| 163 | + // "key exists" when tests are re-run. |
| 164 | + _ = s.metadataKV.Delete(runID) |
| 165 | + rmd := &TFRunMetadata{ |
| 166 | + RunID: runID, |
| 167 | + Organization: "zapier", |
| 168 | + Workspace: "test-ws", |
| 169 | + Action: "plan", |
| 170 | + CommitSHA: "deadbeef", |
| 171 | + MergeRequestProjectNameWithNamespace: "zapier/terraform-services", |
| 172 | + MergeRequestIID: 1, |
| 173 | + VcsProvider: vcs, |
| 174 | + } |
| 175 | + if err := s.AddRunMeta(rmd); err != nil { |
| 176 | + t.Fatalf("could not seed RunMetadata for %q: %v", runID, err) |
| 177 | + } |
| 178 | +} |
| 179 | + |
| 180 | +func publishOnce(t *testing.T, s *Stream, runID, status string) { |
| 181 | + t.Helper() |
| 182 | + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 183 | + defer cancel() |
| 184 | + re := &TFRunEvent{ |
| 185 | + RunID: runID, |
| 186 | + Organization: "zapier", |
| 187 | + Workspace: "test-ws", |
| 188 | + NewStatus: status, |
| 189 | + } |
| 190 | + if err := s.PublishTFRunEvent(ctx, re); err != nil { |
| 191 | + t.Fatalf("PublishTFRunEvent(%q, %q) error: %v", runID, status, err) |
| 192 | + } |
| 193 | +} |
| 194 | + |
| 195 | +func streamMsgs(t *testing.T, js nats.JetStreamContext) uint64 { |
| 196 | + t.Helper() |
| 197 | + info, err := js.StreamInfo(RunEventsStreamName) |
| 198 | + if err != nil { |
| 199 | + t.Fatalf("could not read %s info: %v", RunEventsStreamName, err) |
| 200 | + } |
| 201 | + return info.State.Msgs |
| 202 | +} |
0 commit comments