proxy/internal/app/progressive_detection_test.go

package app_test

import (
	"context"
	"errors"
	"fmt"
	"path/filepath"
	"reflect"
	"sync"
	"testing"

	"proxyctl/internal/app"
	"proxyctl/internal/profile"
	"proxyctl/internal/target"
)

type controlledDetectionTarget struct {
	id          string
	description target.Description
	release     <-chan struct{}
	detection   target.Detection
}

func (t *controlledDetectionTarget) ID() string                      { return t.id }
func (t *controlledDetectionTarget) Description() target.Description { return t.description }
func (t *controlledDetectionTarget) Availability(context.Context) target.Availability {
	panic("progressive detection must use the structured Detector seam")
}
func (t *controlledDetectionTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
	panic("progressive detection must use the structured Detector seam")
}
func (t *controlledDetectionTarget) Confirmation(bool) string { return "" }
func (t *controlledDetectionTarget) SetDesired(context.Context, bool, *profile.Profile) target.OperationResult {
	return target.OperationResult{}
}
func (t *controlledDetectionTarget) Detect(context.Context, *profile.Profile) target.Detection {
	<-t.release
	return t.detection
}

func TestApplicationStreamsIndependentTargetDetectionInProductOrder(t *testing.T) {
	firstRelease := make(chan struct{})
	secondRelease := make(chan struct{})
	failedRelease := make(chan struct{})
	adapters := []target.Target{
		&controlledDetectionTarget{id: "first", description: target.Description{Name: "First", Summary: "first description"}, release: firstRelease, detection: target.Detection{Availability: target.Availability{Available: true, Evidence: "first available"}, Inspection: target.Inspection{State: target.Enabled, Evidence: "first enabled"}}},
		&controlledDetectionTarget{id: "second", description: target.Description{Name: "Second", Summary: "second description"}, release: secondRelease, detection: target.Detection{Availability: target.Availability{Available: true, Evidence: "second available"}, Inspection: target.Inspection{State: target.External, Evidence: "second external"}}},
		&controlledDetectionTarget{id: "failed", description: target.Description{Name: "Failed", Summary: "failed description"}, release: failedRelease, detection: target.Detection{Err: errors.New("inspection command failed")}},
	}
	application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), adapters)
	if err != nil {
		t.Fatal(err)
	}

	before := application.Snapshot()
	assertTargetOrder(t, before, []string{"first", "second", "failed"})
	for _, item := range before.Targets {
		if item.DetectionPhase != app.TargetChecking || item.Description.Name == "" || item.Description.Summary == "" {
			t.Fatalf("Target identity was not useful before detection: %#v", item)
		}
	}

	stream := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
	checking := stream.Snapshot
	for _, item := range checking.Targets {
		if item.DetectionPhase != app.TargetChecking {
			t.Fatalf("Target did not begin in Checking: %#v", item)
		}
	}
	updates := stream.Updates
	initial := <-updates
	if !initial.IsRunning() || len(initial.Progress()) != 3 {
		t.Fatalf("initial detection progress = %#v", initial)
	}

	close(secondRelease)
	second := <-updates
	if second.Status().TargetID != "second" || second.Snapshot.Targets[1].DetectionPhase != app.TargetReady || second.Snapshot.Targets[1].State != target.External {
		t.Fatalf("second completion = %#v", second)
	}
	assertOnlyDetectionChanged(t, checking, second.Snapshot, "second")
	assertTargetOrder(t, second.Snapshot, []string{"first", "second", "failed"})

	close(failedRelease)
	failed := <-updates
	if failed.Status().TargetID != "failed" || failed.Snapshot.Targets[2].DetectionPhase != app.TargetDetectionFailed || failed.Snapshot.Targets[2].DetectionErr == nil {
		t.Fatalf("failed completion = %#v", failed)
	}
	assertOnlyDetectionChanged(t, second.Snapshot, failed.Snapshot, "failed")

	close(firstRelease)
	first := <-updates
	if first.Status().TargetID != "first" || first.Snapshot.Targets[0].DetectionPhase != app.TargetReady || first.Snapshot.Targets[0].State != target.Enabled {
		t.Fatalf("first completion = %#v", first)
	}
	assertOnlyDetectionChanged(t, failed.Snapshot, first.Snapshot, "first")

	final := <-updates
	if !final.IsFailed() {
		t.Fatalf("final update = %#v", final)
	}
	assertTargetOrder(t, final.Snapshot, []string{"first", "second", "failed"})
	if final.Snapshot.Targets[0].State != target.Enabled || final.Snapshot.Targets[1].State != target.External || final.Snapshot.Targets[2].DetectionPhase != app.TargetDetectionFailed {
		t.Fatalf("final snapshot = %#v", final.Snapshot.Targets)
	}
}

type detectionRequest struct {
	result chan target.Detection
}

type overlappingDetectionTarget struct {
	requests chan detectionRequest
}

func (t *overlappingDetectionTarget) ID() string { return "overlap" }
func (t *overlappingDetectionTarget) Description() target.Description {
	return target.Description{Name: "Overlap", Summary: "controlled overlapping detection"}
}
func (t *overlappingDetectionTarget) Availability(context.Context) target.Availability {
	return target.Availability{Available: true}
}
func (t *overlappingDetectionTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
	return target.Inspection{State: target.Disabled}
}
func (t *overlappingDetectionTarget) Confirmation(bool) string { return "" }
func (t *overlappingDetectionTarget) SetDesired(context.Context, bool, *profile.Profile) target.OperationResult {
	return target.OperationResult{}
}
func (t *overlappingDetectionTarget) Detect(context.Context, *profile.Profile) target.Detection {
	request := detectionRequest{result: make(chan target.Detection, 1)}
	t.requests <- request
	return <-request.result
}

func TestOverlappingTargetDetectionIsRejectedAtTheSerializedOperationEntry(t *testing.T) {
	adapter := &overlappingDetectionTarget{requests: make(chan detectionRequest, 1)}
	application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
	if err != nil {
		t.Fatal(err)
	}

	older := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
	request := <-adapter.requests
	newer := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
	closed := <-newer.Updates
	if closed.Status().Kind != app.StatusBlocked {
		t.Fatalf("overlapping detection = %#v", closed)
	}
	request.result <- target.Detection{
		Availability: target.Availability{Available: true, Evidence: "authoritative availability"},
		Inspection:   target.Inspection{State: target.Enabled, Evidence: "authoritative inspection"},
	}
	for range older.Updates {
	}
	got := application.Snapshot().Targets[0]
	if !got.Available || got.State != target.Enabled {
		t.Fatalf("serialized detection result = %#v", got)
	}
}

type mutationInteractionTarget struct {
	mu      sync.Mutex
	enabled bool
	started chan struct{}
	release chan struct{}
}

func (t *mutationInteractionTarget) ID() string { return "mutation" }
func (t *mutationInteractionTarget) Description() target.Description {
	return target.Description{Name: "Mutation", Summary: "detection and mutation interaction"}
}
func (t *mutationInteractionTarget) Availability(context.Context) target.Availability {
	return target.Availability{Available: true}
}
func (t *mutationInteractionTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
	t.mu.Lock()
	defer t.mu.Unlock()
	state := target.Disabled
	if t.enabled {
		state = target.Enabled
	}
	return target.Inspection{State: state, Evidence: "current mutation state"}
}
func (t *mutationInteractionTarget) Confirmation(bool) string { return "" }
func (t *mutationInteractionTarget) SetDesired(context.Context, bool, *profile.Profile) target.OperationResult {
	t.mu.Lock()
	t.enabled = true
	t.mu.Unlock()
	return target.OperationResult{Inspection: target.Inspection{State: target.Enabled, Evidence: "mutation completed"}}
}
func (t *mutationInteractionTarget) Detect(context.Context, *profile.Profile) target.Detection {
	select {
	case <-t.started:
	default:
		close(t.started)
	}
	<-t.release
	return target.Detection{
		Availability: target.Availability{Available: true, Evidence: "stale detection"},
		Inspection:   target.Inspection{State: target.Disabled, Evidence: "stale detection"},
	}
}

func TestDispatchMutationWaitsForInFlightTargetDetection(t *testing.T) {
	adapter := &mutationInteractionTarget{started: make(chan struct{}), release: make(chan struct{})}
	application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
	if err != nil {
		t.Fatal(err)
	}
	created := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}})
	if created.Err != nil {
		t.Fatal(created.Err)
	}
	if activated := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); activated.Err != nil {
		t.Fatal(activated.Err)
	}

	detection := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
	<-adapter.started
	mutationDone := make(chan app.Result, 1)
	go func() {
		mutationDone <- application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
	}()
	close(adapter.release)
	for range detection.Updates {
	}
	result := <-mutationDone
	if result.Err != nil || result.Snapshot.Targets[0].State != target.Enabled {
		t.Fatalf("mutation result = %#v", result)
	}
	if got := application.Snapshot().Targets[0]; got.State != target.Enabled || got.Evidence != "mutation completed" {
		t.Fatalf("serialized mutation state: %#v", got)
	}
}

func TestSnapshotIsSafeDuringConcurrentStateMutation(t *testing.T) {
	application, err := app.New(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")))
	if err != nil {
		t.Fatal(err)
	}

	const count = 40
	start := make(chan struct{})
	var readers sync.WaitGroup
	for range 4 {
		readers.Add(1)
		go func() {
			defer readers.Done()
			<-start
			for range count * 4 {
				_ = application.Snapshot()
			}
		}()
	}
	close(start)
	for i := range count {
		result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{
			Name: fmt.Sprintf("profile-%02d", i), SingleURL: "http://proxy.example:8080",
		}})
		if result.Err != nil {
			t.Fatal(result.Err)
		}
	}
	readers.Wait()
	if got := len(application.Snapshot().Profiles); got != count {
		t.Fatalf("Profile count = %d, want %d", got, count)
	}
}

func assertTargetOrder(t *testing.T, snapshot app.Snapshot, want []string) {
	t.Helper()
	got := make([]string, len(snapshot.Targets))
	for i := range snapshot.Targets {
		got[i] = snapshot.Targets[i].ID
	}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("Target order = %#v, want %#v", got, want)
	}
}

func assertOnlyDetectionChanged(t *testing.T, before, after app.Snapshot, changedID string) {
	t.Helper()
	if len(before.Targets) != len(after.Targets) {
		t.Fatalf("Target count changed from %d to %d", len(before.Targets), len(after.Targets))
	}
	for i := range before.Targets {
		if before.Targets[i].ID == changedID {
			if reflect.DeepEqual(before.Targets[i], after.Targets[i]) {
				t.Fatalf("completed Target %q did not change", changedID)
			}
			continue
		}
		if !reflect.DeepEqual(before.Targets[i], after.Targets[i]) {
			t.Fatalf("completion for %q changed unrelated Target %q: before=%#v after=%#v", changedID, before.Targets[i].ID, before.Targets[i], after.Targets[i])
		}
	}
}