| 165 | |
| 166 | |
| 167 | def test_rotate_stream(tmp_path): |
| 168 | sa = save.Save() |
| 169 | with taddons.context(sa) as tctx: |
| 170 | tctx.configure(sa, save_stream_file=str(tmp_path / "a.txt")) |
| 171 | f1 = tflow.tflow(resp=True) |
| 172 | f2 = tflow.tflow(resp=True) |
| 173 | sa.request(f1) |
| 174 | sa.response(f1) |
| 175 | sa.request(f2) # second request already started. |
| 176 | tctx.configure(sa, save_stream_file=str(tmp_path / "b.txt")) |
| 177 | sa.response(f2) |
| 178 | sa.done() |
| 179 | |
| 180 | assert len(rd(tmp_path / "a.txt")) == 1 |
| 181 | assert len(rd(tmp_path / "b.txt")) == 1 |
| 182 | |
| 183 | |
| 184 | def test_disk_full(tmp_path, monkeypatch, capsys): |