Janady13/missionmatch / internal/integration/pipeline_integration_test.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/integration/pipeline_integration_test.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/integration/pipeline_integration_test.go
Original SHA-256	bc0adbe4d8b53d43684b161ef355755dc97e06d8435aa9504adeaa376c1fa73c
Git blob SHA	271ea4d24129633d945104b135da04344103f4f2
Reading-copy SHA-256	bc0adbe4d8b53d43684b161ef355755dc97e06d8435aa9504adeaa376c1fa73c
Original source ↗
Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/integration/pipeline_integration_test.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/integration/pipeline_integration_test.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Recovered source transcript

Download complete source text ↓406 lines · 14.5 KB
//go:build integration

package integration

import (
	"context"
	"errors"
	"sync"
	"testing"

	"github.com/google/uuid"

	"github.com/missionworkforce/missionmatch/internal/modules/recruiting"
)

// WARNING. These tests create organizations, users, candidates and job orders. Point MM_APP_DSN and
// MM_MIGRATOR_DSN at a scratch or staging database, never at production.

func seedPipelineCandidate(t *testing.T, org uuid.UUID, name string) uuid.UUID {
	t.Helper()
	var id uuid.UUID
	if err := migrator.QueryRow(context.Background(),
		`INSERT INTO candidates (organization_id, full_name, clearance_level, clearance_poly, clearance_status)
		 VALUES ($1,$2,'none','none','unknown') RETURNING id`, org, name).Scan(&id); err != nil {
		t.Fatalf("seed candidate: %v", err)
	}
	return id
}

// newPipeline returns a service, a context, a requisition and n candidates.
func newPipeline(t *testing.T, seats int, n int) (*recruiting.Service, context.Context, uuid.UUID, []uuid.UUID) {
	t.Helper()
	org := seedOrg(t)
	_, ctx := seedRecruiter(t, org, "Pipeline Recruiter")
	svc := reqSvc()
	r, err := svc.CreateRequisition(ctx, recruiting.CreateRequisitionInput{Title: "Pipeline Role", Openings: seats})
	if err != nil {
		t.Fatalf("create requisition: %v", err)
	}
	cands := make([]uuid.UUID, n)
	for i := range cands {
		cands[i] = seedPipelineCandidate(t, org, "Cand "+string(rune('A'+i)))
	}
	return svc, ctx, uuid.MustParse(r.ID), cands
}

// TestPipelineWalkRecordsEveryStep walks a candidate up the ladder and checks the row, the history and
// the high water mark agree at every step.
func TestPipelineWalkRecordsEveryStep(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 1)

	a, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{
		CandidateID: cands[0].String(), SourceTouch: "referral"})
	if err != nil {
		t.Fatalf("submit: %v", err)
	}
	if a.Stage != "sourced" || a.FurthestStage != "sourced" {
		t.Fatalf("new application at %q (furthest %q), want sourced", a.Stage, a.FurthestStage)
	}
	id := uuid.MustParse(a.ID)

	for _, step := range []string{"screened", "submitted", "interviewing", "offered", "placed"} {
		prev := a.Stage
		a, err = svc.MoveStage(ctx, id, recruiting.MoveStageInput{ExpectedStage: prev, ToStage: step})
		if err != nil {
			t.Fatalf("%s to %s: %v", prev, step, err)
		}
		if a.Stage != step || a.FurthestStage != step {
			t.Errorf("after moving to %s the row says %q and the mark says %q", step, a.Stage, a.FurthestStage)
		}
		if a.DaysInStage < 0 {
			t.Errorf("negative time in stage at %s", step)
		}
	}

	d, err := svc.GetApplication(ctx, id)
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	// One opening event plus five moves.
	if len(d.Events) != 6 {
		t.Fatalf("history has %d events, want 6", len(d.Events))
	}
	if d.Events[0].FromStage != "" {
		t.Errorf("the opening event has predecessor %q, want none", d.Events[0].FromStage)
	}
	// Every later event must chain: its from_stage is the previous event's to_stage. This is what makes
	// the history a chain rather than a bag of claims.
	for i := 1; i < len(d.Events); i++ {
		if d.Events[i].FromStage != d.Events[i-1].ToStage {
			t.Errorf("event %d came from %q but the previous event ended at %q",
				i, d.Events[i].FromStage, d.Events[i-1].ToStage)
		}
		if d.Events[i].DaysInFrom < 0 {
			t.Errorf("event %d spent negative time in its previous stage", i)
		}
	}
	if d.Events[len(d.Events)-1].ActorName != "Pipeline Recruiter" {
		t.Errorf("actor name = %q, want the recruiter recorded at write time", d.Events[len(d.Events)-1].ActorName)
	}
}

// TestStaleExpectedStageIsRefused. Two recruiters looking at the same board must not both move the same
// candidate. The expected stage is the token, and the database checks it under a row lock.
func TestStaleExpectedStageIsRefused(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 1)
	a, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: cands[0].String()})
	if err != nil {
		t.Fatalf("submit: %v", err)
	}
	id := uuid.MustParse(a.ID)

	if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "screened"}); err != nil {
		t.Fatalf("first move: %v", err)
	}
	// The second caller still believes the application is at sourced.
	_, err = svc.MoveStage(ctx, id, recruiting.MoveStageInput{ExpectedStage: "sourced", ToStage: "submitted"})
	if !errors.Is(err, recruiting.ErrConflict) {
		t.Fatalf("a stale move returned %v, want a conflict", err)
	}
}

// TestHighWaterMarkNeverFalls. A candidate rejected at interview must still count as having reached
// interview, or the funnel would shrink retroactively every time someone was dispositioned.
func TestHighWaterMarkNeverFalls(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 1)
	a, _ := svc.Submit(ctx, req, recruiting.CreateApplicationInput{
		CandidateID: cands[0].String(), Stage: "interviewing"})
	id := uuid.MustParse(a.ID)
	if a.FurthestStage != "interviewing" {
		t.Fatalf("submitting straight to interviewing left the mark at %q", a.FurthestStage)
	}

	a, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "interviewing", ToStage: "rejected",
		DispositionCode: "interview_performance", Reason: "Did not pass the panel"})
	if err != nil {
		t.Fatalf("reject: %v", err)
	}
	if a.Stage != "rejected" {
		t.Fatalf("stage = %q, want rejected", a.Stage)
	}
	if a.FurthestStage != "interviewing" {
		t.Errorf("the mark fell to %q on rejection, want it held at interviewing", a.FurthestStage)
	}
	if a.DispositionCode != "interview_performance" {
		t.Errorf("disposition code = %q", a.DispositionCode)
	}

	// Moving backwards down the ladder must also not lower the mark.
	a, err = svc.MoveStage(ctx, id, recruiting.MoveStageInput{ExpectedStage: "rejected", ToStage: "screened"})
	if err != nil {
		t.Fatalf("reactivate: %v", err)
	}
	if a.FurthestStage != "interviewing" {
		t.Errorf("reactivating at screened lowered the mark to %q", a.FurthestStage)
	}
	if a.DispositionCode != "" {
		t.Errorf("reactivating left a stale disposition code %q", a.DispositionCode)
	}
}

// TestTerminalStageDemandsAValidReason. The coded reason is what makes an adverse impact report
// computable, so it cannot be optional and it cannot be from the wrong outcome.
func TestTerminalStageDemandsAValidReason(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 1)
	a, _ := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: cands[0].String()})
	id := uuid.MustParse(a.ID)

	if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "rejected"}); err == nil {
		t.Error("a rejection with no reason code was accepted")
	}
	if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "rejected", DispositionCode: "accepted_other_offer"}); err == nil {
		t.Error("a candidate withdrawal reason was accepted as an employer rejection")
	}
	if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "screened", DispositionCode: "not_qualified"}); err == nil {
		t.Error("a reason code was accepted on a non terminal stage")
	}
	if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "rejected", DispositionCode: "clearance_gap"}); err != nil {
		t.Errorf("a valid rejection was refused: %v", err)
	}
}

// TestPlacementsCannotExceedOpenings. A role with two seats cannot place three people, and the refusal
// must survive the last seat being taken concurrently.
func TestPlacementsCannotExceedOpenings(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 2, 3)
	ids := make([]uuid.UUID, 3)
	for i, c := range cands {
		a, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: c.String()})
		if err != nil {
			t.Fatalf("submit %d: %v", i, err)
		}
		ids[i] = uuid.MustParse(a.ID)
	}
	for i := 0; i < 2; i++ {
		if _, err := svc.MoveStage(ctx, ids[i], recruiting.MoveStageInput{
			ExpectedStage: "sourced", ToStage: "placed"}); err != nil {
			t.Fatalf("place %d: %v", i, err)
		}
	}
	if _, err := svc.MoveStage(ctx, ids[2], recruiting.MoveStageInput{
		ExpectedStage: "sourced", ToStage: "placed"}); err == nil {
		t.Fatal("a third candidate was placed into a role with two seats")
	}

	b, err := svc.Board(ctx, req)
	if err != nil {
		t.Fatalf("board: %v", err)
	}
	if b.Placed != 2 || b.SeatsLeft != 0 {
		t.Errorf("board says placed=%d seats_left=%d, want 2 and 0", b.Placed, b.SeatsLeft)
	}
}

// TestConcurrentPlacementsRespectTheLastSeat. Two recruiters placing different candidates into the last
// remaining seat at the same instant: exactly one may win.
func TestConcurrentPlacementsRespectTheLastSeat(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 8)
	ids := make([]uuid.UUID, len(cands))
	for i, c := range cands {
		a, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: c.String()})
		if err != nil {
			t.Fatalf("submit: %v", err)
		}
		ids[i] = uuid.MustParse(a.ID)
	}

	var wg sync.WaitGroup
	errs := make([]error, len(ids))
	start := make(chan struct{})
	for i, id := range ids {
		wg.Add(1)
		go func(i int, id uuid.UUID) {
			defer wg.Done()
			<-start
			_, errs[i] = svc.MoveStage(ctx, id, recruiting.MoveStageInput{
				ExpectedStage: "sourced", ToStage: "placed"})
		}(i, id)
	}
	close(start)
	wg.Wait()

	won := 0
	for _, e := range errs {
		if e == nil {
			won++
		}
	}
	if won != 1 {
		t.Fatalf("%d of %d concurrent placements succeeded into one seat, want exactly 1", won, len(ids))
	}
	b, _ := svc.Board(ctx, req)
	if b.Placed != 1 {
		t.Errorf("board says %d placed, want 1", b.Placed)
	}
}

// TestFunnelIsMonotonicOnRealData. The pure test proves the arithmetic; this proves the query feeding it
// produces marks that satisfy the same property.
func TestFunnelIsMonotonicOnRealData(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 6, 6)
	targets := []string{"sourced", "screened", "submitted", "interviewing", "offered", "placed"}
	for i, c := range cands {
		a, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: c.String()})
		if err != nil {
			t.Fatalf("submit: %v", err)
		}
		id := uuid.MustParse(a.ID)
		cur := "sourced"
		for _, step := range targets[1 : i+1] {
			if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
				ExpectedStage: cur, ToStage: step}); err != nil {
				t.Fatalf("advance to %s: %v", step, err)
			}
			cur = step
		}
		// Dispose half of them, which must not change the funnel at all.
		if i%2 == 0 && cur != "placed" {
			if _, err := svc.MoveStage(ctx, id, recruiting.MoveStageInput{
				ExpectedStage: cur, ToStage: "rejected", DispositionCode: "not_qualified"}); err != nil {
				t.Fatalf("dispose: %v", err)
			}
		}
	}

	b, err := svc.Board(ctx, req)
	if err != nil {
		t.Fatalf("board: %v", err)
	}
	if len(b.Funnel) != 6 {
		t.Fatalf("funnel has %d rungs, want 6", len(b.Funnel))
	}
	for i := 0; i+1 < len(b.Funnel); i++ {
		if b.Funnel[i+1].Count > b.Funnel[i].Count {
			t.Errorf("funnel widened on real data: %s=%d then %s=%d",
				b.Funnel[i].Stage, b.Funnel[i].Count, b.Funnel[i+1].Stage, b.Funnel[i+1].Count)
		}
	}
	if b.Funnel[0].Count != 6 {
		t.Errorf("first rung counted %d of 6 applications", b.Funnel[0].Count)
	}
	if b.Total != 6 {
		t.Errorf("board total = %d, want 6", b.Total)
	}
	// Dispositions must be grouped by code, which is the shape the adverse impact report needs.
	if len(b.Dispositions) == 0 {
		t.Error("three applications were rejected but no disposition breakdown was produced")
	}
	// Every stage gets a column even when empty, so the board does not change shape as people move.
	if len(b.Columns) != 8 {
		t.Errorf("board has %d columns, want 8 (six rungs plus two outcomes)", len(b.Columns))
	}
}

// TestDuplicateSubmissionIsRefused. The same candidate cannot be on one job order twice, and the
// database is the authority because two recruiters can submit at the same instant.
func TestDuplicateSubmissionIsRefused(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 2, 1)
	if _, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: cands[0].String()}); err != nil {
		t.Fatalf("first submit: %v", err)
	}
	if _, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: cands[0].String()}); err == nil {
		t.Fatal("the same candidate was submitted to one job order twice")
	}

	// Concurrently, too.
	svc2, ctx2, req2, cands2 := newPipeline(t, 4, 1)
	var wg sync.WaitGroup
	errs := make([]error, 6)
	start := make(chan struct{})
	for i := 0; i < 6; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			<-start
			_, errs[i] = svc2.Submit(ctx2, req2, recruiting.CreateApplicationInput{CandidateID: cands2[0].String()})
		}(i)
	}
	close(start)
	wg.Wait()
	won := 0
	for _, e := range errs {
		if e == nil {
			won++
		}
	}
	if won != 1 {
		t.Errorf("%d of 6 concurrent duplicate submissions succeeded, want exactly 1", won)
	}
}

// TestCannotSubmitToADeadJobOrder. Recording work against a cancelled role inflates the funnel of an
// order nobody is filling.
func TestCannotSubmitToADeadJobOrder(t *testing.T) {
	svc, ctx, req, cands := newPipeline(t, 1, 2)
	r, err := svc.GetRequisition(ctx, req)
	if err != nil {
		t.Fatalf("get requisition: %v", err)
	}
	if _, err := svc.ChangeStatus(ctx, req, recruiting.ChangeStatusInput{
		ExpectedUpdatedAt: r.UpdatedAt, Status: "cancelled", Note: "client pulled the role"}); err != nil {
		t.Fatalf("cancel: %v", err)
	}
	if _, err := svc.Submit(ctx, req, recruiting.CreateApplicationInput{CandidateID: cands[0].String()}); err == nil {
		t.Fatal("a candidate was submitted to a cancelled job order")
	}
}

// TestPipelineTenantIsolation. One tenant must never see another's applications, with a control proving
// the query was not simply returning nothing.
func TestPipelineTenantIsolation(t *testing.T) {
	svcA, ctxA, reqA, candsA := newPipeline(t, 1, 1)
	a, err := svcA.Submit(ctxA, reqA, recruiting.CreateApplicationInput{CandidateID: candsA[0].String()})
	if err != nil {
		t.Fatalf("submit in A: %v", err)
	}
	_, ctxB, _, _ := newPipeline(t, 1, 1)

	if _, err := svcA.GetApplication(ctxB, uuid.MustParse(a.ID)); err == nil {
		t.Fatal("tenant B fetched tenant A's application")
	}
	if _, err := svcA.Board(ctxB, reqA); err == nil {
		t.Fatal("tenant B read tenant A's pipeline board")
	}
	listB, err := svcA.ListApplications(ctxB, "", "", 100, 0)
	if err != nil {
		t.Fatalf("list in B: %v", err)
	}
	for _, x := range listB.Applications {
		if x.ID == a.ID {
			t.Fatal("tenant A's application appeared in tenant B's list")
		}
	}
	// Control.
	listA, err := svcA.ListApplications(ctxA, "", "", 100, 0)
	if err != nil || listA.Total != 1 {
		t.Fatalf("tenant A sees %d of its own applications, want 1 (err=%v)", listA.Total, err)
	}
}