Coverage for tests/test_dav.py: 14%

1# This file is part of lsst-resources. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12import concurrent 

13import hashlib 

14import io 

15import os.path 

16import random 

17import shutil 

18import socket 

19import stat 

20import string 

21import tempfile 

22import time 

23import unittest 

24import zlib 

25from collections.abc import Callable 

26from datetime import datetime 

27from threading import Thread 

28from typing import Any, cast 

29from zipfile import ZipFile, ZipInfo 

30 

31try: 

32 from cheroot import wsgi 

33 from wsgidav.wsgidav_app import WsgiDAVApp 

34except ImportError: 

35 WsgiDAVApp = None 

36 

37try: 

38 import fsspec 

39except ImportError: 

40 fsspec = None 

41 AbstractFileSystem = type 

42 

43from lsst.resources import ResourcePath 

44from lsst.resources._resourceHandles._davResourceHandle import ( 

45 DavReadResourceHandle, 

46) 

47from lsst.resources.dav import ( 

48 DavResourcePathConfig, 

49 dav_globals, 

50) 

51from lsst.resources.davutils import ( 

52 DavConfig, 

53 DavConfigPool, 

54 TokenAuthorizer, 

55) 

56from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase 

57from lsst.resources.utils import get_tempdir, makeTestTempDir, removeTestTempDir 

58 

59TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

60 

61 

62class GenericDavTestCase(GenericTestCase, unittest.TestCase): 

63 """Generic tests of dav URIs.""" 

64 

65 scheme = "dav" 

66 netloc = "host.example.org" 

67 

68 def test_dav_root_uri(self): 

69 root_uri_test_cases = { 

70 # input : expected 

71 "dav://host.example.org": "dav://host.example.org/", 

72 "dav://host.example.org/some/path": "dav://host.example.org/", 

73 "dav://host.example.org:12345": "dav://host.example.org:12345/", 

74 "dav://host.example.org:12345/some/path": "dav://host.example.org:12345/", 

75 "dav://user:password@host.example.org": "dav://user:password@host.example.org/", 

76 "dav://user:password@host.example.org/some/path": "dav://user:password@host.example.org/", 

77 "dav://user:password@host.example.org:12345/some/path": "dav://user:password@host.example.org:12345/", 

78 "dav://user:password@host.example.org/some/path#fragment": "dav://user:password@host.example.org/", 

79 "dav://user:password@host.example.org/some/path?param=value": "dav://user:password@host.example.org/", 

80 "dav://user:password@host.example.org/some/path;parameters": "dav://user:password@host.example.org/", 

81 } 

82 for path, expected in root_uri_test_cases.items(): 

83 self.assertEqual(ResourcePath(expected), ResourcePath(path).root_uri()) 

84 

85 clean_path_test_cases = { 

86 # input : expected 

87 "dav://host.example.org/": "dav://host.example.org/", 

88 "dav://host.example.org/some/path": "dav://host.example.org/some/path", 

89 "dav://host.example.org////some/path///": "dav://host.example.org/some/path/", 

90 "dav://host.example.org/some/path///": "dav://host.example.org/some/path/", 

91 "dav://host.example.org/some/./././path": "dav://host.example.org/some/path", 

92 "dav://host.example.org/a/b/c/d/../../": "dav://host.example.org/a/b/", 

93 "dav://host.example.org/a/b/c/d/../../../../../../": "dav://host.example.org/", 

94 } 

95 for path, expected in clean_path_test_cases.items(): 

96 self.assertEqual(ResourcePath(path).geturl(), expected) 

97 

98 

99class DavReadWriteTestCase(GenericReadWriteTestCase, unittest.TestCase): 

100 """Test with a real webDAV server, as opposed to mocking responses.""" 

101 

102 scheme = "dav" 

103 local_files_to_remove: list[str] = [] 

104 MEGABYTE: int = 1024 * 1024 

105 

106 @classmethod 

107 def setUpClass(cls): 

108 cls.webdav_tmpdir = tempfile.mkdtemp(prefix="webdav-server-test-") 

109 cls.server_thread = None 

110 

111 # Reinitialize globals. 

112 dav_globals._reset() 

113 

114 # Should we test against a running server? 

115 # 

116 # This is convenient for testing against real servers in the 

117 # developer environment by initializing the environment variable 

118 # LSST_RESOURCES_DAV_TEST_SERVER_URL with the URL of the server, e.g. 

119 # dav://host.example.org:1234/path/to/top/dir 

120 if (test_endpoint := os.getenv("LSST_RESOURCES_DAV_TEST_SERVER_URL")) is not None: 

121 # Run this test case against the specified server. 

122 uri = ResourcePath(test_endpoint) 

123 cls.scheme = uri.scheme 

124 cls.netloc = uri.netloc 

125 cls.base_path = uri.path 

126 elif WsgiDAVApp is not None: 

127 # WsgiDAVApp is available, launch a local server in its own 

128 # thread to expose a local temporary directory and run this 

129 # test case against it. 

130 cls.port_number = cls._get_port_number() 

131 cls.stop_webdav_server = False 

132 cls.server_thread = Thread( 

133 target=cls._serve_webdav, 

134 args=(cls, cls.webdav_tmpdir, cls.port_number, lambda: cls.stop_webdav_server), 

135 daemon=True, 

136 ) 

137 cls.server_thread.start() 

138 

139 # Wait for it to start 

140 time.sleep(1) 

141 

142 # Initialize the server endpoint 

143 cls.netloc = f"127.0.0.1:{cls.port_number}" 

144 else: 

145 cls.skipTest( 

146 cls, 

147 "neither WsgiDAVApp is available nor a webDAV test endpoint is configured to test against", 

148 ) 

149 

150 @classmethod 

151 def tearDownClass(cls): 

152 # Stop the WsgiDAVApp server, if any 

153 if WsgiDAVApp is not None: 

154 # Shut down of the webdav server and wait for the thread to exit 

155 cls.stop_webdav_server = True 

156 if cls.server_thread is not None: 

157 cls.server_thread.join() 

158 

159 # Remove local temporary files 

160 for file in cls.local_files_to_remove: 

161 if os.path.exists(file): 

162 os.remove(file) 

163 

164 # Remove temp dir 

165 if cls.webdav_tmpdir: 

166 shutil.rmtree(cls.webdav_tmpdir, ignore_errors=True) 

167 

168 def tearDown(self): 

169 if self.tmpdir: 

170 self.tmpdir.remove_dir(recursive=True) 

171 

172 super().tearDown() 

173 

174 def test_dav_file_handle(self): 

175 # Upload a new file with known contents. 

176 contents = "These are some \n bytes to read" 

177 remote_file = self.tmpdir.join(self._get_file_name()) 

178 self.assertIsNone(remote_file.write(data=contents, overwrite=True)) 

179 

180 # Test that the correct handle is returned. 

181 with remote_file.open("rb") as handle: 

182 self.assertIsInstance(handle, DavReadResourceHandle) 

183 

184 # Test reading byte ranges works 

185 with remote_file.open("rb") as handle: 

186 sub_contents = contents[:10] 

187 handle = cast(DavReadResourceHandle, handle) 

188 result = handle.read(len(sub_contents)).decode() 

189 self.assertEqual(result, sub_contents) 

190 

191 # Verify the position. 

192 self.assertEqual(handle.tell(), len(sub_contents)) 

193 

194 # Jump back to the beginning and test if reading the whole file 

195 # prompts the internal buffer to be read. 

196 handle.seek(0) 

197 self.assertEqual(handle.tell(), 0) 

198 result = handle.read().decode() 

199 self.assertEqual(result, contents) 

200 

201 # Check that flush works on read-only handle. 

202 handle.flush() 

203 

204 # Verify reading as a string handle works as expected. 

205 with remote_file.open("r") as handle: 

206 self.assertIsInstance(handle, io.TextIOWrapper) 

207 

208 handle = cast(io.TextIOWrapper, handle) 

209 self.assertIsInstance(handle.buffer, DavReadResourceHandle) 

210 

211 # Check if string methods work. 

212 result = handle.read() 

213 self.assertEqual(result, contents) 

214 

215 # Check that flush works on read-only handle. 

216 handle.flush() 

217 

218 # Verify that write modes invoke the default base method 

219 with remote_file.open("w") as handle: 

220 self.assertIsInstance(handle, io.StringIO) 

221 

222 def test_dav_mkdir(self): 

223 # Check creation and deletion of an empty directory 

224 subdir = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

225 self.assertIsNone(subdir.mkdir()) 

226 self.assertTrue(subdir.exists()) 

227 self.assertTrue(subdir.isdir()) 

228 self.assertEqual(subdir.size(), 0) 

229 

230 # Creating an existing remote directory must succeed 

231 self.assertIsNone(subdir.mkdir()) 

232 

233 # Deleting an existing directory must succeed 

234 self.assertIsNone(subdir.remove()) 

235 

236 # Deleting a non-existing directory must succeed 

237 subir_not_exists = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

238 self.assertIsNone(subir_not_exists.remove()) 

239 

240 # Attempting to create a directory at a path where a file exists 

241 # must raise 

242 file = self.tmpdir.join(self._get_file_name(), forceDirectory=False) 

243 file.write(data=None, overwrite=True) 

244 self.assertTrue(file.exists()) 

245 

246 existing_file = self.tmpdir.join(file.basename(), forceDirectory=True) 

247 with self.assertRaises(NotADirectoryError): 

248 self.assertIsNone(existing_file.mkdir()) 

249 

250 # mkdir must create all missing ancestors 

251 subsubdir = subdir.join("a/b/c/d/e", forceDirectory=True) 

252 self.assertIsNone(subsubdir.mkdir()) 

253 self.assertTrue(subsubdir.exists()) 

254 

255 def test_dav_upload_download(self): 

256 # Test upload a randomly-generated file via write() with and without 

257 # overwrite 

258 local_file, file_size = self._generate_file() 

259 with open(local_file, "rb") as f: 

260 data = f.read() 

261 

262 remote_file = self.tmpdir.join(self._get_file_name()) 

263 self.assertIsNone(remote_file.write(data, overwrite=True)) 

264 self.assertTrue(remote_file.exists()) 

265 self.assertEqual(remote_file.size(), file_size) 

266 

267 # Write without overwrite must raise since target file exists 

268 with self.assertRaises(FileExistsError): 

269 remote_file.write(data, overwrite=False) 

270 

271 # Download the file we just uploaded. Compute and compare a digest of 

272 # the uploaded and downloaded data and ensure they match 

273 downloaded_data = remote_file.read() 

274 self.assertEqual(len(downloaded_data), file_size) 

275 upload_digest = self._compute_digest(data) 

276 download_digest = self._compute_digest(downloaded_data) 

277 self.assertEqual(upload_digest, download_digest) 

278 os.remove(local_file) 

279 

280 def test_dav_as_local(self): 

281 contents = str.encode("12345") 

282 remote_file = self.tmpdir.join(self._get_file_name()) 

283 self.assertIsNone(remote_file.write(data=contents, overwrite=True)) 

284 self.assertTrue(remote_file.exists()) 

285 self.assertEqual(remote_file.size(), len(contents)) 

286 

287 with remote_file._as_local() as local_uri: 

288 self.assertTrue(local_uri.isTemporary) 

289 self.assertTrue(os.path.exists(local_uri.ospath)) 

290 self.assertTrue(os.stat(local_uri.ospath).st_size, len(contents)) 

291 self.assertEqual(local_uri.read(), contents) 

292 self.assertFalse(local_uri.exists()) 

293 

294 def test_dav_size(self): 

295 # Retrieving the size of a non-existent file must raise. 

296 remote_file = self.tmpdir.join(self._get_file_name()) 

297 with self.assertRaises(FileNotFoundError): 

298 remote_file.size() 

299 

300 # The size of a directory using a file-like path must be zero 

301 remote_dir = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

302 self.assertIsNone(remote_dir.mkdir()) 

303 self.assertTrue(remote_dir.exists()) 

304 self.assertEqual(remote_dir.size(), 0) 

305 

306 dir_as_file = ResourcePath(remote_dir.geturl().rstrip("/"), forceDirectory=False) 

307 self.assertEqual(dir_as_file.size(), 0) 

308 

309 def test_dav_upload_creates_dir(self): 

310 # Uploading a file to a non existing directory must ensure its 

311 # parent directories are automatically created and upload succeeds 

312 non_existing_dir = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

313 non_existing_dir = non_existing_dir.join(self._get_dir_name(), forceDirectory=True) 

314 non_existing_dir = non_existing_dir.join(self._get_dir_name(), forceDirectory=True) 

315 remote_file = non_existing_dir.join(self._get_file_name()) 

316 

317 local_file, file_size = self._generate_file() 

318 with open(local_file, "rb") as f: 

319 data = f.read() 

320 self.assertIsNone(remote_file.write(data, overwrite=True)) 

321 

322 self.assertTrue(remote_file.exists()) 

323 self.assertEqual(remote_file.size(), file_size) 

324 self.assertTrue(remote_file.parent().exists()) 

325 

326 downloaded_data = remote_file.read() 

327 upload_digest = self._compute_digest(data) 

328 download_digest = self._compute_digest(downloaded_data) 

329 self.assertEqual(upload_digest, download_digest) 

330 

331 def test_dav_transfer_from(self): 

332 # Transfer from local file via "copy", with and without overwrite 

333 remote_file = self.tmpdir.join(self._get_file_name()) 

334 local_file, _ = self._generate_file() 

335 source_file = ResourcePath(local_file) 

336 self.assertIsNone(remote_file.transfer_from(source_file, transfer="copy", overwrite=True)) 

337 self.assertTrue(remote_file.exists()) 

338 self.assertEqual(remote_file.size(), source_file.size()) 

339 with self.assertRaises(FileExistsError): 

340 remote_file.transfer_from(ResourcePath(local_file), transfer="copy", overwrite=False) 

341 

342 # Transfer from remote file via "copy", with and without overwrite 

343 source_file = remote_file 

344 target_file = self.tmpdir.join(self._get_file_name()) 

345 self.assertIsNone(target_file.transfer_from(source_file, transfer="copy", overwrite=True)) 

346 self.assertTrue(target_file.exists()) 

347 self.assertEqual(target_file.size(), source_file.size()) 

348 

349 # Transfer without overwrite must raise since target resource exists 

350 with self.assertRaises(FileExistsError): 

351 target_file.transfer_from(source_file, transfer="copy", overwrite=False) 

352 

353 # Test transfer from local file via "move", with and without overwrite 

354 source_file = ResourcePath(local_file) 

355 source_size = source_file.size() 

356 target_file = self.tmpdir.join(self._get_file_name()) 

357 self.assertIsNone(target_file.transfer_from(source_file, transfer="move", overwrite=True)) 

358 self.assertTrue(target_file.exists()) 

359 self.assertEqual(target_file.size(), source_size) 

360 self.assertFalse(source_file.exists()) 

361 

362 # Test transfer without overwrite must raise since target resource 

363 # exists 

364 local_file, file_size = self._generate_file() 

365 with self.assertRaises(FileExistsError): 

366 source_file = ResourcePath(local_file) 

367 target_file.transfer_from(source_file, transfer="move", overwrite=False) 

368 

369 # Test transfer from remote file via "move" with and without overwrite 

370 # must succeed 

371 source_file = target_file 

372 source_size = source_file.size() 

373 target_file = self.tmpdir.join(self._get_file_name()) 

374 self.assertIsNone(target_file.transfer_from(source_file, transfer="move", overwrite=True)) 

375 self.assertTrue(target_file.exists()) 

376 self.assertEqual(target_file.size(), source_size) 

377 self.assertFalse(source_file.exists()) 

378 

379 # Transfer without overwrite must raise since target resource exists 

380 with self.assertRaises(FileExistsError): 

381 source_file = ResourcePath(local_file) 

382 target_file.transfer_from(source_file, transfer="move", overwrite=False) 

383 

384 def test_dav_handle(self): 

385 # Resource handle must succeed 

386 remote_file = self.tmpdir.join(self._get_file_name()) 

387 data = "abcdefghi" 

388 self.assertIsNone(remote_file.write(data, overwrite=True)) 

389 with remote_file.open("rb") as handle: 

390 handle.seek(1) 

391 self.assertEqual(handle.read(4).decode("utf-8"), data[1:5]) 

392 

393 # If we read the whole file, ensure we cache its contents 

394 with remote_file.open("rb") as handle: 

395 handle.read() 

396 self.assertIsNotNone(handle._cache) 

397 

398 handle.seek(1) 

399 self.assertEqual(handle.read(4).decode("utf-8"), data[1:5]) 

400 

401 # Upload a multi-megabyte file and ensure a partial read succeeds 

402 # without caching the entire file. 

403 data = io.BytesIO(b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09" * self.MEGABYTE) 

404 self.assertIsNone(remote_file.write(data, overwrite=True)) 

405 file_size: int = remote_file.size() 

406 file_offset: int = random.randint(self.MEGABYTE, file_size) 

407 bytes_to_read: int = random.randint(self.MEGABYTE // 2, self.MEGABYTE) 

408 with remote_file.open("rb") as handle: 

409 data.seek(file_offset) 

410 handle.seek(file_offset) 

411 self.assertEqual(handle.tell(), data.tell()) 

412 self.assertEqual(handle.read(bytes_to_read), data.read(bytes_to_read)) 

413 self.assertIsNone(handle._cache) 

414 

415 # Test readinto() 

416 with remote_file.open("rb") as handle: 

417 buffer = bytearray(random.randint(self.MEGABYTE, 2 * self.MEGABYTE)) 

418 offset = random.randint(file_size // 3, file_size // 2) 

419 

420 # Check the returned read count is as expected 

421 handle.seek(offset) 

422 count = handle.readinto(buffer) 

423 self.assertEqual(count, len(buffer)) 

424 

425 # Check the contents of the returned buffer is as expected 

426 handle.seek(offset) 

427 self.assertTrue(handle.read(count) == buffer) 

428 

429 def test_dav_repeated_write(self): 

430 data = io.BytesIO(b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09" * self.MEGABYTE) 

431 remote_file = self.tmpdir.join(self._get_file_name()) 

432 

433 # Consecutive writes to the same file must succeed. It was noticed 

434 # that XRootD server sometimes fail this operation with an error 

435 # like: 

436 # 

437 # status 423 [Output file /path/to/file is already opened by 

438 # 1 writer; open denied.] 

439 self.assertIsNone(remote_file.write(data, overwrite=True)) 

440 self.assertIsNone(remote_file.write(data, overwrite=True)) 

441 self.assertIsNone(remote_file.write(data, overwrite=True)) 

442 

443 def test_dav_remove(self): 

444 # Deletion of an existing remote file must succeed 

445 local_file, file_size = self._generate_file() 

446 with open(local_file, "rb") as f: 

447 remote_file = self.tmpdir.join(self._get_file_name()) 

448 self.assertIsNone(remote_file.write(f, overwrite=True)) 

449 

450 self.assertTrue(remote_file.exists()) 

451 self.assertEqual(remote_file.size(), file_size) 

452 self.assertIsNone(remote_file.remove()) 

453 

454 # Deletion of a non-existing remote file must succeed 

455 non_existing_file = self.tmpdir.join(self._get_file_name()) 

456 self.assertIsNone(non_existing_file.remove()) 

457 

458 # Deletion of a non-empty remote directory must raise 

459 subdir = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

460 self.assertIsNone(subdir.mkdir()) 

461 self.assertTrue(subdir.exists()) 

462 with open(local_file, "rb") as f: 

463 remote_file = subdir.join("file_to_remove") 

464 remote_file.write(f, overwrite=True) 

465 self.assertTrue(remote_file.exists()) 

466 self.assertEqual(remote_file.size(), file_size) 

467 

468 with self.assertRaises(IsADirectoryError): 

469 subdir.remove() 

470 self.assertTrue(subdir.exists()) 

471 

472 # Recursively removing a deep hierarchy of directories must succeed 

473 top = subdir 

474 for level in ["one", "two", "three", "four"]: 

475 subdir = subdir.join(level, forceDirectory=True) 

476 subdir.mkdir() 

477 self.assertTrue(subdir.exists()) 

478 self.assertTrue(subdir.isdir()) 

479 

480 top.remove_dir(recursive=True) 

481 self.assertFalse(top.exists()) 

482 

483 def test_dav_to_fsspec(self): 

484 if fsspec is None: 

485 self.skipTest("fsspec not available") 

486 

487 # Upload a randomly-generated file via write() with overwrite. 

488 local_file, file_size = self._generate_file() 

489 with open(local_file, "rb") as f: 

490 data = f.read() 

491 

492 remote_file = self.tmpdir.join(self._get_file_name()) 

493 self.assertIsNone(remote_file.write(data, overwrite=True)) 

494 self.assertTrue(remote_file.exists()) 

495 self.assertEqual(remote_file.size(), file_size) 

496 

497 # Ensure that the contents of the remote file we just 

498 # uploaded is identical to the contents of that file when 

499 # retrieved via fsspec.open(), with and without a context manager. 

500 fsys, path = remote_file.to_fsspec() 

501 file = fsys.open(path) 

502 self.assertEqual(data, file.read()) 

503 file.close() 

504 

505 with fsys.open(path) as file: 

506 self.assertEqual(data, file.read()) 

507 

508 # Ensure the file system inherits from `fsspec.AbstractFileSystem` 

509 # as parquet expects that. 

510 self.assertTrue(isinstance(fsys, fsspec.AbstractFileSystem)) 

511 

512 # Ensure properties of the remote file are consistent with those 

513 # same properties retrieved via the file system. 

514 self.assertTrue(path, remote_file.geturl()) 

515 self.assertTrue(fsys.exists(path)) 

516 self.assertTrue(fsys.isfile(path)) 

517 self.assertFalse(fsys.isdir(path)) 

518 self.assertEqual(fsys.size(path), remote_file.size()) 

519 self.assertEqual(remote_file._stat().last_modified, fsys.modified(path)) 

520 

521 info = fsys.info(path) 

522 self.assertTrue(info["name"], path) 

523 self.assertTrue(info["type"], "file") 

524 self.assertTrue(info["size"], remote_file.size()) 

525 

526 # Ensure that the file system raises with methods attempting to 

527 # modifying the file system. 

528 with self.assertRaises(NotImplementedError): 

529 fsys.rm(path) 

530 

531 # Ensure that the file system raises with methods not implemented. 

532 with self.assertRaises(NotImplementedError): 

533 fsys.fsid() 

534 

535 not_implemented_methods = [ 

536 fsys.mkdir, 

537 fsys.makedirs, 

538 fsys.rmdir, 

539 fsys.walk, 

540 fsys.find, 

541 fsys.walk, 

542 fsys.find, 

543 fsys.du, 

544 fsys.glob, 

545 fsys.rm_file, 

546 fsys.rm, 

547 fsys.touch, 

548 fsys.ukey, 

549 fsys.created, 

550 ] 

551 for method in not_implemented_methods: 

552 with self.assertRaises(NotImplementedError): 

553 method(path="xxx") 

554 

555 # Ensure that the file system raises with methods with any path 

556 # different from the only file returned by to_fsspec(). 

557 file_not_found_methods = [fsys.info, fsys.ls, fsys.modified, fsys.size] 

558 for method in file_not_found_methods: 

559 with self.assertRaises(FileNotFoundError): 

560 method(path="xxx") 

561 

562 def test_dav_parquet_read(self): 

563 # Check we can read a parquet file via to_fsspec() 

564 

565 if fsspec is None: 

566 self.skipTest("fsspec not available") 

567 

568 try: 

569 import numpy as np 

570 import pyarrow as pa # type: ignore 

571 import pyarrow.parquet as pq # type: ignore 

572 

573 # Create a local parquet file and upload it. Ensure it is bigger 

574 # than the default buffer size so that it is not entirely 

575 # cached when read by parquet library. 

576 num_rows = 1_000_000 

577 data = { 

578 "one": np.arange(num_rows, dtype=np.int32), 

579 "two": np.arange(num_rows, dtype=np.int64), 

580 "three": np.arange(num_rows, dtype=np.float64), 

581 } 

582 table = pa.Table.from_pydict(data) 

583 local_file = self._make_local_temp_file() 

584 pq.write_table(table, local_file) 

585 local_file_size = os.stat(local_file).st_size 

586 

587 remote_file = self.tmpdir.join("file.parquet") 

588 with open(local_file, "rb") as file: 

589 remote_file.write(file, overwrite=True) 

590 

591 self.assertTrue(remote_file.exists()) 

592 self.assertEqual(remote_file.size(), local_file_size) 

593 

594 # Read the remote file we just uploaded via parquet, using 

595 # similar function as used by 

596 # `lsst.daf.butler.formatters.ParquetFormatter`. 

597 fsys, path = remote_file.to_fsspec() 

598 schema = pq.read_schema(path, filesystem=fsys) 

599 for column in data.keys(): 

600 self.assertTrue(column in schema.names) 

601 

602 table = pq.read_table(path, filesystem=fsys, use_threads=True, use_pandas_metadata=False) 

603 for column in data.keys(): 

604 self.assertTrue(column in table.column_names) 

605 self.assertEqual(table.num_rows, num_rows) 

606 self.assertEqual(table.num_columns, len(data.keys())) 

607 

608 # Convert the parquet table to a Python dictionnary and compare 

609 # its contents with the data originally used to create the 

610 # parquet file. 

611 data_from_parquet = table.to_pydict() 

612 for column in data.keys(): 

613 self.assertTrue(np.array_equal(data[column], data_from_parquet[column])) 

614 

615 except ImportError: 

616 self.skipTest("numpy or pyarrow are not available") 

617 

618 def test_dav_zip(self): 

619 # Check we can read back a zip file 

620 

621 # Create a local zip file composed of a random number of identical 

622 # files. 

623 local_file, local_file_size = self._generate_file() 

624 with open(local_file, "rb") as lf: 

625 local_file_digest = self._compute_digest(lf.read()) 

626 

627 num_members = random.randint(10, 20) 

628 basename = os.path.basename(local_file) 

629 member_names = [f"{basename}-{i}" for i in range(num_members)] 

630 zip_file_name = self._make_local_temp_file() 

631 with ZipFile(zip_file_name, mode="w") as zf: 

632 for name in member_names: 

633 zf.write(local_file, name) 

634 

635 # Upload the zip file to the server 

636 with open(zip_file_name, mode="rb") as file: 

637 remote_zip_file = self.tmpdir.join("example.zip") 

638 self.assertIsNone(remote_zip_file.write(file, overwrite=True)) 

639 self.assertEqual(os.stat(zip_file_name).st_size, remote_zip_file.size()) 

640 

641 # Read the zip file back and check its contents. 

642 with remote_zip_file.open("rb") as fd: 

643 # Check the names of member files match 

644 zf = ZipFile(fd) 

645 zip_members = [info.filename for info in zf.infolist()] 

646 for name in member_names: 

647 self.assertTrue(name in zip_members) 

648 

649 # Check the sizes of member files match 

650 for file_size in [info.file_size for info in zf.infolist()]: 

651 self.assertTrue(file_size, local_file_size) 

652 

653 # Check that the contents of a randomly-selected member file 

654 # is identical to the original file. 

655 random_member = random.choice(zf.infolist()) 

656 with zf.open(random_member) as member: 

657 self.assertEqual(local_file_digest, self._compute_digest(member.read())) 

658 

659 # Concurrently read all the members of the remote zip file 

660 def download_zip_member(zfile: ZipFile, zinfo: ZipInfo) -> tuple[int, str]: 

661 # Download member of `zfile` designated by `zinfo` and return 

662 # its size in bytes and a checksum of its contents. 

663 with zfile.open(zinfo.filename) as member: 

664 data = member.read() 

665 return len(data), self._compute_digest(data) 

666 

667 with remote_zip_file.open("rb") as fd: 

668 zfile = ZipFile(fd) 

669 executor = concurrent.futures.ThreadPoolExecutor(max_workers=10) 

670 futures = [executor.submit(download_zip_member, zfile, zinfo) for zinfo in zfile.infolist()] 

671 

672 # Gather results and check they match the expected values 

673 for future in concurrent.futures.as_completed(futures): 

674 member_size, member_digest = future.result() 

675 self.assertEqual(member_size, local_file_size) 

676 self.assertEqual(member_digest, local_file_digest) 

677 

678 def test_dav_info(self): 

679 def check_metadata_fields(metadata: dict[str, Any]): 

680 for field in ("name", "size", "type", "last_modified", "checksums"): 

681 self.assertTrue(field in metadata) 

682 

683 # Retrieve and check metadata details about an non-existing object 

684 subdir = self.tmpdir.join("inexistent", forceDirectory=True) 

685 metadata = subdir.info() 

686 check_metadata_fields(metadata) 

687 self.assertEqual(metadata["size"], None) 

688 self.assertEqual(metadata["type"], None) 

689 self.assertEqual(len(metadata["checksums"]), 0) 

690 self.assertEqual(metadata["last_modified"], datetime.min) 

691 

692 # Retrieve and check metadata details about an existing directory 

693 subdir = self.tmpdir.join(self._get_dir_name(), forceDirectory=True) 

694 self.assertIsNone(subdir.mkdir()) 

695 self.assertTrue(subdir.exists()) 

696 metadata = subdir.info() 

697 check_metadata_fields(metadata) 

698 

699 self.assertEqual(metadata["size"], 0) 

700 self.assertEqual(metadata["type"], "directory") 

701 self.assertEqual(len(metadata["checksums"]), 0) 

702 self.assertEqual(metadata["last_modified"], subdir._stat().last_modified) 

703 

704 # Retrieve and check metadata details about existing file 

705 local_file, local_file_size = self._generate_file() 

706 with open(local_file, "rb") as file: 

707 content = file.read() 

708 md5_checksum = self._compute_digest(content, algorithm="md5") 

709 adler32_checksum = self._compute_digest(content, algorithm="adler32") 

710 

711 remote_file = self.tmpdir.join("example.data") 

712 with open(local_file, mode="rb") as file: 

713 self.assertIsNone(remote_file.write(file, overwrite=True)) 

714 self.assertEqual(os.stat(local_file).st_size, remote_file.size()) 

715 

716 metadata = remote_file.info() 

717 check_metadata_fields(metadata)