(self)
| 269 | self.fs.open_download_stream_by_name("test") |
| 270 | |
| 271 | def test_get_version(self): |
| 272 | self.fs.upload_from_stream("test", b"foo") |
| 273 | time.sleep(0.01) |
| 274 | self.fs.upload_from_stream("test", b"bar") |
| 275 | time.sleep(0.01) |
| 276 | self.fs.upload_from_stream("test", b"baz") |
| 277 | time.sleep(0.01) |
| 278 | |
| 279 | self.assertEqual(b"foo", (self.fs.open_download_stream_by_name("test", revision=0)).read()) |
| 280 | self.assertEqual(b"bar", (self.fs.open_download_stream_by_name("test", revision=1)).read()) |
| 281 | self.assertEqual(b"baz", (self.fs.open_download_stream_by_name("test", revision=2)).read()) |
| 282 | |
| 283 | self.assertEqual(b"baz", (self.fs.open_download_stream_by_name("test", revision=-1)).read()) |
| 284 | self.assertEqual(b"bar", (self.fs.open_download_stream_by_name("test", revision=-2)).read()) |
| 285 | self.assertEqual(b"foo", (self.fs.open_download_stream_by_name("test", revision=-3)).read()) |
| 286 | |
| 287 | with self.assertRaises(NoFile): |
| 288 | self.fs.open_download_stream_by_name("test", revision=3) |
| 289 | with self.assertRaises(NoFile): |
| 290 | self.fs.open_download_stream_by_name("test", revision=-4) |
| 291 | |
| 292 | def test_upload_from_stream(self): |
| 293 | oid = self.fs.upload_from_stream("test_file", BytesIO(b"hello world"), chunk_size_bytes=1) |
nothing calls this directly
no test coverage detected