Coverage for tests/test

1# This file is part of lsst-resources.

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.

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

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

12import hashlib

13import io

14import os.path

15import pickle

16import random

17import re

18import shutil

19import socket

20import stat

21import string

22import tempfile

23import time

24import unittest

25import unittest.mock

26import warnings

27from collections.abc import Callable

28from datetime import UTC

29from threading import Thread

30from typing import cast

32try:

33 from cheroot import wsgi

34 from wsgidav.wsgidav_app import WsgiDAVApp

35except ImportError:

36 WsgiDAVApp = None

38import requests

39import responses

40import responses.matchers

42import lsst.resources

43from lsst.resources import ResourceInfo, ResourcePath

44from lsst.resources._resourceHandles._httpResourceHandle import (

45 HttpReadResourceHandle,

46 parse_content_range_header,

47)

48from lsst.resources.http import (

49 BearerTokenAuth,

50 HttpResourcePath,

51 HttpResourcePathConfig,

52 SessionStore,

53 _get_dav_and_server_headers,

54 _is_protected,

55)

56from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase

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

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

62class GenericHttpTestCase(GenericTestCase, unittest.TestCase):

63 """Generic tests of http URIs."""

65 scheme = "http"

66 netloc = "server.example"

68 def test_root_uri(self):

69 self.assertEqual(ResourcePath("http://server.com").root_uri(), ResourcePath("http://server.com/"))

70 self.assertEqual(

71 ResourcePath("http://user:password@server.com:3000/").root_uri(),

72 ResourcePath("http://user:password@server.com:3000/"),

73 )

74 self.assertEqual(

75 ResourcePath("http://user:password@server.com:3000/some/path").root_uri(),

76 ResourcePath("http://user:password@server.com:3000/"),

77 )

78 self.assertEqual(

79 ResourcePath("http://user:password@server.com:3000/some/path#fragment").root_uri(),

80 ResourcePath("http://user:password@server.com:3000/"),

81 )

82 self.assertEqual(

83 ResourcePath("http://user:password@server.com:3000/some/path?param=value").root_uri(),

84 ResourcePath("http://user:password@server.com:3000/"),

85 )

86 self.assertEqual(

87 ResourcePath("http://user:password@server.com:3000/some/path;parameters").root_uri(),

88 ResourcePath("http://user:password@server.com:3000/"),

89 )

91 @responses.activate

92 def test_extra_headers(self):

93 url = "http://test.example/something.txt"

94 path = HttpResourcePath.create_http_resource_path(

95 url, extra_headers={"Authorization": "Bearer my-token"}

96 )

98 self.assertEqual(str(path), "http://test.example/something.txt")

99 self.assertEqual(path._extra_headers, {"Authorization": "Bearer my-token"})

100

101 # Make sure that headers are added to requests.

102 responses.add(

103 responses.GET,

104 url,

105 b"test",

106 match=[responses.matchers.header_matcher({"Authorization": "Bearer my-token"})],

107 )

108 self.assertEqual(path.read(), b"test")

109

110 # Make sure that headers are added to fsspec.

111 # This triggers logic for "webdav" vs "not-webdav" that does an OPTIONS

112 # request, so we need to check that too.

113 responses.add(

114 responses.OPTIONS,

115 "http://test.example/",

116 match=[responses.matchers.header_matcher({"Authorization": "Bearer my-token"})],

117 )

118 fs, _ = path.to_fsspec()

119 self.assertEqual(fs.client_kwargs.get("headers"), {"Authorization": "Bearer my-token"})

120

121 # Extra headers should be preserved through pickle, to ensure that

122 # `mtransfer` and similar methods work in multi-process mode.

123 dump = pickle.dumps(path)

124 restored = pickle.loads(dump)

125 self.assertEqual(restored._extra_headers, {"Authorization": "Bearer my-token"})

126

127 # Extra headers should be preserved when making a modified copy of the

128 # ResourcePath using replace() or the ResourcePath constructor.

129 replacement = path.replace(forceDirectory=True)

130 self.assertEqual(replacement._extra_headers, {"Authorization": "Bearer my-token"})

131 copy = ResourcePath(path, forceDirectory=True)

132 self.assertEqual(copy._extra_headers, {"Authorization": "Bearer my-token"})

133

134 @responses.activate

135 def test_get_info(self):

136 _get_dav_and_server_headers.cache_clear()

137 url = "http://test.example/something.txt"

138 responses.add(responses.OPTIONS, "http://test.example/", status=200)

139 responses.add(

140 responses.HEAD,

141 url,

142 status=200,

143 headers={

144 "Content-Length": "123",

145 "Last-Modified": "Wed, 12 Mar 2025 10:11:13 GMT",

146 "Digest": "md5=rL0Y20zC+Fzt72VPzMSk2A==, sha-256=def456",

147 },

148 )

149

150 info = ResourcePath(url).get_info()

151 self.assertIsInstance(info, ResourceInfo)

152 self.assertTrue(info.is_file)

153 self.assertEqual(info.size, 123)

154 self.assertEqual(info.last_modified.tzinfo, UTC)

155 self.assertEqual(info.last_modified.year, 2025)

156 self.assertEqual(info.checksums, {"md5": "rL0Y20zC+Fzt72VPzMSk2A==", "sha-256": "def456"})

157 self.assertEqual(len(responses.calls), 2)

158

159 @responses.activate

160 def test_open_presigned_s3_url_uses_range_requests(self):

161 """Opening a presigned S3 URL must read byte ranges on demand rather

162 than downloading the whole object.

163 """

164 _get_dav_and_server_headers.cache_clear()

165 responses.add(responses.OPTIONS, "http://s3.test/", status=200)

166

167 body = b"0123456789abcdef"

168 url = "http://s3.test/big.dat?AWSAccessKeyId=key&Signature=sig&Expires=1000"

169

170 def serve_range(request):

171 # A presigned URL is signed for GET only, so the HEAD used to probe

172 # for range support is emulated with a one-byte ranged GET. Such a

173 # request is answered with 206, never 200.

174 byte_range = request.headers.get("Range")

175 if byte_range is None: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true

176 return (200, {"Accept-Ranges": "bytes"}, body)

177 # An open-ended range such as "bytes=4-" runs to the end of body.

178 start, end = re.fullmatch(r"bytes=(\d+)-(\d*)", byte_range).groups()

179 first = int(start)

180 if first >= len(body): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 return (416, {"Accept-Ranges": "bytes"}, b"")

182 last = int(end) if end else len(body) - 1

183 chunk = body[first : last + 1]

184 return (

185 206,

186 {

187 "Accept-Ranges": "bytes",

188 "Content-Range": f"bytes {first}-{first + len(chunk) - 1}/{len(body)}",

189 },

190 chunk,

191 )

192

193 responses.add_callback(responses.GET, url, callback=serve_range)

194

195 with ResourcePath(url).open("rb") as handle:

196 self.assertIsInstance(handle, HttpReadResourceHandle)

197 handle.seek(-4, io.SEEK_END)

198 self.assertEqual(handle.read(), b"cdef")

199

200 gets = [call.request for call in responses.calls if call.request.method == "GET"]

201 self.assertTrue(gets)

202 for request in gets:

203 self.assertIn("Range", request.headers)

204

205 @responses.activate

206 def test_open_presigned_s3_url_reuses_probed_size(self):

207 """The one-byte GET used to probe for range support already reports the

208 total size, so the handle must not ask the server for it again.

209 """

210 _get_dav_and_server_headers.cache_clear()

211 responses.add(responses.OPTIONS, "http://s3.test/", status=200)

212

213 body = b"0123456789abcdef"

214 url = "http://s3.test/big.dat?AWSAccessKeyId=key&Signature=sig&Expires=1000"

215

216 def serve_range(request):

217 byte_range = request.headers.get("Range")

218 if byte_range is None: 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true

219 return (200, {"Accept-Ranges": "bytes"}, body)

220 start, end = re.fullmatch(r"bytes=(\d+)-(\d*)", byte_range).groups()

221 first = int(start)

222 last = int(end) if end else len(body) - 1

223 chunk = body[first : last + 1]

224 return (

225 206,

226 {

227 "Accept-Ranges": "bytes",

228 "Content-Range": f"bytes {first}-{first + len(chunk) - 1}/{len(body)}",

229 },

230 chunk,

231 )

232

233 responses.add_callback(responses.GET, url, callback=serve_range)

234

235 with ResourcePath(url).open("rb") as handle:

236 handle.seek(-4, io.SEEK_END)

237 self.assertEqual(handle.read(), b"cdef")

238

239 probes = [

240 call.request

241 for call in responses.calls

242 if call.request.method == "GET" and call.request.headers.get("Range") == "bytes=0-0"

243 ]

244 self.assertEqual(len(probes), 1)

245

246

247class HttpReadWriteWebdavTestCase(GenericReadWriteTestCase, unittest.TestCase):

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

249

250 scheme = "http"

251 local_files_to_remove: list[str] = []

252

253 @classmethod

254 def setUpClass(cls):

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

256 cls.server_thread = None

257

258 # Disable warnings about socket connections left open. We purposedly

259 # keep network connections to the remote server open and have no

260 # means through the API exposed by Requests of actually close the

261 # underlyng sockets to make tests pass without warning.

262 warnings.filterwarnings(action="ignore", message=r"unclosed.*socket", category=ResourceWarning)

263

264 # Should we test against a running server?

265 #

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

267 # developer environment by initializing the environment variable

268 # LSST_RESOURCES_HTTP_TEST_SERVER_URL with the URL of the server, e.g.

269 # https://dav.example.org:1234/path/to/top/dir

270 if (test_endpoint := os.getenv("LSST_RESOURCES_HTTP_TEST_SERVER_URL")) is not None: 270 ↛ 272line 270 didn't jump to line 272 because the condition on line 270 was never true

271 # Run this test case against the specified server.

272 uri = ResourcePath(test_endpoint)

273 cls.scheme = uri.scheme

274 cls.netloc = uri.netloc

275 cls.base_path = uri.path

276 elif WsgiDAVApp is not None: 276 ↛ 280line 276 didn't jump to line 280 because the condition on line 276 was never true

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

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

279 # test case against it.

280 cls.port_number = cls._get_port_number()

281 cls.stop_webdav_server = False

282 cls.server_thread = Thread(

283 target=cls._serve_webdav,

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

285 daemon=True,

286 )

287 cls.server_thread.start()

288

289 # Wait for it to start

290 time.sleep(1)

291

292 # Initialize the server endpoint

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

294 else:

295 cls.skipTest(

296 cls,

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

298 )

299

300 @classmethod

301 def tearDownClass(cls):

302 # Stop the WsgiDAVApp server, if any

303 if WsgiDAVApp is not None:

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

305 cls.stop_webdav_server = True

306 if cls.server_thread is not None:

307 cls.server_thread.join()

308

309 # Remove local temporary files

310 for file in cls.local_files_to_remove:

311 if os.path.exists(file):

312 os.remove(file)

313

314 # Remove temp dir

315 if cls.webdav_tmpdir:

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

317

318 # Reset the warnings filter.

319 warnings.resetwarnings()

320

321 def tearDown(self):

322 if self.tmpdir:

323 self.tmpdir.remove()

324

325 # Clear sessions. Some sockets may be left open, because urllib3

326 # doest not close in-flight connections.

327 # See https://urllib3.readthedocs.io > API Reference >

328 # Pool Manager > clear()

329 # I cannot add the full URL here because it is longer than 79

330 # characters.

331 self.tmpdir._clear_sessions()

332

333 super().tearDown()

334

335 def test_dav_file_handle(self):

336 # Upload a new file with known contents.

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

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

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

340

341 # Test that the correct handle is returned.

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

343 self.assertIsInstance(handle, HttpReadResourceHandle)

344

345 # Test reading byte ranges works

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

347 sub_contents = contents[:10]

348 handle = cast(HttpReadResourceHandle, handle)

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

350 self.assertEqual(result, sub_contents)

351 # Verify there is no internal buffer.

352 self.assertIsNone(handle._completeBuffer)

353 # Verify the position.

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

355

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

357 # prompts the internal buffer to be read.

358 handle.seek(0)

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

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

361 self.assertIsNotNone(handle._completeBuffer)

362 self.assertEqual(result, contents)

363

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

365 handle.flush()

366

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

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

369 self.assertIsInstance(handle, io.TextIOWrapper)

370

371 handle = cast(io.TextIOWrapper, handle)

372 self.assertIsInstance(handle.buffer, HttpReadResourceHandle)

373

374 # Check if string methods work.

375 result = handle.read()

376 self.assertEqual(result, contents)

377

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

379 handle.flush()

380

381 # Verify that write modes invoke the default base method

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

383 self.assertIsInstance(handle, io.StringIO)

384

385 def test_dav_is_dav_enpoint(self):

386 # Ensure the server is a webDAV endpoint

387 self.assertTrue(self.tmpdir.is_webdav_endpoint)

388

389 def test_dav_mkdir(self):

390 # Check creation and deletion of an empty directory

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

392 self.assertIsNone(subdir.mkdir())

393 self.assertTrue(subdir.exists())

394

395 # Creating an existing remote directory must succeed

396 self.assertIsNone(subdir.mkdir())

397

398 # Deletion of an existing directory must succeed

399 self.assertIsNone(subdir.remove())

400

401 # Deletion of an non-existing directory must succeed

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

403 self.assertIsNone(subir_not_exists.remove())

404

405 # Creation of a directory at a path where a file exists must raise

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

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

408 self.assertTrue(file.exists())

409

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

411 with self.assertRaises(NotADirectoryError):

412 self.assertIsNone(existing_file.mkdir())

413

414 def test_dav_upload_download(self):

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

416 # overwrite

417 local_file, file_size = self._generate_file()

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

419 data = f.read()

420

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

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

423 self.assertTrue(remote_file.exists())

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

425

426 # Write without overwrite must raise since target file exists

427 with self.assertRaises(FileExistsError):

428 remote_file.write(data, overwrite=False)

429

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

431 # the uploaded and downloaded data and ensure they match

432 downloaded_data = remote_file.read()

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

434 upload_digest = self._compute_digest(data)

435 download_digest = self._compute_digest(downloaded_data)

436 self.assertEqual(upload_digest, download_digest)

437 os.remove(local_file)

438

439 def test_dav_as_local(self):

440 contents = str.encode("12345")

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

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

443

444 with remote_file._as_local() as local_uri:

445 self.assertTrue(local_uri.isTemporary)

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

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

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

449 self.assertFalse(local_uri.exists())

450

451 def test_dav_size(self):

452 # Size of a non-existent file must raise.

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

454 with self.assertRaises(FileNotFoundError):

455 remote_file.size()

456

457 # Retrieving the size of a remote directory using a file-like path must

458 # raise

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

460 self.assertIsNone(remote_dir.mkdir())

461 self.assertTrue(remote_dir.exists())

462

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

464 with self.assertRaises(IsADirectoryError):

465 dir_as_file.size()

466

467 def test_dav_upload_creates_dir(self):

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

469 # parent directories are automatically created and upload succeeds

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

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

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

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

474

475 local_file, file_size = self._generate_file()

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

477 data = f.read()

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

479

480 self.assertTrue(remote_file.exists())

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

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

483

484 downloaded_data = remote_file.read()

485 upload_digest = self._compute_digest(data)

486 download_digest = self._compute_digest(downloaded_data)

487 self.assertEqual(upload_digest, download_digest)

488 os.remove(local_file)

489

490 def test_dav_transfer_from(self):

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

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

493 local_file, _ = self._generate_file()

494 source_file = ResourcePath(local_file)

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

496 self.assertTrue(remote_file.exists())

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

498 with self.assertRaises(FileExistsError):

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

500

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

502 source_file = remote_file

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

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

505 self.assertTrue(target_file.exists())

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

507

508 # Transfer without overwrite must raise since target resource exists

509 with self.assertRaises(FileExistsError):

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

511

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

513 source_file = ResourcePath(local_file)

514 source_size = source_file.size()

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

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

517 self.assertTrue(target_file.exists())

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

519 self.assertFalse(source_file.exists())

520

521 # Test transfer without overwrite must raise since target resource

522 # exists

523 local_file, file_size = self._generate_file()

524 with self.assertRaises(FileExistsError):

525 source_file = ResourcePath(local_file)

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

527

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

529 # must succeed

530 source_file = target_file

531 source_size = source_file.size()

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

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

534 self.assertTrue(target_file.exists())

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

536 self.assertFalse(source_file.exists())

537

538 # Transfer without overwrite must raise since target resource exists

539 with self.assertRaises(FileExistsError):

540 source_file = ResourcePath(local_file)

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

542

543 def test_dav_handle(self):

544 # Resource handle must succeed

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

546 data = "abcdefghi"

547 self.assertIsNone(target_file.write(data, overwrite=True))

548 with target_file.open("rb") as handle:

549 handle.seek(1)

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

551

552 def test_dav_delete(self):

553 # Deletion of an existing remote file must succeed

554 local_file, file_size = self._generate_file()

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

556 data = f.read()

557

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

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

560 self.assertTrue(remote_file.exists())

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

562 self.assertIsNone(remote_file.remove())

563 os.remove(local_file)

564

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

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

567 self.assertIsNone(non_existing_file.remove())

568

569 # Deletion of a non-empty remote directory must succeed

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

571 self.assertIsNone(subdir.mkdir())

572 self.assertTrue(subdir.exists())

573 local_file, _ = self._generate_file()

574 source_file = ResourcePath(local_file)

575 target_file = self.tmpdir.join(self._get_file_name(), forceDirectory=True)

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

577 self.assertIsNone(subdir.remove())

578 self.assertFalse(subdir.exists())

579 os.remove(local_file)

580

581 def test_dav_to_fsspec(self):

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

583 local_file, file_size = self._generate_file()

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

585 data = f.read()

586

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

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

589 self.assertTrue(remote_file.exists())

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

591 remote_file_url = remote_file.geturl()

592

593 # to_fsspec() may raise if that feature is not specifically

594 # enabled in the environment and remote server is one of the

595 # webDAV servers that support signing URLs.

596 with unittest.mock.patch.dict(os.environ, {}, clear=True):

597 try:

598 # Force reinitialization of the config from the environment

599 HttpResourcePath._reload_config()

600 fsys, url = ResourcePath(remote_file_url).to_fsspec()

601 self.assertEqual(data, fsys.cat(url))

602 except ImportError as e:

603 self.assertTrue("disable" in str(e))

604

605 # Ensure to_fsspec() works if that feature is enabled in the

606 # environment.

607 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_ENABLE_FSSPEC": "true"}, clear=True):

608 try:

609 # Force reinitialization of the config from the environment.

610 HttpResourcePath._reload_config()

611 rpath = ResourcePath(remote_file_url)

612

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

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

615 # retrieved via fsspec.open().

616 fsys, url = rpath.to_fsspec()

617 with fsys.open(url) as f:

618 self.assertEqual(data, f.read())

619

620 # Ensure the contents is identical to the result of

621 # fsspec.cat()

622 self.assertEqual(data, fsys.cat(url))

623

624 # Ensure that attempting to modify a remote via via fsspec

625 # fails, since the returned URL is signed for download only.

626 # fsspec.rm() raises NotImplementedError if it cannot remove

627 # the remote file.

628 if rpath.server_signs_urls:

629 with self.assertRaises(NotImplementedError):

630 fsys, url = rpath.to_fsspec()

631 fsys.rm(url)

632 except NotImplementedError as e:

633 # to_fsspec() must succeed if remote server knows how to

634 # sign URLs

635 if rpath.server_signs_urls:

636 raise e

637

638 # Force reinitialization of the config from the environment and

639 # clean up local file.

640 HttpResourcePath._reload_config()

641 os.remove(local_file)

642

643 @responses.activate

644 def test_is_webdav_endpoint(self):

645 davEndpoint = "http://www.lsstwithwebdav.org"

646 responses.add(responses.OPTIONS, davEndpoint, status=200, headers={"DAV": "1,2,3"})

647 self.assertTrue(ResourcePath(davEndpoint).is_webdav_endpoint)

648

649 plainHttpEndpoint = "http://www.lsstwithoutwebdav.org"

650 responses.add(responses.OPTIONS, plainHttpEndpoint, status=200)

651 self.assertFalse(ResourcePath(plainHttpEndpoint).is_webdav_endpoint)

652

653 notWebdavEndpoint = "http://www.notwebdav.org"

654 responses.add(responses.OPTIONS, notWebdavEndpoint, status=403)

655 self.assertFalse(ResourcePath(notWebdavEndpoint).is_webdav_endpoint)

656

657 @responses.activate

658 def test_plain_http_url_signing(self):

659 # As in test_is_webdav_endpoint above, configure a URL to appear as a

660 # non-webdav HTTP server.

661 plainHttpEndpoint = "http://nonwebdav.test"

662 responses.add(responses.OPTIONS, plainHttpEndpoint, status=200)

663

664 # Plain HTTP URLs are already readable without authentication, so

665 # generating a pre-signed URL is a no-op.

666 path = ResourcePath("http://nonwebdav.test/file#frag")

667 self.assertEqual(

668 path.generate_presigned_get_url(expiration_time_seconds=300), "http://nonwebdav.test/file#frag"

669 )

670

671 # Writing to an arbitrary plain HTTP URL is unlikely to work, so we

672 # don't generate put URLs.

673 with self.assertRaises(NotImplementedError):

674 path.generate_presigned_put_url(expiration_time_seconds=300)

675

676 @responses.activate

677 def test_server_identity(self):

678 server = "MyServer/v1.2.3"

679 endpointWithServer = "http://www.lsstwithserverheader.org"

680 responses.add(responses.OPTIONS, endpointWithServer, status=200, headers={"Server": server})

681 self.assertEqual(ResourcePath(endpointWithServer).server, "myserver")

682

683 endpointWithoutServer = "http://www.lsstwithoutserverheader.org"

684 responses.add(responses.OPTIONS, endpointWithoutServer, status=200)

685 self.assertIsNone(ResourcePath(endpointWithoutServer).server)

686

687 @classmethod

688 def _get_port_number(cls) -> int:

689 """Return a port number the webDAV server can use to listen to."""

690 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

691 s.bind(("127.0.0.1", 0))

692 s.listen()

693 port = s.getsockname()[1]

694 s.close()

695 return port

696

697 def _serve_webdav(self, local_path: str, port: int, stop_webdav_server: Callable[[], bool]):

698 """Start a local webDAV server, listening on http://localhost:port

699 and exposing local_path.

700

701 This server only runs when this test class is instantiated,

702 and then shuts down. The server must be started is a separate thread.

703

704 Parameters

705 ----------

706 port : `int`

707 The port number on which the server should listen

708 local_path : `str`

709 Path to an existing local directory for the server to expose.

710 stop_webdav_server : `Callable[[], bool]`

711 Boolean function which returns True when the server should be

712 stopped.

713 """

714 try:

715 # Start the wsgi server in a separate thread

716 config = {

717 "host": "127.0.0.1",

718 "port": port,

719 "provider_mapping": {"/": local_path},

720 "http_authenticator": {"domain_controller": None},

721 "simple_dc": {"user_mapping": {"*": True}},

722 "verbose": 0,

723 "lock_storage": False,

724 "dir_browser": {

725 "enable": False,

726 "ms_sharepoint_support": False,

727 "libre_office_support": False,

728 "response_trailer": False,

729 "davmount_links": False,

730 },

731 }

732 server = wsgi.Server(wsgi_app=WsgiDAVApp(config), bind_addr=(config["host"], config["port"]))

733 t = Thread(target=server.start, daemon=True)

734 t.start()

735

736 # Shut down the server when done: stop_webdav_server() returns

737 # True when this test suite is being teared down

738 while not stop_webdav_server():

739 time.sleep(1)

740 except KeyboardInterrupt:

741 # Caught Ctrl-C, shut down the server

742 pass

743 finally:

744 server.stop()

745 t.join()

746

747 @classmethod

748 def _get_name(cls, prefix: str) -> str:

749 alphabet = string.ascii_lowercase + string.digits

750 return f"{prefix}-" + "".join(random.choices(alphabet, k=8))

751

752 @classmethod

753 def _get_dir_name(cls) -> str:

754 """Return a randomly selected name for a file"""

755 return cls._get_name(prefix="dir")

756

757 @classmethod

758 def _get_file_name(cls) -> str:

759 """Return a randomly selected name for a file"""

760 return cls._get_name(prefix="file")

761

762 def _generate_file(self, remove_when_done=True) -> tuple[str, int]:

763 """Create a local file of random size with random contents.

764

765 Returns

766 -------

767 path : `str`

768 Path to local temporary file. The caller is responsible for

769 removing the file when appropriate.

770 size : `int`

771 Size of the generated file, in bytes.

772 """

773 megabyte = 1024 * 1024

774 size = random.randint(2 * megabyte, 5 * megabyte)

775 tmpfile, path = tempfile.mkstemp()

776 self.assertEqual(os.write(tmpfile, os.urandom(size)), size)

777 os.close(tmpfile)

778

779 if remove_when_done:

780 HttpReadWriteWebdavTestCase.local_files_to_remove.append(path)

781

782 return path, size

783

784 @classmethod

785 def _compute_digest(cls, data: bytes) -> str:

786 """Compute a SHA256 hash of data."""

787 m = hashlib.sha256()

788 m.update(data)

789 return m.hexdigest()

790

791 @classmethod

792 def _is_server_running(cls, port: int) -> bool:

793 """Return True if there is a server listening on local address

794 127.0.0.1:<port>.

795 """

796 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:

797 try:

798 s.connect(("127.0.0.1", port))

799 return True

800 except ConnectionRefusedError:

801 return False

802

803

804class HttpResourcePathConfigTestCase(unittest.TestCase):

805 """Test for the HttpResourcePathConfig class."""

806

807 def setUp(self):

808 self.tmpdir = ResourcePath(makeTestTempDir(TESTDIR))

809

810 def tearDown(self):

811 if self.tmpdir and self.tmpdir.isLocal: 811 ↛ exitline 811 didn't return from function 'tearDown' because the condition on line 811 was always true

812 removeTestTempDir(self.tmpdir.ospath)

813

814 def test_send_expect_header(self):

815 # Ensure environment variable LSST_HTTP_PUT_SEND_EXPECT_HEADER is

816 # inspected to initialize the HttpResourcePathConfig class.

817 with unittest.mock.patch.dict(os.environ, {}, clear=True):

818 config = HttpResourcePathConfig()

819 self.assertFalse(config.send_expect_on_put)

820

821 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_PUT_SEND_EXPECT_HEADER": "true"}, clear=True):

822 config = HttpResourcePathConfig()

823 self.assertTrue(config.send_expect_on_put)

824

825 def test_enable_fsspec(self):

826 # Ensure environment variable LSST_HTTP_ENABLE_FSSPEC is

827 # inspected to initialize the HttpResourcePathConfig class.

828 with unittest.mock.patch.dict(os.environ, {}, clear=True):

829 config = HttpResourcePathConfig()

830 self.assertFalse(config.fsspec_is_enabled)

831

832 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_ENABLE_FSSPEC": "any value"}, clear=True):

833 config = HttpResourcePathConfig()

834 self.assertTrue(config.fsspec_is_enabled)

835

836 def test_collect_memory_usage(self):

837 # Ensure environment variable LSST_HTTP_COLLECT_MEMORY_USAGE is

838 # inspected to initialize the HttpResourcePathConfig class.

839 with unittest.mock.patch.dict(os.environ, {}, clear=True):

840 config = HttpResourcePathConfig()

841 self.assertFalse(config.collect_memory_usage)

842

843 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_COLLECT_MEMORY_USAGE": "true"}, clear=True):

844 config = HttpResourcePathConfig()

845 self.assertTrue(config.collect_memory_usage)

846

847 def test_timeout(self):

848 # Ensure that when the connect and read timeouts are not specified

849 # the default values are stored in the config.

850 with unittest.mock.patch.dict(os.environ, {}, clear=True):

851 config = HttpResourcePathConfig()

852 self.assertAlmostEqual(config.timeout[0], config.DEFAULT_TIMEOUT_CONNECT)

853 self.assertAlmostEqual(config.timeout[1], config.DEFAULT_TIMEOUT_READ)

854

855 # Ensure that when both the connect and read timeouts are specified

856 # they are both stored in the config.

857 connect_timeout, read_timeout = 100.5, 200.8

858 with unittest.mock.patch.dict(

859 os.environ,

860 {"LSST_HTTP_TIMEOUT_CONNECT": str(connect_timeout), "LSST_HTTP_TIMEOUT_READ": str(read_timeout)},

861 clear=True,

862 ):

863 config = HttpResourcePathConfig()

864 self.assertAlmostEqual(config.timeout[0], connect_timeout)

865 self.assertAlmostEqual(config.timeout[1], read_timeout)

866

867 # Ensure that invalid float values (including NaN values) raise a

868 # ValueError.

869 for value in ("invalid", "NaN"):

870 with unittest.mock.patch.dict(

871 os.environ,

872 {"LSST_HTTP_TIMEOUT_CONNECT": value, "LSST_HTTP_TIMEOUT_READ": value},

873 clear=True,

874 ):

875 with self.assertRaises(ValueError):

876 config = HttpResourcePathConfig()

877 config.timeout()

878

879 def test_front_end_connections(self):

880 # Ensure that when the number of front end connections is not specified

881 # the default comes from the number of workers..

882 with unittest.mock.patch.dict(os.environ, {}, clear=True):

883 config = HttpResourcePathConfig()

884 self.assertEqual(config.front_end_connections, _get_num_workers())

885

886 # Ensure that when the number of front end connections is specified

887 # it is stored in the config.

888 connections = 42

889 with unittest.mock.patch.dict(

890 os.environ, {"LSST_HTTP_FRONTEND_PERSISTENT_CONNECTIONS": str(connections)}, clear=True

891 ):

892 config = HttpResourcePathConfig()

893 self.assertTrue(config.front_end_connections, connections)

894

895 def test_back_end_connections(self):

896 # Ensure that when the number of back end connections is not specified

897 # the default comes from the number of workers.

898 with unittest.mock.patch.dict(os.environ, {}, clear=True):

899 config = HttpResourcePathConfig()

900 self.assertEqual(config.back_end_connections, _get_num_workers())

901

902 # Ensure that when the number of back end connections is specified

903 # it is stored in the config.

904 connections = 42

905 with unittest.mock.patch.dict(

906 os.environ, {"LSST_HTTP_BACKEND_PERSISTENT_CONNECTIONS": str(connections)}, clear=True

907 ):

908 config = HttpResourcePathConfig()

909 self.assertTrue(config.back_end_connections, connections)

910

911 def test_digest_algorithm(self):

912 # Ensure that when no digest is specified in the environment, the

913 # configured digest algorithm is the empty string.

914 with unittest.mock.patch.dict(os.environ, {}, clear=True):

915 config = HttpResourcePathConfig()

916 self.assertEqual(config.digest_algorithm, "")

917

918 # Ensure that an invalid digest algorithm is ignored.

919 digest = "invalid"

920 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_DIGEST": digest}, clear=True):

921 config = HttpResourcePathConfig()

922 self.assertEqual(config.digest_algorithm, "")

923

924 # Ensure that an accepted digest algorithm is stored.

925 for digest in HttpResourcePathConfig().ACCEPTED_DIGESTS:

926 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_DIGEST": digest}, clear=True):

927 config = HttpResourcePathConfig()

928 self.assertTrue(config.digest_algorithm, digest)

929

930 def test_backoff_interval(self):

931 # Ensure that when no backoff interval is defined, the default values

932 # are used.

933 with unittest.mock.patch.dict(os.environ, {}, clear=True):

934 config = HttpResourcePathConfig()

935 self.assertAlmostEqual(config.backoff_min, config.DEFAULT_BACKOFF_MIN)

936 self.assertAlmostEqual(config.backoff_max, config.DEFAULT_BACKOFF_MAX)

937

938 # Ensure that an invalid value for backoff interval is ignored and

939 # the default value is used.

940 with unittest.mock.patch.dict(

941 os.environ, {"LSST_HTTP_BACKOFF_MIN": "XXX", "LSST_HTTP_BACKOFF_MAX": "YYY"}, clear=True

942 ):

943 config = HttpResourcePathConfig()

944 self.assertAlmostEqual(config.backoff_min, config.DEFAULT_BACKOFF_MIN)

945 self.assertAlmostEqual(config.backoff_max, config.DEFAULT_BACKOFF_MAX)

946

947 # Ensure that NaN values are ignored and the defaults values are used.

948 with unittest.mock.patch.dict(

949 os.environ, {"LSST_HTTP_BACKOFF_MIN": "NaN", "LSST_HTTP_BACKOFF_MAX": "NaN"}, clear=True

950 ):

951 config = HttpResourcePathConfig()

952 self.assertAlmostEqual(config.backoff_min, config.DEFAULT_BACKOFF_MIN)

953 self.assertAlmostEqual(config.backoff_max, config.DEFAULT_BACKOFF_MAX)

954

955 # Ensure that when specified, valid limits backoff interval are used.

956 backoff_min, backoff_max = 3.0, 8.0

957 with unittest.mock.patch.dict(

958 os.environ,

959 {"LSST_HTTP_BACKOFF_MIN": str(backoff_min), "LSST_HTTP_BACKOFF_MAX": str(backoff_max)},

960 clear=True,

961 ):

962 config = HttpResourcePathConfig()

963 self.assertAlmostEqual(config.backoff_min, backoff_min)

964 self.assertAlmostEqual(config.backoff_max, backoff_max)

965

966 def test_ca_bundle(self):

967 # Ensure that when no bundle is defined via environment variable

968 # LSST_HTTP_CACERT_BUNDLE either None is returned or the returned

969 # path does exist.

970 with unittest.mock.patch.dict(os.environ, {}, clear=True):

971 config = HttpResourcePathConfig()

972 if config.ca_bundle is not None: 972 ↛ 973line 972 didn't jump to line 973 because the condition on line 972 was never true

973 self.assertTrue(os.path.exists(config.ca_bundle))

974

975 # Ensure that if LSST_HTTP_CACERT_BUNDLE is specified, the returned

976 # path is identical to the value of that variable (we don't check

977 # here that the path actually exists).

978 ca_bundle = "/path/to/bundle/dir"

979 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_CACERT_BUNDLE": ca_bundle}, clear=True):

980 config = HttpResourcePathConfig()

981 self.assertEqual(config.ca_bundle, ca_bundle)

982

983 def test_client_token(self):

984 # Ensure that when no token is defined via environment variable

985 # LSST_HTTP_AUTH_BEARER_TOKEN None is returned.

986 with unittest.mock.patch.dict(os.environ, {}, clear=True):

987 config = HttpResourcePathConfig()

988 self.assertIsNone(config.client_token)

989

990 # Ensure that if LSST_HTTP_AUTH_BEARER_TOKEN is specified, the returned

991 # path is identical to the value of that variable (we don't check

992 # here that the path actually exists).

993 token = "ABCDE12345"

994 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_BEARER_TOKEN": token}, clear=True):

995 config = HttpResourcePathConfig()

996 self.assertEqual(config.client_token, token)

997

998 def test_client_cert_key(self):

999 """Ensure if user certificate and private key are provided via

1000 environment variables, the configuration is correctly configured.

1001 """

1002 # Ensure that when no client certificate nor private key are provided

1003 # via environment variables, both certificate and key are None.

1004 with unittest.mock.patch.dict(os.environ, {}, clear=True):

1005 config = HttpResourcePathConfig()

1006 cert, key = config.client_cert_key

1007 self.assertIsNone(cert)

1008 self.assertIsNone(key)

1009

1010 # Create mock certificate and private key files.

1011 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1012 f.write("CERT")

1013 client_cert = f.name

1014

1015 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1016 f.write("KEY")

1017 client_key = f.name

1018

1019 # Check that if only LSST_HTTP_AUTH_CLIENT_CERT is initialized

1020 # an exception is raised.

1021 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_CLIENT_CERT": client_cert}, clear=True):

1022 with self.assertRaises(ValueError):

1023 HttpResourcePathConfig().client_cert_key

1024

1025 # Check that if only LSST_HTTP_AUTH_CLIENT_KEY is initialized

1026 # an exception is raised.

1027 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_CLIENT_KEY": client_key}, clear=True):

1028 with self.assertRaises(ValueError):

1029 HttpResourcePathConfig().client_cert_key

1030

1031 # Check that the private key file must be accessible only by its owner.

1032 with unittest.mock.patch.dict(

1033 os.environ,

1034 {"LSST_HTTP_AUTH_CLIENT_CERT": client_cert, "LSST_HTTP_AUTH_CLIENT_KEY": client_key},

1035 clear=True,

1036 ):

1037 # Ensure the client certificate is initialized when only the owner

1038 # can read the private key file.

1039 os.chmod(client_key, stat.S_IRUSR)

1040 config = HttpResourcePathConfig()

1041 cert, key = config.client_cert_key

1042 self.assertEqual(cert, client_cert)

1043 self.assertEqual(key, client_key)

1044

1045 # Ensure an exception is raised if either group or other can access

1046 # the private key file.

1047 for mode in (stat.S_IRGRP, stat.S_IWGRP, stat.S_IXGRP, stat.S_IROTH, stat.S_IWOTH, stat.S_IXOTH):

1048 os.chmod(client_key, stat.S_IRUSR | mode)

1049 with self.assertRaises(PermissionError):

1050 HttpResourcePathConfig().client_cert_key

1051

1052 # Check that if environment variable X509_USER_PROXY is initialized

1053 # the configuration uses its value as the client's certificate and key.

1054 with unittest.mock.patch.dict(os.environ, {"X509_USER_PROXY": client_cert}, clear=True):

1055 config = HttpResourcePathConfig()

1056 cert, key = config.client_cert_key

1057 self.assertEqual(cert, client_cert)

1058 self.assertEqual(key, client_cert)

1059

1060

1061class WebdavUtilsTestCase(unittest.TestCase):

1062 """Test for the Webdav related utilities."""

1063

1064 def setUp(self):

1065 self.tmpdir = ResourcePath(makeTestTempDir(TESTDIR))

1066

1067 def tearDown(self):

1068 if self.tmpdir and self.tmpdir.isLocal: 1068 ↛ exitline 1068 didn't return from function 'tearDown' because the condition on line 1068 was always true

1069 removeTestTempDir(self.tmpdir.ospath)

1070

1071 def test_is_protected(self):

1072 self.assertFalse(_is_protected("/this-file-does-not-exist"))

1073

1074 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1075 f.write("XXXX")

1076 file_path = f.name

1077

1078 os.chmod(file_path, stat.S_IRUSR)

1079 self.assertTrue(_is_protected(file_path))

1080

1081 for mode in (stat.S_IRGRP, stat.S_IWGRP, stat.S_IXGRP, stat.S_IROTH, stat.S_IWOTH, stat.S_IXOTH):

1082 os.chmod(file_path, stat.S_IRUSR | mode)

1083 self.assertFalse(_is_protected(file_path))

1084

1085

1086class BearerTokenAuthTestCase(unittest.TestCase):

1087 """Test for the BearerTokenAuth class."""

1088

1089 def setUp(self):

1090 self.tmpdir = ResourcePath(makeTestTempDir(TESTDIR))

1091 self.token = "ABCDE1234"

1092

1093 def tearDown(self):

1094 if self.tmpdir and self.tmpdir.isLocal: 1094 ↛ exitline 1094 didn't return from function 'tearDown' because the condition on line 1094 was always true

1095 removeTestTempDir(self.tmpdir.ospath)

1096

1097 def test_empty_token(self):

1098 """Ensure that when no token is provided the request is not

1099 modified.

1100 """

1101 auth = BearerTokenAuth(None)

1102 auth._refresh()

1103 self.assertIsNone(auth._token)

1104 self.assertIsNone(auth._path)

1105 req = requests.Request("GET", "https://example.org")

1106 self.assertEqual(auth(req), req)

1107

1108 def test_token_value(self):

1109 """Ensure that when a token value is provided, the 'Authorization'

1110 header is added to the requests.

1111 """

1112 auth = BearerTokenAuth(self.token)

1113 req = auth(requests.Request("GET", "https://example.org").prepare())

1114 self.assertEqual(req.headers.get("Authorization"), f"Bearer {self.token}")

1115

1116 def test_token_insecure_http(self):

1117 """Ensure that no 'Authorization' header is attached to a request when

1118 using insecure HTTP.

1119 """

1120 auth = BearerTokenAuth(self.token)

1121 for url in ("http://example.org", "HTTP://example.org", "HttP://example.org"):

1122 req = auth(requests.Request("GET", url).prepare())

1123 self.assertIsNone(req.headers.get("Authorization"))

1124

1125 def test_token_file(self):

1126 """Ensure when the provided token is a file path, its contents is

1127 correctly used in the the 'Authorization' header of the requests.

1128 """

1129 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1130 f.write(self.token)

1131 token_file_path = f.name

1132

1133 # Ensure the request's "Authorization" header is set with the right

1134 # token value

1135 os.chmod(token_file_path, stat.S_IRUSR)

1136 auth = BearerTokenAuth(token_file_path)

1137 req = auth(requests.Request("GET", "https://example.org").prepare())

1138 self.assertEqual(req.headers.get("Authorization"), f"Bearer {self.token}")

1139

1140 # Ensure an exception is raised if either group or other can read the

1141 # token file

1142 for mode in (stat.S_IRGRP, stat.S_IWGRP, stat.S_IXGRP, stat.S_IROTH, stat.S_IWOTH, stat.S_IXOTH):

1143 os.chmod(token_file_path, stat.S_IRUSR | mode)

1144 with self.assertRaises(PermissionError):

1145 BearerTokenAuth(token_file_path)

1146

1147

1148class SessionStoreTestCase(unittest.TestCase):

1149 """Test for the SessionStore class."""

1150

1151 def setUp(self):

1152 self.tmpdir = ResourcePath(makeTestTempDir(TESTDIR))

1153 self.rpath = ResourcePath("https://example.org")

1154

1155 def tearDown(self):

1156 if self.tmpdir and self.tmpdir.isLocal: 1156 ↛ exitline 1156 didn't return from function 'tearDown' because the condition on line 1156 was always true

1157 removeTestTempDir(self.tmpdir.ospath)

1158

1159 def test_ca_cert_bundle(self):

1160 """Ensure that, if specified, a certificate authorities bundle is used

1161 to authentify the remote server.

1162 """

1163 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1164 f.write("CERT BUNDLE")

1165 cert_bundle = f.name

1166

1167 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_CACERT_BUNDLE": cert_bundle}, clear=True):

1168 config = HttpResourcePathConfig()

1169 session = SessionStore(config=config).get(self.rpath)

1170 self.assertEqual(session.verify, cert_bundle)

1171

1172 def test_user_cert(self):

1173 """Ensure if user certificate and private key are provided, they are

1174 used for authenticating the client.

1175 """

1176 # Create mock certificate and private key files.

1177 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1178 f.write("CERT")

1179 client_cert = f.name

1180

1181 with tempfile.NamedTemporaryFile(mode="wt", dir=self.tmpdir.ospath, delete=False) as f:

1182 f.write("KEY")

1183 client_key = f.name

1184

1185 # Check both LSST_HTTP_AUTH_CLIENT_CERT and LSST_HTTP_AUTH_CLIENT_KEY

1186 # must be initialized.

1187 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_CLIENT_CERT": client_cert}, clear=True):

1188 with self.assertRaises(ValueError):

1189 config = HttpResourcePathConfig()

1190 SessionStore(config=config).get(self.rpath)

1191

1192 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_CLIENT_KEY": client_key}, clear=True):

1193 with self.assertRaises(ValueError):

1194 config = HttpResourcePathConfig()

1195 SessionStore(config=config).get(self.rpath)

1196

1197 # Check private key file must be accessible only by its owner.

1198 with unittest.mock.patch.dict(

1199 os.environ,

1200 {"LSST_HTTP_AUTH_CLIENT_CERT": client_cert, "LSST_HTTP_AUTH_CLIENT_KEY": client_key},

1201 clear=True,

1202 ):

1203 # Ensure the session client certificate is initialized when

1204 # only the owner can read the private key file.

1205 os.chmod(client_key, stat.S_IRUSR)

1206 config = HttpResourcePathConfig()

1207 session = SessionStore(config=config).get(self.rpath)

1208 self.assertEqual(session.cert[0], client_cert)

1209 self.assertEqual(session.cert[1], client_key)

1210

1211 # Ensure an exception is raised if either group or other can access

1212 # the private key file.

1213 for mode in (stat.S_IRGRP, stat.S_IWGRP, stat.S_IXGRP, stat.S_IROTH, stat.S_IWOTH, stat.S_IXOTH):

1214 os.chmod(client_key, stat.S_IRUSR | mode)

1215 with self.assertRaises(PermissionError):

1216 config = HttpResourcePathConfig()

1217 SessionStore(config=config).get(self.rpath)

1218

1219 def test_token_env(self):

1220 """Ensure when a token is provided via an environment variable

1221 the sessions are equipped with a BearerTokenAuth.

1222 """

1223 token = "ABCDE"

1224 with unittest.mock.patch.dict(os.environ, {"LSST_HTTP_AUTH_BEARER_TOKEN": token}, clear=True):

1225 config = HttpResourcePathConfig()

1226 session = SessionStore(config=config).get(self.rpath)

1227 self.assertEqual(type(session.auth), lsst.resources.http.BearerTokenAuth)

1228 self.assertEqual(session.auth._token, token)

1229 self.assertIsNone(session.auth._path)

1230

1231 def test_sessions(self):

1232 """Ensure the session caching mechanism works."""

1233 # Ensure the store provides a session for a given URL

1234 root_url = "https://example.org"

1235 config = HttpResourcePathConfig()

1236 store = SessionStore(config=config)

1237 session = store.get(ResourcePath(root_url))

1238 self.assertIsNotNone(session)

1239

1240 # Ensure the sessions retrieved from a single store with the same

1241 # root URIs are equal

1242 for u in (f"{root_url}", f"{root_url}/path/to/file"):

1243 self.assertEqual(session, store.get(ResourcePath(u)))

1244

1245 # Ensure sessions retrieved for different root URIs are different

1246 another_url = "https://another.example.org"

1247 self.assertNotEqual(session, store.get(ResourcePath(another_url)))

1248

1249 # Ensure the sessions retrieved from a single store for URLs with

1250 # different port numbers are different

1251 root_url_with_port = f"{another_url}:12345"

1252 session = store.get(ResourcePath(root_url_with_port))

1253 self.assertNotEqual(session, store.get(ResourcePath(another_url)))

1254

1255 # Ensure the sessions retrieved from a single store with the same

1256 # root URIs (including port numbers) are equal

1257 for u in (f"{root_url_with_port}", f"{root_url_with_port}/path/to/file"):

1258 self.assertEqual(session, store.get(ResourcePath(u)))

1259

1260

1261class TestContentRange(unittest.TestCase):

1262 """Test parsing of Content-Range header."""

1263

1264 def test_full_data(self):

1265 parsed = parse_content_range_header("bytes 123-2555/12345")

1266 self.assertEqual(parsed.range_start, 123)

1267 self.assertEqual(parsed.range_end, 2555)

1268 self.assertEqual(parsed.total, 12345)

1269

1270 parsed = parse_content_range_header(" bytes 0-0/5 ")

1271 self.assertEqual(parsed.range_start, 0)

1272 self.assertEqual(parsed.range_end, 0)

1273 self.assertEqual(parsed.total, 5)

1274

1275 def test_empty_total(self):

1276 parsed = parse_content_range_header("bytes 123-2555/*")

1277 self.assertEqual(parsed.range_start, 123)

1278 self.assertEqual(parsed.range_end, 2555)

1279 self.assertIsNone(parsed.total)

1280

1281 parsed = parse_content_range_header(" bytes 0-0/* ")

1282 self.assertEqual(parsed.range_start, 0)

1283 self.assertEqual(parsed.range_end, 0)

1284 self.assertIsNone(parsed.total)

1285

1286 def test_empty_range(self):

1287 parsed = parse_content_range_header("bytes */12345")

1288 self.assertIsNone(parsed.range_start)

1289 self.assertIsNone(parsed.range_end)

1290 self.assertEqual(parsed.total, 12345)

1291

1292 parsed = parse_content_range_header(" bytes */5 ")

1293 self.assertIsNone(parsed.range_start)

1294 self.assertIsNone(parsed.range_end)

1295 self.assertEqual(parsed.total, 5)

1296

1297 def test_invalid_input(self):

1298 with self.assertRaises(ValueError):

1299 parse_content_range_header("pages 0-10/12")

1300

1301

1302if __name__ == "__main__":

1303 unittest.main()

Coverage for tests/test_http.py: 61%

721 statements