(s3_server, s3_bucket)
| 68 | |
| 69 | @pytest.fixture |
| 70 | def s3_example_s3fs(s3_server, s3_bucket): |
| 71 | s3fs = pytest.importorskip('s3fs') |
| 72 | |
| 73 | host, port, access_key, secret_key = s3_server['connection'] |
| 74 | fs = s3fs.S3FileSystem( |
| 75 | key=access_key, |
| 76 | secret=secret_key, |
| 77 | client_kwargs={ |
| 78 | 'endpoint_url': f'http://{host}:{port}' |
| 79 | } |
| 80 | ) |
| 81 | |
| 82 | test_path = f'{s3_bucket}/{guid()}' |
| 83 | |
| 84 | fs.mkdir(test_path) |
| 85 | yield fs, test_path |
| 86 | try: |
| 87 | fs.rm(test_path, recursive=True) |
| 88 | except FileNotFoundError: |
| 89 | pass |
| 90 | |
| 91 | |
| 92 | @pytest.fixture |
nothing calls this directly
no test coverage detected