453 lines
16 KiB
Go
453 lines
16 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
"vdi-cloud/platform/internal/model"
|
|
"vdi-cloud/platform/internal/store"
|
|
"vdi-cloud/platform/pkg/proxmox"
|
|
)
|
|
|
|
type snapshotPVE struct {
|
|
mu sync.Mutex
|
|
config map[string]any
|
|
target map[string]any
|
|
items []proxmox.Snapshot
|
|
state, outcome string
|
|
pending, denyManage, denyRestore, unsupported bool
|
|
writes int
|
|
}
|
|
|
|
func copySnapshotConfig(c map[string]any) map[string]any {
|
|
o := map[string]any{}
|
|
for k, v := range c {
|
|
o[k] = v
|
|
}
|
|
return o
|
|
}
|
|
func fakeSnapshotPVE(t *testing.T, f *fixture) *snapshotPVE {
|
|
p := &snapshotPVE{state: "stopped", config: map[string]any{"digest": "d1", "cores": 2, "memory": 2048, "scsi0": "local:vm-9000,size=4G", "efidisk0": "local:vm-9000-efi,size=4M", "vmgenid": "old"}, items: []proxmox.Snapshot{{Name: "base", Time: 123, Description: "baseline"}}}
|
|
p.target = copySnapshotConfig(p.config)
|
|
p.target["cores"] = 1
|
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
path := strings.TrimPrefix(r.URL.Path, "/api2/json")
|
|
var data any
|
|
switch {
|
|
case strings.Contains(path, "/tasks/"):
|
|
if p.outcome == "poll-failure" {
|
|
w.WriteHeader(503)
|
|
return
|
|
}
|
|
status, exit := "stopped", "OK"
|
|
if p.outcome == "failed" {
|
|
exit = "TASK ERROR"
|
|
}
|
|
if p.outcome == "running" {
|
|
status = "running"
|
|
}
|
|
data = map[string]any{"status": status, "exitstatus": exit}
|
|
case path == "/access/permissions":
|
|
a, b := 1, 1
|
|
if p.denyManage {
|
|
a = 0
|
|
}
|
|
if p.denyRestore {
|
|
b = 0
|
|
}
|
|
data = map[string]any{r.URL.Query().Get("path"): map[string]int{"VM.Snapshot": a, "VM.Snapshot.Rollback": b}}
|
|
case strings.HasSuffix(path, "/status/current"):
|
|
data = map[string]any{"status": p.state}
|
|
case strings.HasSuffix(path, "/pending"):
|
|
data = []map[string]any{}
|
|
if p.pending {
|
|
data = []map[string]any{{"key": "memory", "pending": 4096}}
|
|
}
|
|
case strings.HasSuffix(path, "/feature"):
|
|
n := 1
|
|
if p.unsupported {
|
|
n = 0
|
|
}
|
|
data = map[string]int{"hasFeature": n}
|
|
case strings.HasSuffix(path, "/snapshot") && r.Method == "GET":
|
|
data = append([]proxmox.Snapshot{{Name: "current"}}, p.items...)
|
|
case strings.Contains(path, "/snapshot/") && strings.HasSuffix(path, "/config"):
|
|
data = p.target
|
|
case strings.HasSuffix(path, "/config") && r.Method == "GET":
|
|
data = p.config
|
|
case r.Method != "GET" && strings.Contains(path, "/snapshot"):
|
|
p.writes++
|
|
_ = r.ParseForm()
|
|
if p.outcome == "reject" {
|
|
w.WriteHeader(403)
|
|
return
|
|
}
|
|
if p.outcome == "ambiguous" {
|
|
w.WriteHeader(500)
|
|
return
|
|
}
|
|
switch {
|
|
case strings.HasSuffix(path, "/rollback"):
|
|
if r.Method != "POST" || r.Form.Get("start") != "0" || len(r.Form) != 1 {
|
|
t.Error("unsafe rollback", r.Method, r.Form)
|
|
}
|
|
p.config = copySnapshotConfig(p.target)
|
|
p.config["vmgenid"] = "new"
|
|
p.config["digest"] = "d2"
|
|
case r.Method == "DELETE":
|
|
if len(r.Form) != 0 {
|
|
t.Error("unexpected deletion flags", r.Form)
|
|
}
|
|
p.items = []proxmox.Snapshot{}
|
|
case r.Method == "POST":
|
|
if r.Form.Get("vmstate") != "0" || len(r.Form) != 3 {
|
|
t.Error("unsafe snapshot create", r.Form)
|
|
}
|
|
p.items = append(p.items, proxmox.Snapshot{Name: r.Form.Get("snapname"), Description: r.Form.Get("description"), Time: 456})
|
|
p.target = copySnapshotConfig(p.config)
|
|
default:
|
|
t.Error("unexpected snapshot write", r.Method, path)
|
|
}
|
|
data = "UPID:snapshot-test"
|
|
if p.outcome == "missing-upid" {
|
|
data = ""
|
|
}
|
|
default:
|
|
t.Error("unexpected snapshot request", r.Method, path)
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
|
|
}))
|
|
t.Cleanup(srv.Close)
|
|
f.server.PVE = &proxmox.Client{BaseURL: srv.URL, HTTP: srv.Client()}
|
|
if _, err := f.db.Exec("UPDATE desktops SET provider='proxmox',node='node-a',vmid=9000,power_state='stopped' WHERE id='desk_demo_windows'"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return p
|
|
}
|
|
|
|
const snapshotEndpoint = "/desktops/desk_demo_windows/snapshots"
|
|
|
|
func snapshotBody(p *snapshotPVE, action string) map[string]any {
|
|
b := map[string]any{"action": action, "name": "base", "digest": "d1"}
|
|
if action == "create" {
|
|
b["name"] = "new_point"
|
|
b["description"] = "before upgrade"
|
|
} else {
|
|
b["revision"] = snapshotRevision(p.items[0])
|
|
}
|
|
if action == "rollback" {
|
|
b["confirmVmid"] = 9000
|
|
}
|
|
return b
|
|
}
|
|
func queueSnapshotTest(t *testing.T, f *fixture, admin string, b map[string]any, key string) model.Task {
|
|
code, out := f.request("POST", snapshotEndpoint+"/actions", admin, b, key)
|
|
expect(t, 202, code, out)
|
|
var task model.Task
|
|
if err := f.db.Get(&task, "SELECT * FROM tasks WHERE id=$1", out["id"]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return task
|
|
}
|
|
func readSnapshotTask(t *testing.T, f *fixture, id string) model.Task {
|
|
var task model.Task
|
|
if err := f.db.Get(&task, "SELECT * FROM tasks WHERE id=$1", id); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return task
|
|
}
|
|
func TestSnapshotRealTaskLifecycle(t *testing.T) {
|
|
for _, action := range []string{"create", "delete", "rollback"} {
|
|
t.Run(action, func(t *testing.T) {
|
|
f := newFixtureWorker(t, false)
|
|
p := fakeSnapshotPVE(t, f)
|
|
admin := f.login("admin")
|
|
b := snapshotBody(p, action)
|
|
code, out := f.request("GET", snapshotEndpoint, f.login("viewer"), nil, "")
|
|
expect(t, 200, code, out)
|
|
if len(out["items"].([]any)) != 1 {
|
|
t.Fatal("synthetic current leaked", out)
|
|
}
|
|
for _, role := range []string{"operator", "viewer"} {
|
|
token := f.login(role)
|
|
code, out = f.request("POST", snapshotEndpoint+"/check", token, b, "")
|
|
expect(t, 403, code, out)
|
|
code, out = f.request("POST", snapshotEndpoint+"/actions", token, b, "denied")
|
|
expect(t, 403, code, out)
|
|
}
|
|
code, out = f.request("POST", snapshotEndpoint+"/check", admin, b, "")
|
|
expect(t, 200, code, out)
|
|
task := queueSnapshotTest(t, f, admin, b, "snapshot-once")
|
|
code, out = f.request("POST", snapshotEndpoint+"/actions", admin, b, "snapshot-once")
|
|
expect(t, 202, code, out)
|
|
if out["id"] != task.ID {
|
|
t.Fatal("duplicate task")
|
|
}
|
|
changed := copySnapshotConfig(b)
|
|
changed["name"] = "different"
|
|
code, out = f.request("POST", snapshotEndpoint+"/actions", admin, changed, "snapshot-once")
|
|
expect(t, 409, code, out)
|
|
code, out = f.request("POST", "/desktops/desk_demo_windows/actions", admin, map[string]string{"action": "start"}, "power")
|
|
expect(t, 409, code, out)
|
|
code, out = f.request("POST", "/desktops/desk_demo_windows/hardware", admin, map[string]any{"kind": "compute", "cores": 4, "sockets": 1, "memoryMiB": 4096, "digest": "d1"}, "hw")
|
|
expect(t, 409, code, out)
|
|
code, out = f.request("PUT", "/desktops/desk_demo_windows/connection", admin, map[string]any{"endpoint": "192.0.2.90", "width": 1920, "height": 1080, "fps": 30, "initialized": true}, "")
|
|
expect(t, 409, code, out)
|
|
if action == "rollback" {
|
|
_, err := f.db.Exec("INSERT INTO connection_tickets(token_hash,terminal_id,desktop_id,expires_at) VALUES('stale','term_demo_001','desk_demo_windows',now()+interval '1 minute')")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
f.server.execute(context.Background(), task)
|
|
task = readSnapshotTask(t, f, task.ID)
|
|
if task.Status != "success" || task.ProviderRef == "" {
|
|
t.Fatal(task)
|
|
}
|
|
var stage string
|
|
_ = f.db.Get(&stage, "SELECT stage FROM snapshot_jobs WHERE task_id=$1", task.ID)
|
|
if stage != "completed" || p.writes != 1 {
|
|
t.Fatal(stage, p.writes)
|
|
}
|
|
var n int
|
|
_ = f.db.Get(&n, "SELECT count(*) FROM audit_logs WHERE action=$1", task.Action)
|
|
if n != 2 {
|
|
t.Fatal("missing submit or completion audit", n)
|
|
}
|
|
if action == "rollback" {
|
|
var setup bool
|
|
_ = f.db.Get(&setup, "SELECT setup_required FROM desktops WHERE id='desk_demo_windows'")
|
|
_ = f.db.Get(&n, "SELECT count(*) FROM connection_tickets WHERE desktop_id='desk_demo_windows'")
|
|
if !setup || n != 0 {
|
|
t.Fatal("rollback did not invalidate initialization and tickets")
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
func TestSnapshotPreflightRestrictions(t *testing.T) {
|
|
f := newFixtureWorker(t, false)
|
|
p := fakeSnapshotPVE(t, f)
|
|
admin := f.login("admin")
|
|
base := copySnapshotConfig(p.config)
|
|
target := copySnapshotConfig(p.target)
|
|
for _, condition := range []string{"running", "pending", "lock", "passthrough", "excluded-disk", "physical", "layout", "memory-list", "memory-config", "revision", "digest", "children", "permissions", "unsupported", "confirm", "active-session", "target-template", "invalid-name", "reserved", "extra-description", "extra-confirm", "extra-revision"} {
|
|
t.Run(condition, func(t *testing.T) {
|
|
p.config = copySnapshotConfig(base)
|
|
p.target = copySnapshotConfig(target)
|
|
p.items = []proxmox.Snapshot{{Name: "base", Time: 123, Description: "baseline"}}
|
|
p.state = "stopped"
|
|
p.pending = false
|
|
p.denyManage = false
|
|
p.denyRestore = false
|
|
p.unsupported = false
|
|
b := snapshotBody(p, "rollback")
|
|
switch condition {
|
|
case "running":
|
|
p.state = "running"
|
|
case "pending":
|
|
p.pending = true
|
|
case "lock":
|
|
p.config["lock"] = "backup"
|
|
case "passthrough":
|
|
p.config["hostpci0"] = "0000:01:00.0"
|
|
case "excluded-disk":
|
|
p.config["scsi0"] = "local:vm-9000,size=4G,snapshot=0"
|
|
case "physical":
|
|
p.config["scsi0"] = "/dev/disk/by-id/test,size=4G"
|
|
case "layout":
|
|
p.target["scsi0"] = "local:vm-9000,size=8G"
|
|
case "memory-list":
|
|
p.items[0].Memory = 1
|
|
b["revision"] = snapshotRevision(p.items[0])
|
|
case "memory-config":
|
|
p.target["vmstate"] = "local:vm-9000-state"
|
|
case "revision":
|
|
b["revision"] = "stale"
|
|
case "digest":
|
|
b["digest"] = "stale"
|
|
case "children":
|
|
b = snapshotBody(p, "delete")
|
|
p.items = append(p.items, proxmox.Snapshot{Name: "child", Parent: "base"})
|
|
case "permissions":
|
|
p.denyManage = true
|
|
p.denyRestore = true
|
|
case "unsupported":
|
|
b = snapshotBody(p, "create")
|
|
p.unsupported = true
|
|
case "confirm":
|
|
b["confirmVmid"] = 100
|
|
case "active-session":
|
|
_, _ = f.db.Exec("UPDATE terminals SET session_desktop_id='desk_demo_windows',last_seen=now() WHERE id='term_demo_001'")
|
|
case "target-template":
|
|
p.target["template"] = 1
|
|
case "invalid-name":
|
|
b["name"] = "../base"
|
|
case "reserved":
|
|
b = snapshotBody(p, "create")
|
|
b["name"] = "current"
|
|
case "extra-description":
|
|
b["description"] = "ignored?"
|
|
case "extra-confirm":
|
|
b = snapshotBody(p, "create")
|
|
b["confirmVmid"] = 9000
|
|
case "extra-revision":
|
|
b = snapshotBody(p, "create")
|
|
b["revision"] = "extra"
|
|
}
|
|
code, out := f.request("POST", snapshotEndpoint+"/check", admin, b, "")
|
|
expect(t, 409, code, out)
|
|
code, out = f.request("POST", snapshotEndpoint+"/actions", admin, b, "block-"+condition)
|
|
expect(t, 409, code, out)
|
|
_, _ = f.db.Exec("UPDATE terminals SET session_desktop_id=NULL WHERE id='term_demo_001'")
|
|
if p.writes != 0 {
|
|
t.Fatal("preflight performed mutation")
|
|
}
|
|
})
|
|
}
|
|
p.config = base
|
|
p.target = target
|
|
p.state = "stopped"
|
|
p.denyManage = true
|
|
p.denyRestore = false
|
|
b := snapshotBody(p, "rollback")
|
|
code, out := f.request("POST", snapshotEndpoint+"/check", admin, b, "")
|
|
expect(t, 200, code, out)
|
|
p.denyManage = false
|
|
p.denyRestore = true
|
|
code, out = f.request("POST", snapshotEndpoint+"/check", admin, b, "")
|
|
expect(t, 200, code, out)
|
|
b["force"] = true
|
|
code, out = f.request("POST", snapshotEndpoint+"/actions", admin, b, "strict")
|
|
expect(t, 400, code, out)
|
|
}
|
|
func TestSnapshotUnknownNeverReplayed(t *testing.T) {
|
|
for _, outcome := range []string{"reject", "ambiguous", "missing-upid", "poll-failure", "failed"} {
|
|
t.Run(outcome, func(t *testing.T) {
|
|
f := newFixtureWorker(t, false)
|
|
p := fakeSnapshotPVE(t, f)
|
|
admin := f.login("admin")
|
|
b := snapshotBody(p, "rollback")
|
|
p.outcome = outcome
|
|
task := queueSnapshotTest(t, f, admin, b, "unknown")
|
|
f.server.execute(context.Background(), task)
|
|
task = readSnapshotTask(t, f, task.ID)
|
|
want := "timeout"
|
|
if outcome == "reject" {
|
|
want = "failed"
|
|
}
|
|
if task.Status != want {
|
|
t.Fatal(task)
|
|
}
|
|
if want == "failed" {
|
|
return
|
|
}
|
|
_, _ = f.db.Exec("UPDATE desktops SET power_state='running',setup_required=false WHERE id='desk_demo_windows'")
|
|
_, _ = f.db.Exec("UPDATE terminals SET token_hash=$1,last_seen=now() WHERE id='term_demo_001'", store.Hash("snapshot-device"))
|
|
code, out := f.request("POST", "/terminal/connect-ticket", "snapshot-device", map[string]string{"desktopId": "desk_demo_windows"}, "")
|
|
expect(t, 409, code, out)
|
|
if out["code"] != "hardware_unconfirmed" {
|
|
t.Fatal(out)
|
|
}
|
|
code, out = f.request("POST", "/desktops/desk_demo_windows/actions", admin, map[string]string{"action": "start"}, "blocked")
|
|
expect(t, 409, code, out)
|
|
_, _ = f.db.Exec("UPDATE tasks SET status='running' WHERE id=$1", task.ID)
|
|
f.server.execute(context.Background(), task)
|
|
task = readSnapshotTask(t, f, task.ID)
|
|
if p.writes != 1 {
|
|
t.Fatal("snapshot replayed", p.writes)
|
|
}
|
|
code, out = f.request("POST", "/tasks/"+task.ID+"/snapshot-reconcile", admin, nil, "")
|
|
expect(t, 409, code, out)
|
|
p.outcome = ""
|
|
// Reconciliation must use its own SQL transaction, even with a single connection.
|
|
f.db.SetMaxOpenConns(1)
|
|
code, out = f.request("POST", "/tasks/"+task.ID+"/snapshot-reconcile", admin, nil, "")
|
|
if outcome == "poll-failure" || outcome == "failed" {
|
|
expect(t, 200, code, out)
|
|
} else {
|
|
expect(t, 409, code, out)
|
|
}
|
|
if p.writes != 1 {
|
|
t.Fatal("reconcile mutated PVE")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
func TestSnapshotRechecksAndExpiry(t *testing.T) {
|
|
f := newFixtureWorker(t, false)
|
|
p := fakeSnapshotPVE(t, f)
|
|
admin := f.login("admin")
|
|
b := snapshotBody(p, "rollback")
|
|
task := queueSnapshotTest(t, f, admin, b, "changed")
|
|
p.target["memory"] = 4096
|
|
f.server.execute(context.Background(), task)
|
|
if readSnapshotTask(t, f, task.ID).Status != "failed" || p.writes != 0 {
|
|
t.Fatal("changed target submitted")
|
|
}
|
|
task = queueSnapshotTest(t, f, admin, b, "prepared-expire")
|
|
f.server.finishSnapshot(task, "", "expired", "")
|
|
claimed, err := f.server.claimSnapshot(context.Background(), task, snapshotInput{Action: "rollback"})
|
|
if err != nil || claimed {
|
|
t.Fatal("expired task could submit", err)
|
|
}
|
|
task = queueSnapshotTest(t, f, admin, b, "submitting-expire")
|
|
claimed, err = f.server.claimSnapshot(context.Background(), task, snapshotInput{Action: "rollback"})
|
|
if err != nil || !claimed {
|
|
t.Fatal(err)
|
|
}
|
|
f.server.finishSnapshot(task, "", "expired", "")
|
|
if readSnapshotTask(t, f, task.ID).Status != "timeout" {
|
|
t.Fatal("submitting expiry released protection")
|
|
}
|
|
}
|
|
func TestSnapshotWorkerDoesNotBlockOrdinaryTasks(t *testing.T) {
|
|
f := newFixtureWorker(t, false)
|
|
p := fakeSnapshotPVE(t, f)
|
|
admin := f.login("admin")
|
|
p.outcome = "running"
|
|
task := queueSnapshotTest(t, f, admin, snapshotBody(p, "create"), "long")
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() { f.server.RunWorker(ctx); close(done) }()
|
|
defer func() { cancel(); <-done }()
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for readSnapshotTask(t, f, task.ID).ProviderRef == "" {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("snapshot never started")
|
|
}
|
|
time.Sleep(25 * time.Millisecond)
|
|
}
|
|
code, out := f.request("POST", "/desktops/desk_demo_kylin/actions", admin, map[string]string{"action": "start"}, "normal")
|
|
expect(t, 202, code, out)
|
|
for readSnapshotTask(t, f, out["id"].(string)).Status != "success" {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("long snapshot blocked ordinary task")
|
|
}
|
|
time.Sleep(25 * time.Millisecond)
|
|
}
|
|
if readSnapshotTask(t, f, task.ID).Status != "running" {
|
|
t.Fatal("snapshot unexpectedly completed")
|
|
}
|
|
}
|
|
func TestSnapshotConfigHash(t *testing.T) {
|
|
a := map[string]any{"cores": 2, "memory": 2048, "vmgenid": "one", "digest": "first", "parent": "old"}
|
|
b := map[string]any{"cores": "2", "memory": "2048", "vmgenid": "two", "digest": "second", "parent": "new", "snaptime": 4}
|
|
if snapshotConfigHash(a) != snapshotConfigHash(b) {
|
|
t.Fatal("transient fields or scalar representation affect hash")
|
|
}
|
|
b["memory"] = 4096
|
|
if snapshotConfigHash(a) == snapshotConfigHash(b) {
|
|
t.Fatal("material config change ignored")
|
|
}
|
|
}
|