package integration import ( "context" "database/sql" "encoding/json" "fmt" "io" "net" "net/http" "os" "path/filepath" "strings" "testing" "time" "git.cloudinit.dev/coreci/orca/internal/acl" "git.cloudinit.dev/coreci/orca/internal/backup" "git.cloudinit.dev/coreci/orca/internal/certpaths" "git.cloudinit.dev/coreci/orca/internal/model" "git.cloudinit.dev/coreci/orca/internal/paths" "git.cloudinit.dev/coreci/orca/internal/secrets" "git.cloudinit.dev/coreci/orca/internal/store" "git.cloudinit.dev/coreci/orca/internal/transport" ) // TestScenario_NsCreate_JobSubmit_List exercises the REQ-087 core // flow end-to-end: create a namespace, submit a job into it, and list // the resulting allocations. func TestScenario_NsCreate_JobSubmit_List(t *testing.T) { h := NewHarness(t) ctx := context.Background() nsName := "webapp" nsDir := paths.NamespaceDir(nsName) for _, sub := range []string{"db", "jobs", "alloc"} { if err := os.MkdirAll(filepath.Join(nsDir, sub), 0o755); err != nil { t.Fatalf("mkdir %s/%s: %v", nsDir, sub, err) } } nsMd := renderNSMd(nsName, []string{paths.DefaultNamespace()}, true, true) if err := os.WriteFile(paths.NSMd(nsName), []byte(nsMd), 0o644); err != nil { t.Fatalf("write ns.md: %v", err) } if _, err := os.Stat(paths.NSMd(nsName)); err != nil { t.Fatalf("ns.md not present: %v", err) } jobID, err := h.SubmitJob(ctx, "webapp-1", "/bin/true") if err != nil { t.Fatalf("SubmitJob: %v", err) } jobs, err := h.Jobs.List(ctx) if err != nil { t.Fatalf("Jobs.List: %v", err) } if len(jobs) != 1 || jobs[0].ID != jobID { t.Errorf("expected 1 job %s, got %+v", jobID, jobs) } if jobs[0].Status != model.JobStatusComplete { t.Errorf("job status = %q, want complete", jobs[0].Status) } tasks, err := h.Tasks.ListByJob(ctx, jobID) if err != nil { t.Fatalf("ListByJob: %v", err) } if len(tasks) != 1 || tasks[0].ExitCode != 0 { t.Errorf("expected 1 complete task, got %+v", tasks) } } // TestScenario_Drain verifies that draining a peer stops its running // allocations and flips the node to "drained". func TestScenario_Drain(t *testing.T) { h := NewHarness(t) peer := h.RegisterPeer("drain-target", "drain-target:22") peer.SetRunning("alloc-d1", "alloc-d2", "alloc-d3") ctx := context.Background() before, err := h.ListAllocs(ctx, "drain-target") if err != nil { t.Fatalf("ListAllocs before: %v", err) } if len(before) != 3 { t.Fatalf("running before drain = %v, want 3", before) } stopped, err := h.DrainNode(ctx, "drain-target") if err != nil { t.Fatalf("DrainNode: %v", err) } if len(stopped) != 3 { t.Errorf("stopped = %v, want 3", stopped) } after, err := h.ListAllocs(ctx, "drain-target") if err != nil { t.Fatalf("ListAllocs after: %v", err) } if len(after) != 0 { t.Errorf("running after drain = %v, want empty", after) } nodes, err := h.Registry.List(ctx) if err != nil { t.Fatalf("Registry.List: %v", err) } for _, n := range nodes { if n.Name == "drain-target" && n.State != model.NodeStateDrained { t.Errorf("node state = %q, want drained", n.State) } } cmds := peer.Commands() if !containsCmd(cmds, "systemctl stop orca-alloc-alloc-d1.service") || !containsCmd(cmds, "systemctl stop orca-alloc-alloc-d2.service") || !containsCmd(cmds, "systemctl stop orca-alloc-alloc-d3.service") { t.Errorf("missing stop commands in recorded: %v", cmds) } } // TestScenario_BackupRestore verifies a backup → destroy → restore // round-trip recovers a usable ORCA_HOME. func TestScenario_BackupRestore(t *testing.T) { h := NewHarness(t) ctx := context.Background() if _, err := h.SubmitJob(ctx, "pre-backup", "/bin/true"); err != nil { t.Fatalf("SubmitJob pre-backup: %v", err) } if err := h.VerifyState(); err != nil { t.Fatalf("VerifyState before backup: %v", err) } // Checkpoint the WAL into the main db file so the backup captures // the committed job rows (the backup excludes *.db-wal sidecars). if _, err := h.DB.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)"); err != nil { t.Fatalf("wal_checkpoint: %v", err) } mk, err := secrets.LoadMasterKey(paths.MasterKeyPath()) if err != nil { t.Fatalf("LoadMasterKey: %v", err) } tarball := filepath.Join(t.TempDir(), "backup.tar.gz") if err := backup.Backup(backup.BackupOptions{ SourceDir: h.Root(), OutputPath: tarball, MasterKey: mk, }); err != nil { t.Fatalf("Backup: %v", err) } if _, err := os.Stat(tarball + ".sig"); err != nil { t.Fatalf("signature file missing: %v", err) } destroyed := h.Root() + "-destroyed" if err := os.Rename(h.Root(), destroyed); err != nil { t.Fatalf("rename to destroy ORCA_HOME: %v", err) } t.Setenv("ORCA_HOME", h.Root()) if err := backup.Restore(backup.RestoreOptions{ InputPath: tarball, TargetDir: h.Root(), MasterKey: mk, Force: true, }); err != nil { t.Fatalf("Restore: %v", err) } if err := h.VerifyState(); err != nil { t.Fatalf("VerifyState after restore: %v", err) } db, err := openDB() if err != nil { t.Fatalf("open db after restore: %v", err) } defer db.Close() jobs, err := jobList(db, ctx) if err != nil { t.Fatalf("job list after restore: %v", err) } if len(jobs) == 0 { t.Errorf("expected restored jobs, got 0") } } // TestScenario_Secrets exercises set/get/list/delete secrets across // namespaces via the secrets package directly. func TestScenario_Secrets(t *testing.T) { h := NewHarness(t) if err := h.VerifyState(); err != nil { t.Fatalf("VerifyState: %v", err) } mk, err := secrets.LoadMasterKey(paths.MasterKeyPath()) if err != nil { t.Fatalf("LoadMasterKey: %v", err) } for _, ns := range []string{"prod", "staging"} { if err := os.MkdirAll(paths.NamespaceDir(ns), 0o755); err != nil { t.Fatalf("mkdir ns %s: %v", ns, err) } } nsKeyProd, err := secrets.DeriveNamespaceKey(mk, "prod") if err != nil { t.Fatalf("DeriveNamespaceKey prod: %v", err) } enc, err := secrets.EncryptEnvFile(nsKeyProd, []string{"API_KEY=hunter2", "DB_PASS=secret"}) if err != nil { t.Fatalf("EncryptEnvFile: %v", err) } if err := writeAtomic(paths.NSSecrets("prod"), []byte(enc), 0o600); err != nil { t.Fatalf("write prod secrets: %v", err) } got, err := secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("prod")))) if err != nil { t.Fatalf("DecryptEnvFile prod: %v", err) } if len(got) != 2 { t.Errorf("prod secrets = %v, want 2 lines", got) } nsKeyStaging, err := secrets.DeriveNamespaceKey(mk, "staging") if err != nil { t.Fatalf("DeriveNamespaceKey staging: %v", err) } enc2, err := secrets.EncryptEnvFile(nsKeyStaging, []string{"TOKEN=abc"}) if err != nil { t.Fatalf("EncryptEnvFile staging: %v", err) } if err := writeAtomic(paths.NSSecrets("staging"), []byte(enc2), 0o600); err != nil { t.Fatalf("write staging secrets: %v", err) } decStaging, err := secrets.DecryptEnvFile(nsKeyStaging, string(mustReadFile(t, paths.NSSecrets("staging")))) if err != nil { t.Fatalf("DecryptEnvFile staging: %v", err) } if len(decStaging) != 1 || decStaging[0] != "TOKEN=abc" { t.Errorf("staging secret = %v, want [TOKEN=abc]", decStaging) } _, err = secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("staging")))) if err == nil { t.Error("decrypting staging with prod key succeeded; want cross-ns isolation failure") } enc3, err := secrets.EncryptEnvFile(nsKeyProd, []string{"API_KEY=hunter2"}) if err != nil { t.Fatalf("EncryptEnvFile delete: %v", err) } if err := writeAtomic(paths.NSSecrets("prod"), []byte(enc3), 0o600); err != nil { t.Fatalf("rewrite prod secrets: %v", err) } decProd, err := secrets.DecryptEnvFile(nsKeyProd, string(mustReadFile(t, paths.NSSecrets("prod")))) if err != nil { t.Fatalf("DecryptEnvFile after delete: %v", err) } if len(decProd) != 1 { t.Errorf("prod after delete = %v, want 1 line", decProd) } } // TestScenario_ACL exercises grant/check/revoke permissions and // persists the ACL state to ClusterDir()/acl.json under the harness's // temp ORCA_HOME. func TestScenario_ACL(t *testing.T) { h := NewHarness(t) if err := h.VerifyState(); err != nil { t.Fatalf("VerifyState: %v", err) } if err := os.MkdirAll(filepath.Dir(paths.ACLPath()), 0o755); err != nil { t.Fatalf("mkdir cluster dir: %v", err) } a := acl.NewACL() id := acl.Identity{Kind: acl.KindOidc, ID: "operator-1"} a.Grant(id, "prod", acl.PermRead|acl.PermWrite) if !a.Check(id, "prod", acl.PermRead) { t.Error("expected read on prod after grant") } if !a.Check(id, "prod", acl.PermWrite) { t.Error("expected write on prod after grant") } if a.Check(id, "prod", acl.PermAdmin) { t.Error("admin should not be granted") } if a.Check(id, "staging", acl.PermRead) { t.Error("cross-ns read should be denied") } admin := acl.Identity{Kind: acl.KindOidc, ID: "root"} a.Grant(admin, "prod", acl.PermAdmin) if !a.Check(admin, "prod", acl.PermRead) { t.Error("admin should imply read") } if !a.Check(admin, "prod", acl.PermWrite) { t.Error("admin should imply write") } type aclState struct { Entries []acl.ACLEntry `json:"entries"` } state := aclState{Entries: a.List()} data, err := json.MarshalIndent(state, "", " ") if err != nil { t.Fatalf("marshal acl: %v", err) } if err := writeAtomic(paths.ACLPath(), data, 0o600); err != nil { t.Fatalf("write acl.json: %v", err) } loaded, err := os.ReadFile(paths.ACLPath()) if err != nil { t.Fatalf("read acl.json: %v", err) } if !strings.Contains(string(loaded), "operator-1") { t.Errorf("acl.json missing operator-1: %s", loaded) } a.Revoke(id, "prod") if a.Check(id, "prod", acl.PermRead) { t.Error("read should be denied after revoke") } if !a.Check(admin, "prod", acl.PermAdmin) { t.Error("admin should survive revoke of operator-1") } } // TestScenario_Metrics starts the metrics HTTP endpoint, then hits // /metrics and /healthz to verify the exposition format. func TestScenario_Metrics(t *testing.T) { h := NewHarness(t) if err := h.VerifyState(); err != nil { t.Fatalf("VerifyState: %v", err) } m := transport.NewMetrics() m.SetGauge("nodes_total", 1) m.SetGauge("allocs_total", 0) mux := http.NewServeMux() mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") if err := m.WritePrometheus(w); err != nil { t.Errorf("WritePrometheus: %v", err) } }) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ok\n")) }) srv := &http.Server{Handler: mux} addr := freeAddr() srv.Addr = addr errCh := make(chan error, 1) go func() { errCh <- srv.ListenAndServe() }() t.Cleanup(func() { _ = srv.Close() }) if ok := waitListen(addr, 5*time.Second); !ok { t.Fatalf("metrics server did not start: %v", <-errCh) } resp, err := http.Get("http://" + addr + "/healthz") if err != nil { t.Fatalf("GET /healthz: %v", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { t.Errorf("/healthz status = %d, want 200", resp.StatusCode) } body, _ := io.ReadAll(resp.Body) if !strings.Contains(string(body), "ok") { t.Errorf("/healthz body = %q, want ok", body) } resp2, err := http.Get("http://" + addr + "/metrics") if err != nil { t.Fatalf("GET /metrics: %v", err) } defer resp2.Body.Close() metricsBody, _ := io.ReadAll(resp2.Body) if !strings.Contains(string(metricsBody), "orca_nodes_total") && !strings.Contains(string(metricsBody), "nodes_total") { t.Errorf("/metrics missing nodes_total: %s", metricsBody) } } // --- helpers --- func renderNSMd(name string, parents []string, inheritsEnv, inheritsSecrets bool) string { var b strings.Builder b.WriteString("---\nkind: Namespace\nname: ") b.WriteString(name) b.WriteString("\n") quoted := make([]string, len(parents)) for i, p := range parents { quoted[i] = fmt.Sprintf("%q", p) } b.WriteString("parents: [") b.WriteString(strings.Join(quoted, ", ")) b.WriteString("]\n") fmt.Fprintf(&b, "inherits_env: %t\n", inheritsEnv) fmt.Fprintf(&b, "inherits_secrets: %t\n", inheritsSecrets) b.WriteString("---\n") return b.String() } func containsCmd(cmds []string, want string) bool { for _, c := range cmds { if c == want { return true } } return false } func writeAtomic(path string, data []byte, mode os.FileMode) error { dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0o755); err != nil { return err } tmp, err := os.CreateTemp(dir, ".test-tmp-*") if err != nil { return err } tmpName := tmp.Name() defer func() { _ = os.Remove(tmpName) }() if _, err := tmp.Write(data); err != nil { _ = tmp.Close() return err } if err := tmp.Chmod(mode); err != nil { _ = tmp.Close() return err } if err := tmp.Close(); err != nil { return err } return os.Rename(tmpName, path) } func mustReadFile(t *testing.T, path string) []byte { t.Helper() b, err := os.ReadFile(path) if err != nil { t.Fatalf("read %s: %v", path, err) } return b } func openDB() (*sql.DB, error) { return store.Open(certpaths.DBPath()) } func jobList(db *sql.DB, ctx context.Context) ([]*model.Job, error) { return store.NewJobRepo(db).List(ctx) } // freeAddr returns a free localhost port for a test HTTP server. func freeAddr() string { l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { panic(fmt.Sprintf("freeAddr: %v", err)) } defer l.Close() return l.Addr().String() } // waitListen polls addr until a TCP dial succeeds or timeout elapses. func waitListen(addr string, timeout time.Duration) bool { deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { c, err := net.DialTimeout("tcp", addr, 100*time.Millisecond) if err == nil { _ = c.Close() return true } time.Sleep(20 * time.Millisecond) } return false }