package controllers import ( "encoding/json" "fmt" "net/http" "net/http/httptest" "os" "path/filepath" "testing" "time" "apiserver/backend" "apiserver/common" "apiserver/models" "apiserver/store" "apiserver/utils" "github.com/gin-gonic/gin" "go.uber.org/zap" ) // mockFetchFile simulates FetchFile by reading from the local filesystem. // It is installed by resetJobsState() so all output endpoint tests use it. func mockFetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*backend.FetchFileResult, error) { data, err := os.ReadFile(filepath) if err != nil { return nil, err } fileSize := uint64(len(data)) if offset >= fileSize { return &backend.FetchFileResult{FileSize: fileSize, BytesRead: 0}, nil } chunk := data[offset:] if maxBytes > 0 && uint32(len(chunk)) > maxBytes { chunk = chunk[:maxBytes] } return &backend.FetchFileResult{ Content: chunk, FileSize: fileSize, BytesRead: uint32(len(chunk)), }, nil } func resetJobsState() { store.InitStores() utils.Log = zap.NewNop() initTestAuthConfig() // ensure JWT parsing works in all output tests fetchFile = mockFetchFile } func setupTestJob(t *testing.T, tmpDir string) { t.Helper() // OutFile/ErrFile use %J placeholder; resolveOutputPath replaces %J with // the formatted job ID (FormatJobID(1001) = "0-0-0-1001" for small test IDs). outFile := filepath.Join(tmpDir, "job_%J.out") errFile := filepath.Join(tmpDir, "job_%J.err") job := models.JobInfo{ Unique_JobID: 1001, JobID: 1001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, // RUN OutFile: outFile, ErrFile: errFile, CWD: tmpDir, } store.Jobs.Put(1001, job) // Write to the resolved paths (with %J replaced by formatted ID "0-0-0-1001"). os.WriteFile(filepath.Join(tmpDir, "job_0-0-0-1001.out"), []byte("line1\nline2\nline3\nline4\nline5\n"), 0644) os.WriteFile(filepath.Join(tmpDir, "job_0-0-0-1001.err"), []byte("err1\nerr2\n"), 0644) } func setupTestRouter() *gin.Engine { gin.SetMode(gin.TestMode) r := gin.New() handler := &JobController{} r.GET("/xce/v1/jobs/:jobids/output", handler.GetJobOutput) return r } // testToken returns a signed JWT for "testuser" valid for 1 hour. // initTestAuthConfig() must be called before using this token. func testToken() string { return "Bearer " + signTestJWT("testuser", "admin", time.Now().Add(time.Hour)) } func decodeResponse(t *testing.T, w *httptest.ResponseRecorder, v interface{}) { t.Helper() t.Logf("response body: %s", w.Body.String()) if err := json.Unmarshal(w.Body.Bytes(), v); err != nil { t.Fatalf("decode response: %v", err) } } func TestGetJobOutput_TailLines(t *testing.T) { resetJobsState() tmpDir := t.TempDir() setupTestJob(t, tmpDir) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/1001/output?lines=2", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Code int `json:"code"` Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if resp.Data.Content != "line4\nline5\n" { t.Errorf("content = %q, want %q", resp.Data.Content, "line4\nline5\n") } if resp.Data.Offset != 30 { t.Errorf("offset = %d, want 30", resp.Data.Offset) } if resp.Data.Done { t.Error("done should be false for running job") } if resp.Data.JobStatus != "RUN" { t.Errorf("job_status = %q, want RUN", resp.Data.JobStatus) } } func TestGetJobOutput_FromOffset(t *testing.T) { resetJobsState() tmpDir := t.TempDir() setupTestJob(t, tmpDir) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/1001/output?offset=12", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Code int `json:"code"` Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if resp.Data.Content != "line3\nline4\nline5\n" { t.Errorf("content = %q, want %q", resp.Data.Content, "line3\nline4\nline5\n") } if resp.Data.Offset != 30 { t.Errorf("offset = %d, want 30", resp.Data.Offset) } } func TestGetJobOutput_FileNotFound(t *testing.T) { resetJobsState() tmpDir := t.TempDir() outFile := filepath.Join(tmpDir, "job_%J.out") job := models.JobInfo{ Unique_JobID: 2001, JobID: 2001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, OutFile: outFile, } store.Jobs.Put(2001, job) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/2001/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) var resp struct { Code int `json:"code"` } decodeResponse(t, w, &resp) if resp.Code != common.ERROR_READ_FILE_CODE { t.Errorf("code = %d, want %d", resp.Code, common.ERROR_READ_FILE_CODE) } } func TestGetJobOutput_NoOutputFileConfigured(t *testing.T) { resetJobsState() job := models.JobInfo{ Unique_JobID: 3001, JobID: 3001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, OutFile: "", } store.Jobs.Put(3001, job) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/3001/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) // resolveOutputPath uses the worker default (/tmp/job_0-0-0-3001_task0.out). // That file does not exist, so we expect a read error (501). var resp struct { Code int `json:"code"` } decodeResponse(t, w, &resp) if resp.Code != common.ERROR_READ_FILE_CODE { t.Errorf("code = %d, want %d", resp.Code, common.ERROR_READ_FILE_CODE) } } // TestGetJobOutput_DefaultPathFallback verifies that when OutFile is empty, // the worker's default output path is used and content is returned successfully. func TestGetJobOutput_DefaultPathFallback(t *testing.T) { resetJobsState() // Use a real-ish packed job ID: cluster=1, apiserver=0, ts=1778311965, seq=0 // FormatJobID → "1-0-1778311965-0" // WorkerDefaultOutputPath → "/tmp/job_1-0-1778311965-0_task0.out" const uniqueID = int64(0x0100000006A04BDD << 0) // MakeJobID(1,0,1778311965,0,0) // Use backend.MakeJobID directly for clarity; compute inline: // clusterID=1 at bit 56, timestamp=1778311965 at bit 20 // packed = (1 << 56) | (1778311965 << 20) const packedID = int64((uint64(1) << 56) | (uint64(1778311965) << 20)) // Install a fetchFile that reads from a tmpdir-based path instead of /tmp // to avoid touching /tmp during tests. tmpDir := t.TempDir() workerDefault := backend.WorkerDefaultOutputPath(packedID, 0) // /tmp/job_1-0-1778311965-0_task0.out formattedID := backend.FormatJobID(packedID) // "1-0-1778311965-0" localFile := filepath.Join(tmpDir, "job_"+formattedID+"_task0.out") os.WriteFile(localFile, []byte("worker output line\n"), 0644) // Override fetchFile for this test to redirect /tmp paths to tmpDir. origFetchFile := fetchFile fetchFile = func(jobid uint64, path string, offset uint64, maxBytes uint32) (*backend.FetchFileResult, error) { if path == workerDefault { path = localFile } return mockFetchFile(jobid, path, offset, maxBytes) } defer func() { fetchFile = origFetchFile }() job := models.JobInfo{ Unique_JobID: packedID, JobID: packedID, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, // RUN OutFile: "", // no OutFile — should fall back to worker default } store.Jobs.Put(uint64(packedID), job) router := setupTestRouter() w := httptest.NewRecorder() jobIDStr := fmt.Sprintf("%d", packedID) req, _ := http.NewRequest("GET", "/xce/v1/jobs/"+jobIDStr+"/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if resp.Data.Content != "worker output line\n" { t.Errorf("content = %q, want %q", resp.Data.Content, "worker output line\n") } } func TestGetJobOutput_PlaceholderResolution(t *testing.T) { resetJobsState() tmpDir := t.TempDir() // FormatJobID(4001) = "0-0-0-4001" for small test IDs outFile := filepath.Join(tmpDir, "job_0-0-0-4001.out") job := models.JobInfo{ Unique_JobID: 4001, JobID: 4001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, OutFile: filepath.Join(tmpDir, "job_%J.out"), } store.Jobs.Put(4001, job) os.WriteFile(outFile, []byte("placeholder resolved\n"), 0644) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/4001/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if resp.Data.Content != "placeholder resolved\n" { t.Errorf("content = %q, want %q", resp.Data.Content, "placeholder resolved\n") } } func TestGetJobOutput_JobNotFound(t *testing.T) { resetJobsState() router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/99999/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusNotFound { t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String()) } } func TestGetJobOutput_InvalidParams(t *testing.T) { resetJobsState() router := setupTestRouter() tests := []struct { name string url string expectCode int }{ {"invalid offset", "/xce/v1/jobs/1001/output?offset=abc", http.StatusBadRequest}, {"invalid lines zero", "/xce/v1/jobs/1001/output?lines=0", http.StatusBadRequest}, {"invalid type", "/xce/v1/jobs/1001/output?type=combined", http.StatusBadRequest}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { w := httptest.NewRecorder() req, _ := http.NewRequest("GET", tc.url, nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != tc.expectCode { t.Errorf("expected %d, got %d: %s", tc.expectCode, w.Code, w.Body.String()) } }) } } func TestLastNLines(t *testing.T) { tests := []struct { name string input []byte n int expect string }{ {"tail 2 of 5", []byte("a\nb\nc\nd\ne\n"), 2, "d\ne\n"}, {"tail all", []byte("a\nb\n"), 100, "a\nb\n"}, {"empty", []byte(""), 10, ""}, {"no trailing newline", []byte("no newline"), 5, "no newline"}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { got := lastNLines(tc.input, tc.n) if got != tc.expect { t.Errorf("lastNLines(%q, %d) = %q, want %q", tc.input, tc.n, got, tc.expect) } }) } } func TestResolveOutputPath(t *testing.T) { // FormatJobID(123) = "0-0-0-123" for small packed IDs (cluster=0, apiserver=0, ts=0, seq=123) job := models.JobInfo{ Unique_JobID: 123, JobID: 123, ArrayIdx: 2, SubmitUser: "alice", CWD: "/home/alice/work", } tests := []struct { input string expected string }{ {"/tmp/job_%J.out", "/tmp/job_0-0-0-123.out"}, {"/tmp/job_%J_%I.out", "/tmp/job_0-0-0-123_2.out"}, {"/tmp/job_%U.out", "/tmp/job_alice.out"}, {"/tmp/job_%C.out", "/tmp/job_/home/alice/work.out"}, // empty path → worker default for task 0 {"", "/tmp/job_0-0-0-123_task0.out"}, {"/tmp/fixed.out", "/tmp/fixed.out"}, } for _, tc := range tests { t.Run(tc.input, func(t *testing.T) { result := resolveOutputPath(tc.input, job) if result != tc.expected { t.Errorf("resolveOutputPath(%q) = %q, want %q", tc.input, result, tc.expected) } }) } } func TestGetJobOutput_Stderr(t *testing.T) { resetJobsState() tmpDir := t.TempDir() errFile := filepath.Join(tmpDir, "job_5001.err") job := models.JobInfo{ Unique_JobID: 5001, JobID: 5001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x02, ErrFile: errFile, } store.Jobs.Put(5001, job) os.WriteFile(errFile, []byte("stderr output\n"), 0644) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/5001/output?type=stderr&lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if resp.Data.Content != "stderr output\n" { t.Errorf("content = %q, want %q", resp.Data.Content, "stderr output\n") } } func TestGetJobOutput_DoneJob(t *testing.T) { resetJobsState() tmpDir := t.TempDir() outFile := filepath.Join(tmpDir, "job_6001.out") job := models.JobInfo{ Unique_JobID: 6001, JobID: 6001, ArrayIdx: 0, SubmitUser: "testuser", JobStatus: 0x08, // DONE OutFile: outFile, } store.Jobs.Put(6001, job) os.WriteFile(outFile, []byte("finished output\n"), 0644) router := setupTestRouter() w := httptest.NewRecorder() req, _ := http.NewRequest("GET", "/xce/v1/jobs/6001/output?lines=10", nil) req.Header.Set("Authorization", testToken()) router.ServeHTTP(w, req) if w.Code != http.StatusOK { t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) } var resp struct { Data OutputData `json:"data"` } decodeResponse(t, w, &resp) if !resp.Data.Done { t.Error("done should be true for finished job") } if resp.Data.JobStatus != "DONE" { t.Errorf("job_status = %q, want DONE", resp.Data.JobStatus) } }