diff --git a/go/core/test/upgrade/config_test.go b/go/core/test/upgrade/config_test.go index a08f391597..db5ee83d14 100644 --- a/go/core/test/upgrade/config_test.go +++ b/go/core/test/upgrade/config_test.go @@ -1,15 +1,62 @@ package upgrade import ( + "encoding/json" "os" "path/filepath" + "strings" "testing" "github.com/stretchr/testify/require" + configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1" + "github.com/kagent-dev/kagent/go/api/v1alpha3" kagentenv "github.com/kagent-dev/kagent/go/core/pkg/env" ) +func TestInvokeHarnesses(t *testing.T) { + manifest, err := os.ReadFile("../e2e/manifests/lifecycle.yaml.tmpl") + require.NoError(t, err) + for _, scenario := range []struct { + name string + repository string + }{ + {"Given a released controller, when preparing invocation fixtures, assert that they use its published runtimes", "ghcr.io/kagent-dev/kagent"}, + {"Given an upgraded controller, when preparing invocation fixtures, assert that they use its locally built runtimes", "localhost:5001/kagent-dev/kagent"}, + } { + t.Run(scenario.name, func(t *testing.T) { + images := &configv1.HarnessImagesConfig{ + Kagent: scenario.repository + "/golang-adk@sha256:" + strings.Repeat("a", 64), + Claude: scenario.repository + "/claude-harness@sha256:" + strings.Repeat("b", 64), + Codex: scenario.repository + "/codex-harness@sha256:" + strings.Repeat("c", 64), + } + encoded, err := invokeHarnesses(manifest, images) + require.NoError(t, err) + var fixtures v1alpha3.HarnessList + require.NoError(t, json.Unmarshal(encoded, &fixtures)) + require.Equal(t, "List", fixtures.Kind) + require.Len(t, fixtures.Items, 4) + want := map[string]string{ + "kagent": images.Kagent, "byo-adk-e2e": images.Kagent, + "claude-e2e": images.Claude, "codex-e2e": images.Codex, + } + for _, fixture := range fixtures.Items { + require.Equal(t, "Harness", fixture.Kind) + require.Contains(t, want, fixture.Name) + require.NotNil(t, fixture.Spec.Workload.Image) + require.Equal(t, want[fixture.Name], *fixture.Spec.Workload.Image) + require.Equal(t, "kagent-default", fixture.Spec.Substrate.WorkerPoolRef.Name) + delete(want, fixture.Name) + } + require.Empty(t, want) + }) + } + t.Run("Given no runtime catalog, when preparing invocation fixtures, assert that setup reports the missing images", func(t *testing.T) { + _, err := invokeHarnesses(manifest, nil) + require.ErrorContains(t, err, "installed controller must provide kagent, claude, and codex runtime images") + }) +} + func TestLoadUpgradeEnv(t *testing.T) { repoRoot := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(repoRoot, "Makefile"), nil, 0o600)) diff --git a/go/core/test/upgrade/invoke_e2e_test.go b/go/core/test/upgrade/invoke_e2e_test.go index bedff1dee3..1d1a55627a 100644 --- a/go/core/test/upgrade/invoke_e2e_test.go +++ b/go/core/test/upgrade/invoke_e2e_test.go @@ -1,8 +1,11 @@ package upgrade import ( + "bytes" "context" + "encoding/json" "fmt" + "io" "os" "os/exec" "path/filepath" @@ -11,7 +14,12 @@ import ( "time" "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + kubeyaml "k8s.io/apimachinery/pkg/util/yaml" + "sigs.k8s.io/yaml" + configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1" kagentenv "github.com/kagent-dev/kagent/go/core/pkg/env" ) @@ -52,6 +60,7 @@ func runInvokeE2E(t *testing.T, env upgradeEnv, treeGoDir, label string) { t.Helper() requireInvokeEnvironment(t, treeGoDir, label) + installInvokeHarnesses(t, env, treeGoDir) port, stop := startPortForward(t, env, controllerServiceName, controllerAPIPort) defer stop() @@ -68,6 +77,60 @@ func runInvokeE2E(t *testing.T, env upgradeEnv, treeGoDir, label string) { "[%s] selected invoke test did not run:\n%s", label, string(out)) } +// Reapply the matching release's fixtures at each invocation stage so upgrades +// and application rollbacks exercise that installation's runtime images too. +func installInvokeHarnesses(t *testing.T, env upgradeEnv, treeGoDir string) { + t.Helper() + + manifest, err := os.ReadFile(filepath.Join(treeGoDir, "core/test/e2e/manifests/lifecycle.yaml.tmpl")) + require.NoError(t, err) + controllerYAML := kubectl(t, env, time.Minute, "get", "configmap", controllerServiceName, + "-n", env.namespace, "-o", `jsonpath={.data.controller\.yaml}`) + var config struct { + HarnessImages *configv1.HarnessImagesConfig `json:"harnessImages"` + } + require.NoError(t, yaml.Unmarshal([]byte(controllerYAML), &config)) + fixtures, err := invokeHarnesses(manifest, config.HarnessImages) + require.NoError(t, err) + path := filepath.Join(t.TempDir(), "harnesses.json") + require.NoError(t, os.WriteFile(path, fixtures, 0o600)) + kubectl(t, env, time.Minute, "apply", "-f", path) +} + +func invokeHarnesses(manifest []byte, images *configv1.HarnessImagesConfig) ([]byte, error) { + if images.GetKagent() == "" || images.GetClaude() == "" || images.GetCodex() == "" { + return nil, fmt.Errorf("installed controller must provide kagent, claude, and codex runtime images") + } + rendered := strings.NewReplacer( + "${KAGENT_E2E_RUNTIME_IMAGE}", images.GetKagent(), + "${KAGENT_E2E_CLAUDE_IMAGE}", images.GetClaude(), + "${KAGENT_E2E_CODEX_IMAGE}", images.GetCodex(), + ).Replace(string(manifest)) + decoder := kubeyaml.NewYAMLOrJSONDecoder(bytes.NewBufferString(rendered), 4096) + fixtures := metav1.List{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "List"}} + for { + var resource runtime.RawExtension + if err := decoder.Decode(&resource); err != nil { + if err == io.EOF { + break + } + return nil, fmt.Errorf("decode lifecycle fixture: %w", err) + } + var metadata metav1.PartialObjectMetadata + if err := json.Unmarshal(resource.Raw, &metadata); err != nil { + return nil, fmt.Errorf("decode lifecycle fixture metadata: %w", err) + } + if metadata.Kind != "Harness" { + continue + } + switch metadata.Name { + case "kagent", "codex-e2e", "claude-e2e", "byo-adk-e2e": + fixtures.Items = append(fixtures.Items, resource) + } + } + return json.Marshal(fixtures) +} + func upgradeE2EEnv(t *testing.T, env upgradeEnv, port int) []string { t.Helper() diff --git a/go/core/test/upgrade/roundtrip_test.go b/go/core/test/upgrade/roundtrip_test.go index 7b1e043d79..28d56e7ba5 100644 --- a/go/core/test/upgrade/roundtrip_test.go +++ b/go/core/test/upgrade/roundtrip_test.go @@ -39,12 +39,20 @@ const ( postgresRole = "kagent_owner" ) -// pgTrackVersion returns the current applied migration version. +// pgTrackVersion returns the current applied migration version, or zero when +// the migration source is disabled and its tracking table does not exist. func pgTrackVersion(t *testing.T, env upgradeEnv, table string) int { t.Helper() + // PostgreSQL resolves table references before evaluating CASE, so an absent + // optional tracking table must be checked in a separate statement. + exists := pgQuery(t, env, fmt.Sprintf("SELECT to_regclass(%s) IS NOT NULL", pgQuote(kagentSchema+"."+table))) + if exists == "f" { + return 0 + } + raw := pgQuery(t, env, fmt.Sprintf( - "SELECT CASE WHEN to_regclass('%[1]s.%[2]s') IS NULL THEN 0 ELSE (SELECT COALESCE(MAX(version_id), 0) FROM %[1]s.%[2]s WHERE is_applied) END", + "SELECT COALESCE(MAX(version_id), 0) FROM %s.%s WHERE is_applied", kagentSchema, table)) return parseInt(t, raw, table+" version") } diff --git a/go/core/test/upgrade/upgrade_test.go b/go/core/test/upgrade/upgrade_test.go index df6a40c03c..4cfd1c7864 100644 --- a/go/core/test/upgrade/upgrade_test.go +++ b/go/core/test/upgrade/upgrade_test.go @@ -430,7 +430,9 @@ func pgMigrationState(t *testing.T, env upgradeEnv) postgresMigrationState { // pgMigrationStateE is the error-returning core of pgMigrationState, for use // inside require.Eventually conditions (see pgQueryE). func pgMigrationStateE(t *testing.T, env upgradeEnv) (postgresMigrationState, error) { - raw, err := pgQueryE(t, env, fmt.Sprintf("SELECT CASE WHEN to_regclass('%[1]s.schema_migrations') IS NULL THEN 0 ELSE (SELECT COALESCE(MAX(version_id), 0) FROM %[1]s.schema_migrations WHERE is_applied) END", kagentSchema)) + // Core migrations are required. Before the tracking table exists, return the + // query error so readiness polling retries until the controller creates it. + raw, err := pgQueryE(t, env, fmt.Sprintf("SELECT COALESCE(MAX(version_id), 0) FROM %s.schema_migrations WHERE is_applied", kagentSchema)) if err != nil { return postgresMigrationState{}, err }