proxmox: contract and observability tests, integration test, testing docs

The contract sweep in provider_test.go: every method and Check returns
within a second of a cancel, every wrapErr path carries the configured
name, op and id, the V0/V1/V2 logging tiers and the no-op lines, and a
Get over the real client whose HTTP spans are children of provider.get.
proxmox_integration_test.go (-tags=integration) holds TestProxmox_lifecycle
and TestProxmoxCheck_realHost; both compile and skip without
PROXMOX_TEST_*, and have not run against a host yet. make
test-integration-proxmox runs them. docs/testing.md gains the proxmox,
nocloud and keyedlock rows, the Live Proxmox gaps and the integration
section; spec entry 21's section 11 list is ticked.

Waits that could hang the package under a mutation are bounded: the
concurrent-Create loop and eight direct blocking calls in older tests.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-23 20:43:00 +00:00
co-authored by Claude Opus 5.5
parent 2f89c1fffb
commit 6e9ef26dcb
11 changed files with 1585 additions and 34 deletions
@@ -0,0 +1,624 @@
//go:build integration
package proxmox
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"sync"
"testing"
"time"
"github.com/go-logr/logr"
"github.com/go-logr/logr/funcr"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
"k8s.io/apimachinery/pkg/types"
logf "sigs.k8s.io/controller-runtime/pkg/log"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
)
// The real-host test of spec entry 21. It runs only with -tags=integration
// and skips itself unless PROXMOX_TEST_* names a host; see docs/testing.md
// for the variables and what one run asserts.
const (
integrationProvider = "proxmox-integration"
integrationImage = "test-template"
integrationMachine = "small"
integrationPort = 3128
// integrationDiskGiB is above the 3 GiB of the recon template, so a run
// on such a template exercises the resize task; a larger template skips
// it.
integrationDiskGiB = 4
// integrationCloudInit is the user-data the VM boots with. Nothing
// asserts on the guest running it; it exists so the sentinel check can
// prove that no span attribute carries the cloud-init text.
integrationCloudInit = "#cloud-config\nruncmd:\n - echo proxmox-integration\n"
defaultIntegrationTimeout = 20 * time.Minute
defaultIntegrationLogLvl = 1
getPollInterval = 5 * time.Second
cleanupBudget = 5 * time.Minute
)
type integrationEnv struct {
url, tokenFile, node, isoStorage string
caFile, storage, bridge string
templateVMID, vlanTag int
vmidMin, vmidMax int
insecure, keep bool
timeout time.Duration
logLevel int
}
// loadIntegrationEnv reads the PROXMOX_TEST_* variables, skipping the test
// under -short or when a required one is missing.
func loadIntegrationEnv(t *testing.T) integrationEnv {
t.Helper()
if testing.Short() {
t.Skip("real-host test skipped under -short")
}
env := integrationEnv{timeout: defaultIntegrationTimeout, logLevel: defaultIntegrationLogLvl}
var templateVMID string
required := []struct {
name string
dst *string
}{
{"PROXMOX_TEST_URL", &env.url},
{"PROXMOX_TEST_TOKEN_FILE", &env.tokenFile},
{"PROXMOX_TEST_NODE", &env.node},
{"PROXMOX_TEST_TEMPLATE_VMID", &templateVMID},
{"PROXMOX_TEST_ISO_STORAGE", &env.isoStorage},
}
for _, r := range required {
v := os.Getenv(r.name)
if v == "" {
t.Skipf("%s is not set; the proxmox integration test needs a real Proxmox VE host (see docs/testing.md)", r.name)
}
*r.dst = v
}
env.templateVMID = envInt(t, "PROXMOX_TEST_TEMPLATE_VMID", 0)
env.caFile = os.Getenv("PROXMOX_TEST_CA_FILE")
env.storage = os.Getenv("PROXMOX_TEST_STORAGE")
env.bridge = os.Getenv("PROXMOX_TEST_BRIDGE")
env.vlanTag = envInt(t, "PROXMOX_TEST_VLAN_TAG", 0)
env.vmidMin = envInt(t, "PROXMOX_TEST_VMID_MIN", defaultVMIDMin)
env.vmidMax = envInt(t, "PROXMOX_TEST_VMID_MAX", defaultVMIDMax)
if env.vmidMin > env.vmidMax {
t.Fatalf("PROXMOX_TEST_VMID_MIN=%d is above PROXMOX_TEST_VMID_MAX=%d", env.vmidMin, env.vmidMax)
}
env.keep = os.Getenv("PROXMOX_TEST_KEEP") != ""
if v := os.Getenv("PROXMOX_TEST_INSECURE"); v != "" {
b, err := strconv.ParseBool(v)
if err != nil {
t.Fatalf("PROXMOX_TEST_INSECURE=%q: %v", v, err)
}
env.insecure = b
}
if v := os.Getenv("PROXMOX_TEST_TIMEOUT"); v != "" {
d, err := time.ParseDuration(v)
if err != nil {
t.Fatalf("PROXMOX_TEST_TIMEOUT=%q: %v", v, err)
}
env.timeout = d
}
env.logLevel = envInt(t, "PROXMOX_TEST_LOG_LEVEL", defaultIntegrationLogLvl)
return env
}
// envInt reads a non-negative integer variable, def when it is unset.
func envInt(t *testing.T, name string, def int) int {
t.Helper()
v := os.Getenv(name)
if v == "" {
return def
}
n, err := strconv.Atoi(v)
if err != nil || n < 0 {
t.Fatalf("%s=%q: want a non-negative integer", name, v)
}
return n
}
// providerConfig is what the operator would load for this host: one node,
// one template, one machine type.
func (e integrationEnv) providerConfig() provider.ProviderConfig {
return provider.ProviderConfig{
Name: integrationProvider,
Type: "proxmox",
Proxmox: &provider.ProxmoxConfig{
URL: e.url,
TokenFile: e.tokenFile,
CAFile: e.caFile,
InsecureSkipTLSVerify: e.insecure,
Nodes: []string{e.node},
ISOStorage: e.isoStorage,
Storage: e.storage,
Bridge: e.bridge,
VLANTag: e.vlanTag,
VMIDRange: &provider.ProxmoxVMIDRange{Min: e.vmidMin, Max: e.vmidMax},
Templates: map[string]provider.ProxmoxTemplate{integrationImage: {VMID: e.templateVMID}},
MachineTypes: map[string]provider.ProxmoxMachineType{
integrationMachine: {Cores: 1, MemoryMiB: 1024, DiskGiB: integrationDiskGiB},
},
},
}
}
// integrationLog routes the provider's logr output into t.Log through funcr
// at the chosen verbosity, keeps every line for assertions, and strips the
// key/value pairs that are constant for the run (stated once up front) so
// each line stays short. Lines arriving after the test ended are dropped:
// t.Log panics then.
type integrationLog struct {
t *testing.T
mu sync.Mutex
drop []string
lines []string
done bool
}
func newIntegrationLog(t *testing.T, level int, drop ...string) (*integrationLog, logr.Logger) {
t.Helper()
sink := &integrationLog{t: t, drop: drop}
t.Cleanup(func() {
sink.mu.Lock()
defer sink.mu.Unlock()
sink.done = true
})
log := funcr.New(func(prefix, args string) { sink.write(prefix, args) }, funcr.Options{Verbosity: level})
return sink, log
}
// dropValue adds a rendered pair (`"key"="value"`) to strip from later lines.
func (l *integrationLog) dropValue(key, value string) {
l.mu.Lock()
defer l.mu.Unlock()
l.drop = append(l.drop, fmt.Sprintf("%q=%q", key, value))
}
func (l *integrationLog) write(prefix, args string) {
l.mu.Lock()
defer l.mu.Unlock()
l.lines = append(l.lines, args)
if l.done {
return
}
for _, d := range l.drop {
args = strings.ReplaceAll(args, " "+d, "")
}
l.t.Logf("%s %s %s", time.Now().Format("15:04:05.000"), prefix, args)
}
func (l *integrationLog) matching(substr string) []string {
l.mu.Lock()
defer l.mu.Unlock()
var out []string
for _, line := range l.lines {
if strings.Contains(line, substr) {
out = append(out, line)
}
}
return out
}
// newIntegrationProvider builds the provider the way New does, with the
// span recorder's tracer on the transport, so the HTTP spans land in the
// same recorder as provider.WithTracing's provider.* spans. The bare client
// is returned too, for the test's own read-only inspection of the host.
func newIntegrationProvider(t *testing.T, pc provider.ProviderConfig, tp *sdktrace.TracerProvider) (*Provider, *httpAPI) {
t.Helper()
cfg := withDefaults(*pc.Proxmox)
client, err := newHTTPAPI(cfg, tracing.WithTracerProvider(tp))
if err != nil {
t.Fatalf("newHTTPAPI error = %v", err)
}
return newProvider(pc.Name, cfg, observe(client, pc.Name)), client
}
// randomUID is a UUID-shaped string, so the run's VM never collides with a
// previous run's leftovers or with a real Proxy's.
func randomUID(t *testing.T) string {
t.Helper()
var b [16]byte
if _, err := rand.Read(b[:]); err != nil {
t.Fatalf("reading random bytes: %v", err)
}
h := hex.EncodeToString(b[:])
return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
}
// stepDone logs one lifecycle step with its duration; the log is the record
// of how long a clone or a boot took on this host.
func stepDone(t *testing.T, step string, start time.Time) {
t.Helper()
t.Logf("step %s: %s", step, time.Since(start).Round(time.Millisecond))
}
// waitUntil polls cond every interval until it is true or ctx ends.
func waitUntil(ctx context.Context, t *testing.T, interval time.Duration, what string, cond func() bool) {
t.Helper()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for !cond() {
select {
case <-ticker.C:
case <-ctx.Done():
t.Fatalf("gave up waiting for %s: %v", what, ctx.Err())
}
}
}
// qemuVMs lists the cluster's QEMU VMs by VMID.
func qemuVMs(ctx context.Context, t *testing.T, client *httpAPI) map[int]vmResource {
t.Helper()
rs, err := client.ClusterResources(ctx)
if err != nil {
t.Fatalf("listing cluster resources: %v", err)
}
out := make(map[int]vmResource, len(rs))
for _, r := range rs {
if r.Type == "qemu" {
out[r.VMID] = r
}
}
return out
}
// vmidByName finds the VM named name, as a crashed Create may have left it.
func vmidByName(ctx context.Context, t *testing.T, client *httpAPI, name string) (int, bool) {
t.Helper()
for vmid, r := range qemuVMs(ctx, t, client) {
if r.Name == name {
return vmid, true
}
}
return 0, false
}
// seedOnStorage reports whether the ISO storage on node holds name's seed.
// It is one extra read-only GET of /nodes/{node}/storage/{storage}/content
// (captured as storage_content_iso.json): the seam has no method to list
// storage content, and gets none for a test.
func seedOnStorage(ctx context.Context, t *testing.T, client *httpAPI, node, storage, name string) bool {
t.Helper()
var content []struct {
VolID string `json:"volid"`
}
path := apiPath("nodes", node, "storage", storage, "content")
if err := client.do(ctx, http.MethodGet, path, url.Values{"content": {"iso"}}, nil, &content); err != nil {
t.Fatalf("listing iso content of %s on %s: %v", storage, node, err)
}
want := seedVolID(storage, name)
for _, c := range content {
if c.VolID == want {
return true
}
}
return false
}
// checkNode runs the health check and requires the one node Ready, with the
// details an admin reads.
func checkNode(ctx context.Context, t *testing.T, built provider.Provider, env integrationEnv) {
t.Helper()
checker, ok := built.(provider.Checker)
if !ok {
t.Fatalf("New() returned %T, which is not a provider.Checker", built)
}
targets := checker.Check(ctx)
if len(targets) != 1 {
t.Fatalf("Check() returned %d targets, want 1: %+v", len(targets), targets)
}
ts := targets[0]
if !ts.Ready || ts.Reason != provider.ReasonReachable {
t.Fatalf("Check() = ready %v reason %q message %q, want Ready/%s (details %v)", ts.Ready, ts.Reason, ts.Message, provider.ReasonReachable, ts.Details)
}
for _, key := range []string{"pveVersion", "kernel", "tokenID", "isoStorageAvailGiB", "templates"} {
if ts.Details[key] == "" {
t.Errorf("Check() details lack %q: %v", key, ts.Details)
}
}
if ts.Details["templates"] != "1/1" {
t.Errorf("Check() templates = %q, want 1/1", ts.Details["templates"])
}
t.Logf("node %s (%s): pve %s, kernel %s, token %s, %s free on %s, templates %s, latency %s",
ts.Name, ts.Address, ts.Details["pveVersion"], ts.Details["kernel"], ts.Details["tokenID"],
ts.Details["isoStorageAvailGiB"]+" GiB", env.isoStorage, ts.Details["templates"], ts.Latency.Round(time.Millisecond))
}
func TestProxmox_lifecycle(t *testing.T) {
env := loadIntegrationEnv(t)
pc := env.providerConfig()
sink, log := newIntegrationLog(t, env.logLevel,
fmt.Sprintf("%q=%q", "provider", integrationProvider),
fmt.Sprintf("%q=%q", "node", env.node))
ctx, cancel := context.WithTimeout(logf.IntoContext(context.Background(), log), env.timeout)
defer cancel()
t.Logf("provider=%s url=%s node=%s template=%d isoStorage=%s storage=%q bridge=%q vlanTag=%d vmids %d..%d budget=%s log level %d; provider, node and name are left off every provider log line",
integrationProvider, env.url, env.node, env.templateVMID, env.isoStorage, env.storage, env.bridge, env.vlanTag,
env.vmidMin, env.vmidMax, env.timeout, env.logLevel)
if env.insecure {
t.Log("PROXMOX_TEST_INSECURE is set: the API's certificate is NOT verified for this run")
}
// 1. New returns without a request.
start := time.Now()
built, err := New(ctx, pc)
elapsed := time.Since(start)
if err != nil {
t.Fatalf("New() error = %v", err)
}
if elapsed > 100*time.Millisecond {
t.Errorf("New() took %s, want under 100ms: it must not make a request", elapsed)
}
if lines := sink.matching(`"msg"="proxmox api call"`); len(lines) > 0 {
t.Errorf("New() called the API before any operation: %q", lines)
}
stepDone(t, "1 New", start)
// 2. Check reports the node Ready with details.
start = time.Now()
checkNode(ctx, t, built, env)
stepDone(t, "2 Check", start)
// 3. Every later call goes through the tracing decorator, with the
// transport's HTTP spans recorded alongside.
sr := tracetest.NewSpanRecorder()
tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
t.Cleanup(func() { _ = tp.Shutdown(context.Background()) })
p, client := newIntegrationProvider(t, pc, tp)
traced := provider.WithTracing(pc.Name, p, tracing.WithTracerProvider(tp))
// 4. Create.
uid := randomUID(t)
name := provider.NameFromUID(types.UID(uid))
sink.dropValue("name", name)
req := provider.CreateRequest{
Name: name,
UID: uid,
Namespace: "integration",
ProxyName: "proxmox-lifecycle",
Placement: provider.Placement{Zone: env.node, MachineType: integrationMachine, Image: integrationImage},
CloudInit: integrationCloudInit,
Port: integrationPort,
}
t.Logf("VM name %s, uid %s", name, uid)
deleted := false
t.Cleanup(func() {
if deleted {
return
}
cctx, ccancel := context.WithTimeout(logf.IntoContext(context.Background(), log), cleanupBudget)
defer ccancel()
vmid, found := vmidByName(cctx, t, client, name)
if env.keep {
msg := "PROXMOX_TEST_KEEP is set: the VM and its seed stay. Clean up as root on " + env.node + " with:\n"
if found {
msg += fmt.Sprintf(" qm stop %d\n qm destroy %d --purge\n", vmid, vmid)
}
t.Log(msg + " pvesm free " + seedVolID(env.isoStorage, name))
return
}
if !found {
// No VM, but a seed may be left: Delete of an absent VM still
// deletes it.
vmid = vmidFor(uid, env.vmidMin, env.vmidMax, 0)
}
if err := p.Delete(cctx, formatProviderID(vmid, name)); err != nil {
t.Errorf("cleanup Delete of VM %d error = %v; remove it and %s by hand", vmid, err, seedVolID(env.isoStorage, name))
}
})
start = time.Now()
id, err := traced.Create(ctx, req)
if err != nil {
t.Fatalf("Create() error = %v", err)
}
vmid, gotName, err := parseProviderID(id)
if err != nil || gotName != name {
t.Fatalf("Create() = %q, want vms/<vmid>/%s (parse error %v)", id, name, err)
}
if vmid < env.vmidMin || vmid > env.vmidMax {
t.Errorf("Create() VMID %d is outside %d..%d", vmid, env.vmidMin, env.vmidMax)
}
sink.dropValue("vmid", strconv.Itoa(vmid))
createReturned := time.Now()
t.Logf("VM %d, id %s", vmid, id)
stepDone(t, "4 Create", start)
// 5. Create again: same ID, nothing new in the cluster.
start = time.Now()
before := len(qemuVMs(ctx, t, client))
again, err := traced.Create(ctx, req)
if err != nil {
t.Fatalf("second Create() error = %v", err)
}
if again != id {
t.Errorf("second Create() = %q, want %q", again, id)
}
if after := len(qemuVMs(ctx, t, client)); after != before {
t.Errorf("second Create() changed the cluster's VM count from %d to %d", before, after)
}
stepDone(t, "5 Create again", start)
// 6. Get reaches Running with an IPv4, and reads back what Create wrote.
start = time.Now()
var inst *provider.Instance
waitUntil(ctx, t, getPollInterval, "the VM to be Running with an IPv4", func() bool {
var err error
inst, err = traced.Get(ctx, id)
if errors.Is(err, provider.ErrTransient) {
t.Logf("Get(%s) is Transient while the guest boots: %v", id, err)
return false
}
if err != nil {
t.Fatalf("Get(%s) error = %v", id, err)
}
return inst.State == provider.StateRunning && net.ParseIP(inst.IP).To4() != nil
})
assertMarkReadBack(ctx, t, client, env.node, vmid, inst, uid)
t.Logf("VM running at %s, %s after Create returned (createdAt %s)", inst.IP, time.Since(createReturned).Round(time.Millisecond), inst.CreatedAt.Format(time.RFC3339))
stepDone(t, "6 Get until Running", start)
// 7. ListByTag contains it.
start = time.Now()
list, err := traced.ListByTag(ctx)
if err != nil {
t.Fatalf("ListByTag() error = %v", err)
}
assertListed(t, list, inst)
stepDone(t, "7 ListByTag", start)
// 8. Reachability, logged only: this machine may not route to the VM's
// network.
start = time.Now()
addr := net.JoinHostPort(inst.IP, "22")
if conn, err := net.DialTimeout("tcp", addr, 3*time.Second); err != nil {
t.Logf("TCP connect to %s failed (informational; the test host may not route to the VM's network): %v", addr, err)
} else {
_ = conn.Close()
t.Logf("TCP connect to %s succeeded", addr)
}
stepDone(t, "8 TCP connect", start)
if env.keep {
t.Log("PROXMOX_TEST_KEEP is set: skipping Delete and the leftovers check")
} else {
// 9. Delete, Get is NotFound, Delete again.
start = time.Now()
if err := traced.Delete(ctx, id); err != nil {
t.Fatalf("Delete() error = %v", err)
}
deleted = true
if _, err := traced.Get(ctx, id); !errors.Is(err, provider.ErrNotFound) {
t.Errorf("Get() after Delete error = %v, want ErrNotFound", err)
}
if err := traced.Delete(ctx, id); err != nil {
t.Errorf("second Delete() error = %v, want nil", err)
}
stepDone(t, "9 Delete, Get, Delete", start)
// 10. No VM with the name and no seed remain.
start = time.Now()
if left, ok := vmidByName(ctx, t, client, name); ok {
t.Errorf("VM %d named %s is still in the cluster after Delete", left, name)
}
if seedOnStorage(ctx, t, client, env.node, env.isoStorage, name) {
t.Errorf("%s is still on %s after Delete", seedVolID(env.isoStorage, name), env.node)
}
stepDone(t, "10 leftovers", start)
}
// 11. Spans: HTTP spans are children of the decorator's.
assertLifecycleSpans(t, sr.Ended())
}
// assertMarkReadBack checks that Get's UID and CreatedAt are the mark
// Create wrote into the VM's description, read back from its config, and
// that CreatedAt is recent.
func assertMarkReadBack(ctx context.Context, t *testing.T, client *httpAPI, node string, vmid int, inst *provider.Instance, uid string) {
t.Helper()
cfg, err := client.VMConfig(ctx, node, vmid)
if err != nil {
t.Fatalf("reading the config of VM %d: %v", vmid, err)
}
written := parseDescription(cfg["description"])
if inst.UID != uid || written.UID != uid {
t.Errorf("Get() UID = %q, the VM's mark says %q, want %q", inst.UID, written.UID, uid)
}
if !inst.CreatedAt.Equal(written.CreatedAt) || inst.CreatedAt.IsZero() {
t.Errorf("Get() CreatedAt = %s, the VM's mark says %s; want them equal and set", inst.CreatedAt, written.CreatedAt)
} else if skew := time.Since(inst.CreatedAt).Abs(); skew > 5*time.Minute {
t.Errorf("Get() CreatedAt = %s, %s from now; want within a few minutes", inst.CreatedAt, skew)
}
}
// assertListed checks that list holds inst's ID with its UID and
// CreatedAt.
func assertListed(t *testing.T, list []provider.Instance, inst *provider.Instance) {
t.Helper()
for _, li := range list {
if li.ID != inst.ID {
continue
}
if li.UID != inst.UID || !li.CreatedAt.Equal(inst.CreatedAt) {
t.Errorf("ListByTag() entry = %+v, want UID %q and CreatedAt %s", li, inst.UID, inst.CreatedAt)
}
return
}
t.Errorf("ListByTag() = %d instances, none with ID %s", len(list), inst.ID)
}
// assertLifecycleSpans checks that at least one provider.create span has an
// HTTP child, and that no attribute of any span carries the cloud-init text.
func assertLifecycleSpans(t *testing.T, spans []sdktrace.ReadOnlySpan) {
t.Helper()
satisfied := false
for _, create := range spans {
if create.Name() != "provider.create" {
continue
}
parent := create.SpanContext().SpanID()
children := 0
for _, s := range spans {
if s.Parent().SpanID() == parent && strings.HasPrefix(s.Name(), "HTTP ") {
children++
}
}
t.Logf("provider.create span %s: %d HTTP children", parent, children)
if children > 0 {
satisfied = true
}
}
if !satisfied {
names := make([]string, len(spans))
for i, s := range spans {
names[i] = s.Name()
}
t.Errorf("no provider.create span has an HTTP child; spans: %v", names)
}
for _, s := range spans {
for _, a := range s.Attributes() {
v := a.Value.Emit()
if strings.Contains(v, integrationCloudInit) || strings.Contains(v, "#cloud-config") {
t.Errorf("span %s attribute %s carries the cloud-init text", s.Name(), a.Key)
}
}
}
}
// TestProxmoxCheck_realHost runs the health check against the host and
// creates nothing.
func TestProxmoxCheck_realHost(t *testing.T) {
env := loadIntegrationEnv(t)
pc := env.providerConfig()
_, log := newIntegrationLog(t, env.logLevel,
fmt.Sprintf("%q=%q", "provider", integrationProvider),
fmt.Sprintf("%q=%q", "node", env.node))
ctx, cancel := context.WithTimeout(logf.IntoContext(context.Background(), log), env.timeout)
defer cancel()
if env.insecure {
t.Log("PROXMOX_TEST_INSECURE is set: the API's certificate is NOT verified for this run")
}
built, err := New(ctx, pc)
if err != nil {
t.Fatalf("New() error = %v", err)
}
checkNode(ctx, t, built, env)
}