322 lines
12 KiB
Go
322 lines
12 KiB
Go
package app_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"proxyctl/internal/app"
|
|
"proxyctl/internal/profile"
|
|
"proxyctl/internal/target"
|
|
)
|
|
|
|
type coordinatedOperationTarget struct {
|
|
id string
|
|
state target.State
|
|
started chan struct{}
|
|
release chan struct{}
|
|
detectionStarted chan struct{}
|
|
detectionRelease chan struct{}
|
|
confirmation string
|
|
failure error
|
|
mu sync.Mutex
|
|
inDetection bool
|
|
overlapObserved chan bool
|
|
}
|
|
|
|
func (t *coordinatedOperationTarget) ID() string { return t.id }
|
|
func (t *coordinatedOperationTarget) Description() target.Description {
|
|
return target.Description{Name: t.id, Summary: "operation stream test Target"}
|
|
}
|
|
func (t *coordinatedOperationTarget) Availability(context.Context) target.Availability {
|
|
return target.Availability{Available: true, Evidence: "available"}
|
|
}
|
|
func (t *coordinatedOperationTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
|
|
return target.Inspection{State: t.state, Evidence: "fresh state"}
|
|
}
|
|
func (t *coordinatedOperationTarget) Confirmation(bool) string { return t.confirmation }
|
|
func (t *coordinatedOperationTarget) Detect(ctx context.Context, active *profile.Profile) target.Detection {
|
|
t.mu.Lock()
|
|
t.inDetection = true
|
|
t.mu.Unlock()
|
|
if t.detectionStarted != nil {
|
|
select {
|
|
case <-t.detectionStarted:
|
|
default:
|
|
close(t.detectionStarted)
|
|
}
|
|
}
|
|
if t.detectionRelease != nil {
|
|
<-t.detectionRelease
|
|
}
|
|
t.mu.Lock()
|
|
t.inDetection = false
|
|
t.mu.Unlock()
|
|
return target.Detection{Availability: t.Availability(ctx), Inspection: t.Inspect(ctx, active)}
|
|
}
|
|
func (t *coordinatedOperationTarget) SetDesired(_ context.Context, enabled bool, _ *profile.Profile) target.OperationResult {
|
|
t.mu.Lock()
|
|
overlapped := t.inDetection
|
|
t.mu.Unlock()
|
|
if t.overlapObserved != nil {
|
|
t.overlapObserved <- overlapped
|
|
}
|
|
if t.started != nil {
|
|
close(t.started)
|
|
}
|
|
if t.release != nil {
|
|
<-t.release
|
|
}
|
|
if t.failure != nil {
|
|
t.state = target.Drifted
|
|
return target.OperationResult{Err: t.failure}
|
|
}
|
|
if enabled {
|
|
t.state = target.Enabled
|
|
} else {
|
|
t.state = target.Disabled
|
|
}
|
|
return target.OperationResult{Inspection: target.Inspection{State: t.state, Evidence: "completed"}}
|
|
}
|
|
|
|
type cancellableOperationTarget struct{ started chan struct{} }
|
|
|
|
func (t *cancellableOperationTarget) ID() string { return "cancellable" }
|
|
func (t *cancellableOperationTarget) Description() target.Description {
|
|
return target.Description{Name: "Cancellable"}
|
|
}
|
|
func (t *cancellableOperationTarget) Availability(context.Context) target.Availability {
|
|
return target.Availability{Available: true}
|
|
}
|
|
func (t *cancellableOperationTarget) Inspect(context.Context, *profile.Profile) target.Inspection {
|
|
return target.Inspection{State: target.Disabled, Evidence: "not changed"}
|
|
}
|
|
func (t *cancellableOperationTarget) Confirmation(bool) string { return "" }
|
|
func (t *cancellableOperationTarget) SetDesired(ctx context.Context, _ bool, _ *profile.Profile) target.OperationResult {
|
|
close(t.started)
|
|
<-ctx.Done()
|
|
return target.OperationResult{Inspection: target.Inspection{State: target.Disabled}, Err: ctx.Err()}
|
|
}
|
|
|
|
func TestInteractiveOperationCancellationReachesTarget(t *testing.T) {
|
|
adapter := &cancellableOperationTarget{started: make(chan struct{})}
|
|
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
createAndActivate(t, application)
|
|
refreshApplicationTargets(t, application)
|
|
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
|
|
<-adapter.started
|
|
operation.Cancel()
|
|
var final app.OperationUpdate
|
|
for update := range operation.Updates {
|
|
final = update
|
|
}
|
|
if !final.IsFailed() || !errors.Is(final.Err(), context.Canceled) {
|
|
t.Fatalf("cancelled operation final update = %#v", final)
|
|
}
|
|
}
|
|
|
|
func TestInteractiveOperationStreamSerializesMutationAndPublishesStructuredProgress(t *testing.T) {
|
|
adapter := &coordinatedOperationTarget{id: "shell", state: target.Disabled, 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)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
refreshApplicationTargets(t, application)
|
|
|
|
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
|
|
running := <-operation.Updates
|
|
runningProgress := running.Progress()
|
|
if !running.IsRunning() || len(runningProgress) != 1 || runningProgress[0].Phase != app.TargetProgressRunning || runningProgress[0].TargetID != "shell" || running.Status().Kind != app.StatusProgress {
|
|
t.Fatalf("running update = %#v", running)
|
|
}
|
|
<-adapter.started
|
|
|
|
blocked := application.StartOperation(app.Intent{Kind: app.DisableTarget, TargetID: "shell"})
|
|
blockedUpdate := <-blocked.Updates
|
|
if !blockedUpdate.IsTerminal() || blockedUpdate.Status().Kind != app.StatusBlocked || !strings.Contains(blockedUpdate.Status().Summary, "already in progress") {
|
|
t.Fatalf("blocked update = %#v", blockedUpdate)
|
|
}
|
|
if _, open := <-blocked.Updates; open {
|
|
t.Fatal("blocked operation stream did not close")
|
|
}
|
|
|
|
close(adapter.release)
|
|
final := <-operation.Updates
|
|
finalProgress := final.Progress()
|
|
if !final.IsSucceeded() || final.Snapshot.Targets[0].State != target.Enabled || len(finalProgress) != 1 || finalProgress[0].Phase != app.TargetProgressSucceeded {
|
|
t.Fatalf("final update = %#v", final)
|
|
}
|
|
if _, open := <-operation.Updates; open {
|
|
t.Fatal("completed operation stream did not close")
|
|
}
|
|
}
|
|
|
|
func TestDetectionOperationSerializesWithSynchronousMutation(t *testing.T) {
|
|
adapter := &coordinatedOperationTarget{
|
|
id: "shell", state: target.Disabled,
|
|
detectionStarted: make(chan struct{}), detectionRelease: make(chan struct{}),
|
|
started: make(chan struct{}), overlapObserved: make(chan bool, 1),
|
|
}
|
|
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
|
|
detection := application.StartOperation(app.Intent{Kind: app.RefreshTargets})
|
|
<-adapter.detectionStarted
|
|
mutationAttempted := make(chan struct{})
|
|
mutationDone := make(chan app.Result, 1)
|
|
go func() {
|
|
close(mutationAttempted)
|
|
mutationDone <- application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
|
|
}()
|
|
<-mutationAttempted
|
|
close(adapter.detectionRelease)
|
|
for range detection.Updates {
|
|
}
|
|
result := <-mutationDone
|
|
if result.Err != nil || result.Snapshot.Targets[0].State != target.Enabled {
|
|
t.Fatalf("serialized mutation = %#v", result)
|
|
}
|
|
if overlapped := <-adapter.overlapObserved; overlapped {
|
|
t.Fatal("mutation overlapped Target detection")
|
|
}
|
|
}
|
|
|
|
func TestStartOperationReturnsBeforeDispatchSerializationIsAvailable(t *testing.T) {
|
|
adapter := &coordinatedOperationTarget{
|
|
id: "shell", state: target.Disabled,
|
|
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)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
refreshApplicationTargets(t, application)
|
|
|
|
mutationDone := make(chan app.Result, 1)
|
|
go func() {
|
|
mutationDone <- application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: "shell"})
|
|
}()
|
|
<-adapter.started // barrier: synchronous Dispatch owns dispatch serialization
|
|
adapter.detectionStarted = make(chan struct{})
|
|
adapter.detectionRelease = make(chan struct{})
|
|
|
|
returned := make(chan app.OperationStream, 1)
|
|
go func() { returned <- application.StartOperation(app.Intent{Kind: app.RefreshTargets}) }()
|
|
var stream app.OperationStream
|
|
select {
|
|
case stream = <-returned:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("StartOperation blocked Bubble Tea while dispatch serialization was held")
|
|
}
|
|
|
|
close(adapter.release)
|
|
if result := <-mutationDone; result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
<-adapter.detectionStarted
|
|
close(adapter.detectionRelease)
|
|
for range stream.Updates {
|
|
}
|
|
}
|
|
|
|
func TestInteractiveOperationStreamUsesStructuredConfirmation(t *testing.T) {
|
|
adapter := &coordinatedOperationTarget{id: "docker-daemon", state: target.Disabled, confirmation: "Docker will restart and may disrupt running work."}
|
|
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "office", SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "office"}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
|
|
refreshApplicationTargets(t, application)
|
|
operation := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID()})
|
|
update := <-operation.Updates
|
|
confirmation := update.Confirmation()
|
|
if !update.IsConfirmation() || confirmation == nil || confirmation.Title == "" || confirmation.CancelLabel == "" || confirmation.Token == "" {
|
|
t.Fatalf("confirmation update = %#v", update)
|
|
}
|
|
// Retry immediately after the terminal confirmation, before observing the
|
|
// first stream close. The Application gate must already be released.
|
|
retry := application.StartOperation(app.Intent{Kind: app.EnableTarget, TargetID: adapter.ID(), Confirm: true, ConfirmationToken: confirmation.Token})
|
|
var final app.OperationUpdate
|
|
for next := range retry.Updates {
|
|
if next.Decode().Kind == app.OperationClosed {
|
|
t.Fatalf("immediate confirmed retry observed stale operation gate: %#v", next)
|
|
}
|
|
final = next
|
|
}
|
|
if !final.IsSucceeded() {
|
|
t.Fatalf("confirmed retry final update = %#v", final)
|
|
}
|
|
if _, open := <-operation.Updates; open {
|
|
t.Fatal("confirmation operation stream did not close")
|
|
}
|
|
}
|
|
|
|
func TestInteractiveProfileTransitionReportsFirstFailureWithoutRollback(t *testing.T) {
|
|
first := &coordinatedOperationTarget{id: "first", state: target.Disabled}
|
|
second := &coordinatedOperationTarget{id: "second", state: target.Disabled}
|
|
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{first, second})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, name := range []string{"old", "new"} {
|
|
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: name, SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
}
|
|
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "old"}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
refreshApplicationTargets(t, application)
|
|
for _, id := range []string{"first", "second"} {
|
|
if result := application.Dispatch(app.Intent{Kind: app.EnableTarget, TargetID: id}); result.Err != nil {
|
|
t.Fatal(result.Err)
|
|
}
|
|
}
|
|
second.failure = errors.New("second failed")
|
|
|
|
operation := application.StartOperation(app.Intent{Kind: app.ActivateProfile, Name: "new"})
|
|
var final app.OperationUpdate
|
|
for update := range operation.Updates {
|
|
final = update
|
|
}
|
|
if !final.IsFailed() || final.FirstFailedTargetID() != "second" || final.Snapshot.ActiveProfile != "new" {
|
|
t.Fatalf("final transition = %#v", final)
|
|
}
|
|
if final.Snapshot.Targets[0].State != target.Enabled || final.Snapshot.Targets[1].State != target.Drifted {
|
|
t.Fatalf("partial Target states = %#v", final.Snapshot.Targets)
|
|
}
|
|
}
|