|
| 1 | +import sys |
| 2 | + |
1 | 3 | import pytest |
2 | 4 |
|
3 | | -from _nebari.utils import JsonDiff, JsonDiffEnum, byte_unit_conversion, deep_merge |
| 5 | +from _nebari.utils import ( |
| 6 | + JsonDiff, |
| 7 | + JsonDiffEnum, |
| 8 | + byte_unit_conversion, |
| 9 | + deep_merge, |
| 10 | + run_subprocess_cmd, |
| 11 | +) |
4 | 12 |
|
5 | 13 |
|
6 | 14 | @pytest.mark.parametrize( |
@@ -136,3 +144,32 @@ def test_deep_merge_empty(): |
136 | 144 |
|
137 | 145 | result = deep_merge() |
138 | 146 | assert result == expected_result |
| 147 | + |
| 148 | + |
| 149 | +size_kb_end_args = [ |
| 150 | + (1, ""), # 1KB no newline |
| 151 | + (1, "\\n"), # 1KB with newline |
| 152 | + (64, ""), # 64KB no newline |
| 153 | + (64, "\\n"), # 64KB with newline |
| 154 | + (128, ""), # 128KB no newline |
| 155 | + (128, "\\n"), # 128KB with newline |
| 156 | +] |
| 157 | + |
| 158 | + |
| 159 | +@pytest.mark.parametrize( |
| 160 | + "size_kb,end", |
| 161 | + size_kb_end_args, |
| 162 | + ids=[ |
| 163 | + f"{params[0]}KB{'_newline' if params[1] else ''}" for params in size_kb_end_args |
| 164 | + ], |
| 165 | +) |
| 166 | +def test_run_subprocess_cmd(size_kb, end): |
| 167 | + """Test large output handling using current Python interpreter""" |
| 168 | + python_exe = sys.executable |
| 169 | + command = [python_exe, "-c", f"print('a' * {size_kb} * 1024, end='{end}')"] |
| 170 | + |
| 171 | + exit_code, output = run_subprocess_cmd( |
| 172 | + command, capture_output=True, strip_errors=True, timeout=1 |
| 173 | + ) |
| 174 | + assert exit_code == 0 |
| 175 | + assert len(output.decode()) == size_kb * 1024 + (1 if end else 0) |
0 commit comments