1package tools
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "os"
8 "path/filepath"
9 "strings"
10 "testing"
11 "time"
12)
13
14// pythonTestAbs builds a platform-native absolute fixture path: on
15// Windows it is anchored to the TempDir volume so filepath.IsAbs holds
16// and Abs/Clean/Join behave like production paths.
17func pythonTestAbs(t *testing.T, elems ...string) string {
18 t.Helper()
19 base := t.TempDir()
20 if len(elems) == 0 {
21 return base
22 }
23 return filepath.Join(base, filepath.Join(elems...))
24}
25
26// volName returns the volume anchor of the test TempDir ("" on Unix,
27// e.g. "C:" on Windows) for fixtures that build rooted relatives.
28func volName(t *testing.T) string {
29 t.Helper()
30 return filepath.VolumeName(t.TempDir()) + string(os.PathSeparator)
31}
32
33func testPythonDeps() pythonResolveDeps {
34 return pythonResolveDeps{
35 GOOS: "linux",
36 CWD: "/session",
37 Getenv: func(string) string {
38 return ""
39 },
40 LookPath: func(string) (string, error) {
41 return "", errors.New("not found")
42 },
43 IsFile: func(string) bool { return false },
44 EvalSym: func(s string) (string, error) {
45 return s, nil
46 },
47 LayoutOK: func(string) bool { return true },
48 }
49}
50
51func TestPythonProbeArgvIsolated(t *testing.T) {
52 argv := pythonProbeArgv("/usr/bin/python3")
53 if len(argv) != 6 {
54 t.Fatalf("argv = %q, want 6 elements", argv)
55 }
56 if argv[0] != "/usr/bin/python3" {
57 t.Fatalf("argv[0] = %q", argv[0])
58 }
59 joined := strings.Join(argv[1:], " ")
60 for _, want := range []string{"-I", "-S", "-B", "-c"} {
61 if !strings.Contains(joined, want) {
62 t.Fatalf("argv missing %q: %q", want, joined)
63 }
64 }
65 // Bytecode suppression must be explicit. The probe code itself contains
66 // a ";" but stays a single argv element: execution is direct, never a
67 // concatenated shell command.
68 foundB := false
69 for i, a := range argv {
70 if a == "-B" {
71 foundB = true
72 }
73 if i < len(argv)-1 && (strings.Contains(a, "&&") || a == "|" || a == ";") {
74 t.Fatalf("probe argv must not contain shell metacharacters: %q", argv)
75 }
76 }
77 if !foundB {
78 t.Fatalf("probe argv missing -B: %q", argv)
79 }
80 if !strings.Contains(argv[5], "sys.version_info") || !strings.Contains(argv[5], "json.dumps") {
81 t.Fatalf("probe code must emit structured JSON: %q", argv[5])
82 }
83}
84
85func TestResolveConfiguredPathHandling(t *testing.T) {
86 // Fixture roots below use the platform-native absolute form so the
87 // expectations hold on both Unix and Windows runners.
88 session := pythonTestAbs(t, "session")
89 absConfigured := pythonTestAbs(t, "opt", "py", "bin", "python")
90 if got := resolveConfiguredPythonPath(session, absConfigured); got != absConfigured {
91 t.Fatalf("absolute = %q", got)
92 }
93 if got := resolveConfiguredPythonPath(session, filepath.Join("envs", "py", "bin", "python")); got != filepath.Join(session, "envs", "py", "bin", "python") {
94 t.Fatalf("relative = %q", got)
95 }
96 // No shell expansion: ~ and $VAR stay literal path segments.
97 if got := resolveConfiguredPythonPath(session, filepath.Join("~", "bin", "python")); strings.Contains(got, string(os.PathSeparator)+"root") || !strings.Contains(got, "~") {
98 t.Fatalf("must not expand ~: %q", got)
99 }
100 if got := resolveConfiguredPythonPath(session, "$HOME"+string(os.PathSeparator)+filepath.Join("bin", "python")); !strings.Contains(got, "$HOME") {
101 t.Fatalf("must not expand env vars: %q", got)
102 }
103 // Paths with spaces survive.
104 spacedSession := pythonTestAbs(t, "my dir")
105 if got := resolveConfiguredPythonPath(spacedSession, filepath.Join("env dir", "py")); got != filepath.Join(spacedSession, "env dir", "py") {
106 t.Fatalf("spaces = %q", got)
107 }
108 // A relative session CWD is anchored to absolute once, so relative
109 // configured paths join the intended directory instead of stacking.
110 if got := resolveConfiguredPythonPath("project", filepath.Join("envs", "py", "bin", "python")); !filepath.IsAbs(got) || !strings.HasSuffix(got, filepath.Join("project", "envs", "py", "bin", "python")) {
111 t.Fatalf("relative cwd must absolutize: %q", got)
112 }
113}
114
115func TestResolveConfiguredMissingNoFallback(t *testing.T) {
116 deps := testPythonDeps()
117 deps.Configured = "/opt/missing/python"
118 lookedUp := false
119 deps.LookPath = func(string) (string, error) {
120 lookedUp = true
121 return "/usr/bin/python3", nil
122 }
123 probed := false
124 deps.Probe = func(context.Context, string) (pythonVersion, error) {
125 probed = true
126 return pythonVersion{Major: 3}, nil
127 }
128 _, err := resolvePythonInterpreter(context.Background(), deps)
129 if err == nil || !strings.Contains(err.Error(), "configured python_interpreter") {
130 t.Fatalf("err = %v, want configured failure", err)
131 }
132 if lookedUp || probed {
133 t.Fatalf("missing configured interpreter must not fall through (lookup=%v probe=%v)", lookedUp, probed)
134 }
135}
136
137func TestResolveConfiguredPython2NoFallback(t *testing.T) {
138 deps := testPythonDeps()
139 configured := pythonTestAbs(t, "opt", "python2", "bin", "python")
140 deps.Configured = configured
141 deps.IsFile = func(s string) bool { return s == configured }
142 probes := 0
143 deps.Probe = func(_ context.Context, exe string) (pythonVersion, error) {
144 probes++
145 if exe != configured {
146 t.Fatalf("probed %q", exe)
147 }
148 return pythonVersion{}, errors.New("probe returned Python 2 (need Python 3)")
149 }
150 deps.LookPath = func(string) (string, error) {
151 t.Fatal("must not consult PATH after configured failure")
152 return "", errors.New("not found")
153 }
154 _, err := resolvePythonInterpreter(context.Background(), deps)
155 if err == nil || !strings.Contains(err.Error(), "configured python_interpreter") {
156 t.Fatalf("err = %v", err)
157 }
158 if probes != 1 {
159 t.Fatalf("probes = %d, want 1", probes)
160 }
161}
162
163func TestResolveVirtualEnvCandidateConstruction(t *testing.T) {
164 venvRoot := pythonTestAbs(t, "opt", "venv")
165 if got := pythonVenvCandidate(pythonTestAbs(t, "s"), venvRoot, "linux"); got != filepath.Join(venvRoot, "bin", "python") {
166 t.Fatalf("unix = %q", got)
167 }
168 if got := pythonVenvCandidate(pythonTestAbs(t, "s"), venvRoot, "windows"); got != filepath.Join(venvRoot, "Scripts", "python.exe") {
169 t.Fatalf("windows = %q", got)
170 }
171 if got := pythonVenvCandidate(filepath.Join(volName(t), "s cwd"), "rel env", "linux"); got != filepath.Join(volName(t), "s cwd", "rel env", "bin", "python") {
172 t.Fatalf("relative root with spaces = %q", got)
173 }
174}
175
176func TestResolveVirtualEnvTakesPriorityNoFallback(t *testing.T) {
177 deps := testPythonDeps()
178 deps.Getenv = func(k string) string {
179 if k == "VIRTUAL_ENV" {
180 return "/opt/venv"
181 }
182 return ""
183 }
184 deps.IsFile = func(s string) bool { return false }
185 _, err := resolvePythonInterpreter(context.Background(), deps)
186 if err == nil || !strings.Contains(err.Error(), "active VIRTUAL_ENV") {
187 t.Fatalf("err = %v, want VIRTUAL_ENV failure without fallback", err)
188 }
189}
190
191func TestResolveLocalVenvInvalidReported(t *testing.T) {
192 dir := t.TempDir()
193 if err := os.MkdirAll(filepath.Join(dir, ".venv"), 0o755); err != nil {
194 t.Fatal(err)
195 }
196 deps := testPythonDeps()
197 deps.CWD = dir
198 deps.LookPath = func(string) (string, error) {
199 t.Fatal("present .venv must not fall through to PATH")
200 return "", errors.New("not found")
201 }
202 _, err := resolvePythonInterpreter(context.Background(), deps)
203 if err == nil || !strings.Contains(err.Error(), ".venv") {
204 t.Fatalf("err = %v, want .venv failure", err)
205 }
206}
207
208func TestResolvePathFallbackPython2ThenPython3(t *testing.T) {
209 deps := testPythonDeps()
210 // Absolute LookPath fixtures: a rooted Unix-style path is not
211 // absolute on Windows, where Abs would prepend the drive's CWD.
212 dir := t.TempDir()
213 py3 := filepath.Join(dir, "python3")
214 py := filepath.Join(dir, "python")
215 deps.LookPath = func(name string) (string, error) {
216 if name == "python3" {
217 return py3, nil
218 }
219 if name == "python" {
220 return py, nil
221 }
222 return "", errors.New("not found")
223 }
224 deps.Probe = func(_ context.Context, exe string) (pythonVersion, error) {
225 if exe == py3 {
226 return pythonVersion{}, errors.New("probe returned Python 2 (need Python 3)")
227 }
228 return pythonVersion{Major: 3, Minor: 12, Micro: 1}, nil
229 }
230 got, err := resolvePythonInterpreter(context.Background(), deps)
231 if err != nil {
232 t.Fatal(err)
233 }
234 if got.Path != py || got.Version.String() != "3.12.1" {
235 t.Fatalf("got %+v", got)
236 }
237}
238
239func TestResolvePathMissingGivesGuidance(t *testing.T) {
240 deps := testPythonDeps()
241 _, err := resolvePythonInterpreter(context.Background(), deps)
242 if err == nil {
243 t.Fatal("want error")
244 }
245 for _, want := range []string{"existing Python 3", "python_interpreter", "never installs"} {
246 if !strings.Contains(err.Error(), want) {
247 t.Fatalf("error missing %q: %q", want, err)
248 }
249 }
250}
251
252func TestResolveCancellationStopsDiscovery(t *testing.T) {
253 deps := testPythonDeps()
254 deps.LookPath = func(name string) (string, error) {
255 return "/usr/bin/" + name, nil
256 }
257 deps.Probe = func(context.Context, string) (pythonVersion, error) {
258 t.Fatal("cancelled discovery must not probe")
259 return pythonVersion{}, nil
260 }
261 ctx, cancel := context.WithCancel(context.Background())
262 cancel()
263 _, err := resolvePythonInterpreter(ctx, deps)
264 if !errors.Is(err, context.Canceled) {
265 t.Fatalf("err = %v, want context.Canceled", err)
266 }
267}
268
269func TestResolveProbeDeadlineRespected(t *testing.T) {
270 deps := testPythonDeps()
271 deps.Configured = "/opt/py/bin/python"
272 deps.IsFile = func(string) bool { return true }
273 deps.Probe = func(ctx context.Context, _ string) (pythonVersion, error) {
274 <-ctx.Done()
275 return pythonVersion{}, ctx.Err()
276 }
277 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
278 defer cancel()
279 start := time.Now()
280 _, err := resolvePythonInterpreter(ctx, deps)
281 if err == nil {
282 t.Fatal("want deadline error")
283 }
284 if time.Since(start) > 3*time.Second {
285 t.Fatalf("probe did not respect deadline")
286 }
287}
288
289func TestLimitedWriterBoundsProbeOutput(t *testing.T) {
290 var buf bytes.Buffer
291 w := &limitedWriter{W: &buf, N: 8}
292 n, err := w.Write([]byte("1234567890abcdef"))
293 if err != nil || n != 16 {
294 t.Fatalf("Write = %d, %v, want 16, nil", n, err)
295 }
296 if buf.Len() != 8 || buf.String() != "12345678" {
297 t.Fatalf("bounded buffer = %q, want %q", buf.String(), "12345678")
298 }
299 // Further writes are dropped but report success.
300 if n, _ := w.Write([]byte("more")); n != 4 || buf.Len() != 8 {
301 t.Fatalf("overflow write = %d len %d", n, buf.Len())
302 }
303}
304
305func TestWindowsRejectsStoreAliasBothNames(t *testing.T) {
306 for _, name := range []string{"python3", "python"} {
307 deps := testPythonDeps()
308 deps.GOOS = "windows"
309 deps.CWD = `C:\proj`
310 alias := `C:\Users\test\AppData\Local\Microsoft\WindowsApps\` + name + `.exe`
311 deps.LookPath = func(n string) (string, error) {
312 if n == name {
313 return alias, nil
314 }
315 return "", errors.New("not found")
316 }
317 deps.EvalSym = func(s string) (string, error) { return s, nil }
318 deps.LayoutOK = func(string) bool { return true }
319 deps.Probe = func(context.Context, string) (pythonVersion, error) {
320 t.Fatalf("alias %q must be rejected before any probe", alias)
321 return pythonVersion{}, nil
322 }
323 _, err := resolvePythonInterpreter(context.Background(), deps)
324 if err == nil || !strings.Contains(err.Error(), "existing Python 3") {
325 t.Fatalf("%s: err = %v, want missing-interpreter fallback", name, err)
326 }
327 }
328}
329
330func TestWindowsRejectsLauncherAndManagerBeforeProbe(t *testing.T) {
331 cases := map[string]string{
332 "launcher": `C:\Windows\py.exe`,
333 "manager": `C:\Program Files\Python\python-manager.exe`,
334 "wrapper": `C:\tools\weird-python.exe`,
335 }
336 for kind, exe := range cases {
337 deps := testPythonDeps()
338 deps.GOOS = "windows"
339 deps.Configured = exe
340 deps.IsFile = func(string) bool { return true }
341 deps.EvalSym = func(s string) (string, error) { return s, nil }
342 deps.LayoutOK = func(p string) bool {
343 // Only the wrapper case reaches layout verification with false.
344 return kind == "launcher" || kind == "manager"
345 }
346 probed := false
347 deps.Probe = func(context.Context, string) (pythonVersion, error) {
348 probed = true
349 return pythonVersion{Major: 3}, nil
350 }
351 _, err := resolvePythonInterpreter(context.Background(), deps)
352 if err == nil {
353 t.Fatalf("%s %q accepted", kind, exe)
354 }
355 if probed && kind != "launcher" && kind != "manager" {
356 // Wrapper must also be rejected before probing.
357 t.Fatalf("%s %q probed an unverifiable wrapper", kind, exe)
358 }
359 if probed {
360 t.Fatalf("%s %q must be rejected before any probe", kind, exe)
361 }
362 }
363}
364
365func TestWindowsSymlinkTargetClassified(t *testing.T) {
366 deps := testPythonDeps()
367 deps.GOOS = "windows"
368 deps.Configured = `C:\tools\python.exe`
369 deps.IsFile = func(string) bool { return true }
370 deps.EvalSym = func(string) (string, error) {
371 return `C:\Users\test\AppData\Local\Microsoft\WindowsApps\python.exe`, nil
372 }
373 deps.LayoutOK = func(string) bool { return true }
374 deps.Probe = func(context.Context, string) (pythonVersion, error) {
375 t.Fatal("symlink target alias must be rejected before probe")
376 return pythonVersion{}, nil
377 }
378 _, err := resolvePythonInterpreter(context.Background(), deps)
379 if err == nil || !strings.Contains(strings.ToLower(err.Error()), "store") {
380 t.Fatalf("err = %v, want Store rejection via resolved target", err)
381 }
382}
383
384func TestWindowsNeverUsesPyLauncherFallback(t *testing.T) {
385 deps := testPythonDeps()
386 deps.GOOS = "windows"
387 seen := []string{}
388 deps.LookPath = func(name string) (string, error) {
389 seen = append(seen, name)
390 return "", errors.New("not found")
391 }
392 _, _ = resolvePythonInterpreter(context.Background(), deps)
393 for _, name := range seen {
394 if name == "py" || name == "pyw" {
395 t.Fatalf("resolver must never consult the py launcher, saw %q", seen)
396 }
397 }
398}
399
400func TestResolvePathResultAbsolutized(t *testing.T) {
401 // A relative PATH entry can yield a relative interpreter path. The
402 // probe runs in the process directory while execution runs in the
403 // session CWD, so resolution must anchor it: probing, classification,
404 // and execution must address the same executable.
405 rel := filepath.Join("rel", "bin", "python3")
406 wantAbs, err := filepath.Abs(rel)
407 if err != nil {
408 t.Fatal(err)
409 }
410 deps := testPythonDeps()
411 deps.LookPath = func(name string) (string, error) {
412 if name == "python3" {
413 return rel, nil
414 }
415 return "", errors.New("not found")
416 }
417 deps.IsFile = func(s string) bool { return s == wantAbs }
418 var probed string
419 deps.Probe = func(_ context.Context, exe string) (pythonVersion, error) {
420 probed = exe
421 return pythonVersion{Major: 3, Minor: 1, Micro: 0}, nil
422 }
423 got, err := resolvePythonInterpreter(context.Background(), deps)
424 if err != nil {
425 t.Fatal(err)
426 }
427 if !filepath.IsAbs(got.Path) || got.Path != wantAbs {
428 t.Fatalf("resolved path = %q, want absolutized %q", got.Path, wantAbs)
429 }
430 if probed != got.Path {
431 t.Fatalf("probed %q but returned %q", probed, got.Path)
432 }
433}
434
435func TestPythonChildEnvKeepsHostOnUnix(t *testing.T) {
436 t.Setenv("ZUT_PYTHON_TEST_SENTINEL", "kept")
437 found := false
438 for _, kv := range pythonChildEnv() {
439 if kv == "ZUT_PYTHON_TEST_SENTINEL=kept" {
440 found = true
441 }
442 }
443 if !found {
444 t.Fatal("host environment must be inherited")
445 }
446}
447
448func TestPythonChildEnvStripsLauncherOptInsCaseInsensitively(t *testing.T) {
449 // Windows environment keys are case-insensitive; pythonChildEnv only
450 // filters on Windows, so this test pins the matching contract via the
451 // same normalization on synthetic input.
452 input := []string{
453 "PYLAUNCHER_ALLOW_INSTALL=1",
454 "pylauncher_always_install=1",
455 "PyLauncher_Allow_Install=1",
456 "PYTHON_MANAGER_AUTOMATIC_INSTALL=1",
457 "python_manager_automatic_install=0",
458 "PATH=/usr/bin",
459 "ZUT_PYTHON_TEST_SENTINEL=kept",
460 }
461 var kept []string
462 for _, kv := range input {
463 key := kv
464 if i := strings.IndexByte(kv, '='); i >= 0 {
465 key = kv[:i]
466 }
467 switch strings.ToUpper(key) {
468 case "PYLAUNCHER_ALLOW_INSTALL", "PYLAUNCHER_ALWAYS_INSTALL", "PYTHON_MANAGER_AUTOMATIC_INSTALL":
469 continue
470 }
471 kept = append(kept, kv)
472 }
473 if len(kept) != 2 || kept[0] != "PATH=/usr/bin" || kept[1] != "ZUT_PYTHON_TEST_SENTINEL=kept" {
474 t.Fatalf("kept = %q, want only PATH and sentinel", kept)
475 }
476}