orca/tests/integration/scenarios_test.go

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
}