egress-proxies-operator/internal/provider/kvm2/create_test.go

package kvm2

import (
	"bytes"
	"errors"
	"math"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/digitalocean/go-libvirt"

	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/nocloud"
)

func TestCreate_repeatReturnsSameIDAndCreatesNothing(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	first := fx.create(testName)
	before := len(fx.fake.callLog())
	defines, vols, uploads := fx.fake.callCount("DomainDefineXML"), fx.fake.callCount("StorageVolCreateXML"), fx.fake.callCount("StorageVolUpload")

	second := fx.create(testName)
	if second != first {
		t.Errorf("second Create = %q, want %q", second, first)
	}
	if got := fx.fake.callCount("DomainDefineXML"); got != defines {
		t.Errorf("DomainDefineXML calls after repeat = %d, want %d", got, defines)
	}
	if got := fx.fake.callCount("StorageVolCreateXML"); got != vols {
		t.Errorf("StorageVolCreateXML calls after repeat = %d, want %d", got, vols)
	}
	if got := fx.fake.callCount("StorageVolUpload"); got != uploads {
		t.Errorf("StorageVolUpload calls after repeat = %d, want %d", got, uploads)
	}
	if got := fx.fake.callCount("DomainCreate"); got != 1 {
		t.Errorf("DomainCreate calls = %d, want 1", got)
	}
	repeat := fx.fake.callLog()[before:]
	for _, rpc := range repeat {
		switch rpc {
		case "DomainLookupByName", "DomainGetMetadata", "DomainGetState":
		default:
			t.Errorf("repeat Create made %s; want only lookup, metadata and state reads: %v", rpc, repeat)
		}
	}
	if len(fx.logs.matching("kvm2 domain already defined")) != 1 {
		t.Errorf("no 'already defined' line:\n%s", strings.Join(fx.logs.all(), "\n"))
	}
}

func TestCreate_resumesAfterEachPartialStep(t *testing.T) {
	t.Parallel()
	overlay, seed := overlayVolumeName(testName), seedVolumeName(testName)
	tests := []struct {
		name  string
		setup func(fx *fixture)
		check func(t *testing.T, fx *fixture)
	}{
		{
			name: "base without marker is refetched",
			setup: func(fx *fixture) {
				fx.putVol(fx.baseName(), []byte("truncated"))
			},
			check: func(t *testing.T, fx *fixture) {
				if got := fx.fake.vol(fx.baseName()); got == nil || !bytes.Equal(got.content, fx.image) {
					t.Errorf("base volume not refetched")
				}
				if fx.fake.vol(fx.baseName()+".sha256") == nil {
					t.Errorf("marker not written")
				}
			},
		},
		{
			name: "overlay exists is reused",
			setup: func(fx *fixture) {
				fx.cacheImage()
				fx.putVol(overlay, []byte("overlay"))
			},
			check: func(t *testing.T, fx *fixture) {
				if got := fx.fake.vol(overlay); got == nil || string(got.content) != "overlay" {
					t.Errorf("overlay volume was replaced")
				}
				if len(fx.logs.matching("overlay volume already exists")) != 1 {
					t.Errorf("no reuse line:\n%s", strings.Join(fx.logs.all(), "\n"))
				}
			},
		},
		{
			name: "truncated seed is replaced",
			setup: func(fx *fixture) {
				fx.cacheImage()
				fx.putVol(seed, []byte("cut"))
			},
			check: func(t *testing.T, fx *fixture) {
				got := fx.fake.vol(seed)
				if got == nil || len(got.content) <= 3 || !bytes.Contains(got.content, []byte(nocloud.SeedLabel)) {
					t.Errorf("seed volume not rebuilt")
				}
			},
		},
		{
			name: "domain defined but shutoff is started",
			setup: func(fx *fixture) {
				fx.cacheImage()
				fx.putVol(overlay, []byte("overlay"))
				fx.putVol(seed, []byte("seed"))
				fx.fake.addDomain(testName, libvirt.DomainShutoff, managedMetadata(testUID, time.Now()))
			},
			check: func(t *testing.T, fx *fixture) {
				if got := fx.fake.callCount("DomainDefineXML"); got != 0 {
					t.Errorf("DomainDefineXML calls = %d, want 0", got)
				}
				if got := fx.fake.callCount("DomainCreate"); got != 1 {
					t.Errorf("DomainCreate calls = %d, want 1", got)
				}
			},
		},
		{
			name: "domain running is returned as is",
			setup: func(fx *fixture) {
				fx.fake.addDomain(testName, libvirt.DomainRunning, managedMetadata(testUID, time.Now()))
			},
			check: func(t *testing.T, fx *fixture) {
				for _, rpc := range []string{"DomainDefineXML", "DomainCreate", "StorageVolCreateXML"} {
					if got := fx.fake.callCount(rpc); got != 0 {
						t.Errorf("%s calls = %d, want 0", rpc, got)
					}
				}
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 1)
			tc.setup(fx)
			fx.create(testName)
			fx.withDomain(testName, func(d *fakeDomain) {
				if d.state != libvirt.DomainRunning {
					t.Errorf("domain state = %s, want running", d.state)
				}
			})
			tc.check(t, fx)
		})
	}
}

func TestCreate_returnsOnlyWhenRunning(t *testing.T) {
	t.Parallel()
	t.Run("returns after the domain reports running", func(t *testing.T) {
		t.Parallel()
		fx := newFixture(t, 1)
		fx.fake.startDelayPolls = 3
		fx.create(testName)
		fx.withDomain(testName, func(d *fakeDomain) {
			if d.state != libvirt.DomainRunning {
				t.Errorf("Create returned with domain state %s", d.state)
			}
		})
		if got := fx.fake.callCount("DomainGetState"); got < 4 {
			t.Errorf("DomainGetState calls = %d, want at least 4 (one before start, three polls)", got)
		}
	})
	t.Run("never running is transient after the cap", func(t *testing.T) {
		t.Parallel()
		fx := newFixture(t, 1, withRunningWait(time.Millisecond, 30*time.Millisecond))
		fx.fake.startDelayPolls = math.MaxInt32
		start := time.Now()
		id, err := fx.p.Create(fx.ctx, fx.req(testName))
		if id != "" {
			t.Errorf("Create = %q, want no ID", id)
		}
		assertProviderError(t, err, "create", provider.ErrTransient)
		if !strings.Contains(err.Error(), "did not reach running") {
			t.Errorf("error = %v, want the wait cap named", err)
		}
		if elapsed := time.Since(start); elapsed > 2*time.Second {
			t.Errorf("Create took %v, want the shrunk cap to end it", elapsed)
		}
	})
}

func TestCreate_domainThatDiesAfterStartIsTransientAtOnce(t *testing.T) {
	t.Parallel()
	for _, state := range []libvirt.DomainState{libvirt.DomainCrashed, libvirt.DomainShutoff, libvirt.DomainPaused} {
		t.Run(state.String(), func(t *testing.T) {
			t.Parallel()
			// The cap is well over the bound asserted below, so a poll that
			// ran to the cap would fail the test.
			fx := newFixture(t, 1, withRunningWait(time.Millisecond, 10*time.Second))
			fx.fake.stateAfterStart = state
			start := time.Now()
			id, err := fx.p.Create(fx.ctx, fx.req(testName))
			elapsed := time.Since(start)
			if id != "" {
				t.Errorf("Create = %q, want no ID", id)
			}
			assertProviderError(t, err, "create", provider.ErrTransient)
			if !strings.Contains(err.Error(), "right after start") {
				t.Errorf("error = %v, want the state after start named", err)
			}
			if elapsed > time.Second {
				t.Errorf("Create took %v, want it to give up at once, not poll to the cap", elapsed)
			}
			if got := fx.fake.callCount("DomainCreate"); got != 1 {
				t.Errorf("DomainCreate calls = %d, want 1", got)
			}
		})
	}
}

func TestCreate_pausedDomainIsPermanent(t *testing.T) {
	t.Parallel()
	for _, state := range []libvirt.DomainState{libvirt.DomainPaused, libvirt.DomainPmsuspended} {
		t.Run(state.String(), func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 1)
			fx.fake.addDomain(testName, state, managedMetadata(testUID, time.Now()))
			id, err := fx.p.Create(fx.ctx, fx.req(testName))
			if id != "" {
				t.Errorf("Create = %q, want no ID", id)
			}
			assertProviderError(t, err, "create", provider.ErrPermanent)
			if !strings.Contains(err.Error(), "cannot be started by the operator") {
				t.Errorf("error = %v, want the admin action named", err)
			}
			for _, rpc := range []string{"DomainCreate", "DomainDefineXML", "DomainDestroy"} {
				if got := fx.fake.callCount(rpc); got != 0 {
					t.Errorf("%s calls = %d, want 0", rpc, got)
				}
			}
			fx.withDomain(testName, func(d *fakeDomain) {
				if d.state != state {
					t.Errorf("domain state = %s, want it left %s", d.state, state)
				}
			})
		})
	}
}

func TestCreate_rejectsUnknownPlacement(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name    string
		mutate  func(req *provider.CreateRequest)
		wantMsg string
	}{
		{
			name:    "zone",
			mutate:  func(req *provider.CreateRequest) { req.Placement.Zone = "nas9" },
			wantMsg: `placement.zone "nas9" is not in kvm2.hosts`,
		},
		{
			name:    "machineType",
			mutate:  func(req *provider.CreateRequest) { req.Placement.MachineType = "xlarge" },
			wantMsg: `placement.machineType "xlarge" is not in kvm2.machineTypes`,
		},
		{
			name:    "image",
			mutate:  func(req *provider.CreateRequest) { req.Placement.Image = "ubuntu-24" },
			wantMsg: `placement.image "ubuntu-24" is neither a key in kvm2.images nor an http(s) URL`,
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 1)
			req := fx.req(testName)
			tc.mutate(&req)
			id, err := fx.p.Create(fx.ctx, req)
			if id != "" {
				t.Errorf("Create = %q, want no ID", id)
			}
			assertProviderError(t, err, "create", provider.ErrPermanent)
			if !strings.Contains(err.Error(), tc.wantMsg) {
				t.Errorf("error = %q, want substring %q", err, tc.wantMsg)
			}
			if got := fx.fake.callLog(); len(got) != 0 {
				t.Errorf("backend calls = %v, want none before validation", got)
			}
			if got := fx.dialer.dials.Load(); got != 0 {
				t.Errorf("dials = %d, want 0", got)
			}
		})
	}
}

func TestCreate_adoptionWithForeignUIDIsPermanent(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	fx.fake.addDomain(testName, libvirt.DomainRunning, managedMetadata("other-uid", time.Now()))
	id, err := fx.p.Create(fx.ctx, fx.req(testName))
	if id != "" {
		t.Errorf("Create = %q, want no ID", id)
	}
	assertProviderError(t, err, "create", provider.ErrPermanent)
	if !strings.Contains(err.Error(), "other-uid") || !strings.Contains(err.Error(), testUID) {
		t.Errorf("error = %q, want both UIDs named", err)
	}
	for _, rpc := range []string{"DomainCreate", "DomainDefineXML", "StorageVolCreateXML", "DomainDestroy"} {
		if got := fx.fake.callCount(rpc); got != 0 {
			t.Errorf("%s calls = %d, want 0", rpc, got)
		}
	}
}

func TestCreate_adoptsOnlyMarkedDomains(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name      string
		metadata  string
		wantAdopt bool
	}{
		{name: "matching uid", metadata: managedMetadata(testUID, time.Now()), wantAdopt: true},
		{name: "managed, empty uid", metadata: managedMetadata("", time.Now()), wantAdopt: true},
		{name: "no metadata element", metadata: ""},
		{name: "element without the managed mark", metadata: `<instance><uid>` + testUID + `</uid></instance>`},
		{name: "managed false", metadata: `<instance><managed>false</managed><uid>` + testUID + `</uid></instance>`},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 1)
			fx.fake.addDomain(testName, libvirt.DomainRunning, tc.metadata)
			id, err := fx.p.Create(fx.ctx, fx.req(testName))
			if tc.wantAdopt {
				if err != nil || id != formatProviderID(testHost, testName) {
					t.Fatalf("Create = %q, %v; want the existing domain adopted", id, err)
				}
			} else {
				if id != "" {
					t.Errorf("Create = %q, want no ID", id)
				}
				assertProviderError(t, err, "create", provider.ErrPermanent)
				if !strings.Contains(err.Error(), "carries no kvm2 mark") {
					t.Errorf("error = %v, want the missing mark named", err)
				}
			}
			for _, rpc := range []string{"DomainDefineXML", "DomainCreate", "StorageVolCreateXML", "DomainDestroy"} {
				if got := fx.fake.callCount(rpc); got != 0 {
					t.Errorf("%s calls = %d, want 0", rpc, got)
				}
			}
		})
	}
}

func TestCreate_poolOutOfSpaceIsQuotaExceeded(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	fx.fake.mu.Lock()
	fx.fake.pools[fakePoolName].available = 1 << 30
	fx.fake.mu.Unlock()
	_, err := fx.p.Create(fx.ctx, fx.req(testName))
	assertProviderError(t, err, "create", provider.ErrQuotaExceeded)
	if !strings.Contains(err.Error(), fakePoolName) || !strings.Contains(err.Error(), "needed") {
		t.Errorf("error = %q, want the pool and the needed bytes named", err)
	}
	for _, rpc := range []string{"StorageVolCreateXML", "StorageVolUpload", "DomainDefineXML"} {
		if got := fx.fake.callCount(rpc); got != 0 {
			t.Errorf("%s calls = %d, want 0", rpc, got)
		}
	}
}

func TestCreate_libvirtNotFoundIsTransientNotNotFound(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name string
		rpc  string
		err  error
	}{
		{name: "volume vanished", rpc: "StorageVolGetPath", err: fakeErr(libvirt.ErrNoStorageVol, "Storage volume not found")},
		{name: "domain vanished", rpc: "DomainGetState", err: fakeErr(libvirt.ErrNoDomain, "Domain not found")},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			fx := newFixture(t, 1)
			fx.fake.failWith(tc.rpc, tc.err)
			_, err := fx.p.Create(fx.ctx, fx.req(testName))
			assertProviderError(t, err, "create", provider.ErrTransient)
			if errors.Is(err, provider.ErrNotFound) {
				t.Errorf("Create error = %v, must never be NotFound", err)
			}
		})
	}
}

func TestCreate_cloudInitSentinelNeverLoggedOrInErrors(t *testing.T) {
	t.Parallel()
	const sentinel = "SENTINEL-s3cr3t-passw0rd-7f3a"
	fx := newFixture(t, 2)
	fx.dialer.backend = observe(fx.fake, testProvider, testHost)
	fx.fake.failWith("StorageVolUpload", fakeErr(libvirt.ErrInternalError, "internal error: upload failed"))
	req := fx.req(testName)
	req.CloudInit = "#cloud-config\nwrite_files:\n  - content: " + sentinel + "\n"
	_, err := fx.p.Create(fx.ctx, req)
	assertProviderError(t, err, "create", provider.ErrTransient)
	if strings.Contains(err.Error(), sentinel) {
		t.Errorf("error carries the cloud-init: %v", err)
	}
	lines := fx.logs.all()
	if len(lines) == 0 {
		t.Fatal("no log lines captured at V(2)")
	}
	for _, l := range lines {
		if strings.Contains(l, sentinel) {
			t.Errorf("log line carries the cloud-init: %s", l)
		}
	}
	if got := fx.logs.matching(`"cloudInitBytes"=` + strconv.Itoa(len(req.CloudInit))); len(got) != 1 {
		t.Errorf("cloudInitBytes line = %d, want 1:\n%s", len(got), strings.Join(lines, "\n"))
	}
}

func TestCreate_metadataWrittenInDefineCall(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	fx.p.now = func() time.Time { return time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) }
	fx.create(testName)
	fx.withDomain(testName, func(d *fakeDomain) {
		for _, want := range []string{`xmlns="` + metadataURI + `"`, "<uid>" + testUID + "</uid>", "<managed>true</managed>", "<createdAt>2026-09-16T12:00:00Z</createdAt>"} {
			if !strings.Contains(d.xml, want) {
				t.Errorf("defined XML lacks %s:\n%s", want, d.xml)
			}
		}
		meta, err := parseInstanceMetadata(d.metadata)
		if err != nil {
			t.Fatalf("metadata the define stored does not parse: %v", err)
		}
		if !meta.managed() || meta.UID != testUID || meta.createdAt().IsZero() {
			t.Errorf("metadata = %+v, want managed, uid %s and a createdAt", meta, testUID)
		}
	})
}

func TestCreate_seedAlwaysRebuilt(t *testing.T) {
	t.Parallel()
	fx := newFixture(t, 1)
	fx.create(testName)
	seed := seedVolumeName(testName)
	first := fx.fake.vol(seed)
	if first == nil {
		t.Fatal("no seed volume after Create")
	}
	uploads := fx.fake.callCount("StorageVolUpload")

	// A crash between the seed upload and the define leaves the volumes
	// without a domain; the next attempt must not trust the seed.
	fx.fake.mu.Lock()
	delete(fx.fake.domains, testName)
	fx.fake.mu.Unlock()
	fx.create(testName)
	second := fx.fake.vol(seed)
	if second == nil || second == first {
		t.Errorf("seed volume reused across attempts")
	}
	if got := fx.fake.callCount("StorageVolUpload"); got != uploads+1 {
		t.Errorf("StorageVolUpload calls = %d, want %d", got, uploads+1)
	}
	if len(second.content) != len(first.content) || !bytes.Contains(second.content, []byte(nocloud.SeedLabel)) {
		t.Errorf("rebuilt seed is %d bytes, first was %d; want a complete seed of the same size", len(second.content), len(first.content))
	}
}