|
| 1 | +"""Daily real-server probes for GenericArchive upstream-drift detection. |
| 2 | +
|
| 3 | +These tests run on the contract tier (daily cron) and hit the live |
| 4 | +LPP and CDA mirror endpoints. Failures here mean an upstream changed |
| 5 | +something we depend on — the relevant cassettes in |
| 6 | +tests/cassettes/test_direct_archive_downloader/ or test_file_access/ |
| 7 | +need re-recording, or a code-side adjustment. |
| 8 | +
|
| 9 | +Keep small (4-5 probes); the bulk of GenericArchive coverage runs on |
| 10 | +the unit tier via cassettes. |
| 11 | +""" |
| 12 | + |
| 13 | +from __future__ import annotations |
| 14 | + |
| 15 | +import unittest |
| 16 | +from datetime import datetime |
| 17 | + |
| 18 | +import pytest |
| 19 | + |
| 20 | +import speasy as spz |
| 21 | +from speasy.core.any_files import any_loc_open |
| 22 | +from speasy.products import SpeasyVariable |
| 23 | + |
| 24 | +pytestmark = pytest.mark.contract |
| 25 | + |
| 26 | + |
| 27 | +class GenericArchiveContractProbes(unittest.TestCase): |
| 28 | + |
| 29 | + def test_can_open_remote_http_text_resource(self) -> None: |
| 30 | + """LPP cache server still serves HTML at /cache/.""" |
| 31 | + f = any_loc_open( |
| 32 | + "http://sciqlop.lpp.polytechnique.fr/cache/", mode="r" |
| 33 | + ) |
| 34 | + content = f.read() |
| 35 | + self.assertIn("<", content) |
| 36 | + |
| 37 | + def test_can_open_remote_http_binary_resource(self) -> None: |
| 38 | + """LPP data server still serves binary files at /data/.""" |
| 39 | + f = any_loc_open( |
| 40 | + "https://hephaistos.lpp.polytechnique.fr/data/LFR/SW/LFR-FSW/3.0.0.0/fsw", |
| 41 | + mode="rb", |
| 42 | + ) |
| 43 | + content = f.read(4) |
| 44 | + self.assertGreater(len(content), 0) |
| 45 | + |
| 46 | + def test_mms_fpi_burst_returns_data(self) -> None: |
| 47 | + """MMS FPI burst products still resolve via the CDA mirror. |
| 48 | + Skipped from the unit tier because cassettes are ~370-650 MB.""" |
| 49 | + result = spz.get_data( |
| 50 | + "archive/cda/MMS/MMS1/FPI/BURST/MOMS/mms1_fpi_brst_l2_des_moms/mms1_des_energyspectr_mz_brst", |
| 51 | + "2018-01-30T10", |
| 52 | + "2018-01-30T11", # 1 hour to limit transfer |
| 53 | + disable_cache=True, |
| 54 | + ) |
| 55 | + self.assertIsInstance(result, SpeasyVariable) |
| 56 | + |
| 57 | + def test_remote_text_resource_assertion_still_holds(self) -> None: |
| 58 | + """The Vbias HTML resource still returns a parseable HTML document. |
| 59 | + Skipped from the unit tier because the response body is ~137 MB.""" |
| 60 | + f = any_loc_open( |
| 61 | + "https://hephaistos.lpp.polytechnique.fr/data/jeandet/Vbias.html", |
| 62 | + mode="r", |
| 63 | + ) |
| 64 | + head = f.read(1024) |
| 65 | + self.assertIn("<!DOCTYPE", head.upper()) |
0 commit comments