From d8480e67d50f60e0b972e0fd0b9dd6f676aa5430 Mon Sep 17 00:00:00 2001 From: Jeff Epler Date: Fri, 25 Apr 2025 08:23:23 +0200 Subject: [PATCH] Add timeout & improve coverage Looks like the NIST FTP server is not happy with urllib and it fails (ugh!) by hanging for a long time. Maybe I'll have to delete it. (it works with curl, which uses EPSV and not PASV. python uses PASV for all ipv4 ftp and EPSV for ipv6, instead of trying EPSV first) Set a reasonable timeout. The format of `leapsecond sources` has been modified for better readability. --- leapseconddata/__init__.py | 8 +++++--- leapseconddata/__main__.py | 33 +++++++++++++++++++++------------ testleapseconddata.py | 20 ++++++++++---------- 3 files changed, 36 insertions(+), 25 deletions(-) diff --git a/leapseconddata/__init__.py b/leapseconddata/__init__.py index 320b2b6..3c04de4 100755 --- a/leapseconddata/__init__.py +++ b/leapseconddata/__init__.py @@ -260,6 +260,7 @@ class LeapSecondData: *, check_hash: bool = True, custom_sources: Sequence[str] = (), + timeout: float | None = 60, ) -> LeapSecondData: """Get the list of leap seconds from a standard source. @@ -276,7 +277,7 @@ class LeapSecondData: for location in itertools.chain(custom_sources, cls.standard_file_sources, cls.standard_network_sources): logging.debug("Trying leap second data from %s", location) try: - candidate = cls.from_url(location, check_hash=check_hash) + candidate = cls.from_url(location, check_hash=check_hash, timeout=timeout) except InvalidHashError: logging.warning("Invalid hash while reading %s", location) continue @@ -288,7 +289,7 @@ class LeapSecondData: if candidate.valid(when): logging.info("Using leap second data from %s", location) return candidate - logging.warning("Validity expired for %s", location) + logging.warning(f"Validity expired for {location} at {candidate.valid_until} (checking validity at {when})") raise ValidityError("No valid leap-second.list file could be found") @@ -314,6 +315,7 @@ class LeapSecondData: url: str, *, check_hash: bool = True, + timeout: float | None = 60, ) -> LeapSecondData | None: """Retrieve the leap second list from a local file @@ -321,7 +323,7 @@ class LeapSecondData: :param check_hash: Whether to check the embedded hash """ try: - with urllib.request.urlopen(url) as open_file: + with urllib.request.urlopen(url, timeout=timeout) as open_file: return cls.from_open_file(open_file, check_hash=check_hash) except urllib.error.URLError: # pragma no cover return None diff --git a/leapseconddata/__main__.py b/leapseconddata/__main__.py index 42674a7..b608f99 100644 --- a/leapseconddata/__main__.py +++ b/leapseconddata/__main__.py @@ -166,27 +166,36 @@ def table(ctx: click.Context, *, start: datetime.datetime, end: datetime.datetim @cli.command -def sources() -> None: - """Print information about leap-second.list data sources""" +@click.option("--timeout", type=float, default=12, metavar="[SECS]") +@click.argument("urls", type=str, nargs=-1) +def sources(*, timeout: float, urls: list[str]) -> None: + """Print information about leap-second.list data sources + + If no URLs are specified, print information about all standard sources. + If one or more URLs are specified, check them instead. + """ first = True - for location in LeapSecondData.standard_file_sources + LeapSecondData.standard_network_sources: + locations = urls if urls else LeapSecondData.standard_file_sources + LeapSecondData.standard_network_sources + for location in locations: if not first: print() first = False + print(f"{location}:") try: - leap_second_data = LeapSecondData.from_url(location, check_hash=True) - except InvalidHashError: # pragma no coverage - print(f"{location}: Invalid hash") - leap_second_data = LeapSecondData.from_url(location, check_hash=False) - except Exception as e: # pragma no coverage # noqa: BLE001 - print(f"{location}: {e}") + leap_second_data = LeapSecondData.from_url(location, check_hash=True, timeout=timeout) + except InvalidHashError as e: + print(f" {e}") + leap_second_data = LeapSecondData.from_url(location, check_hash=False, timeout=timeout) + except Exception as e: # noqa: BLE001 + print(f" {e}") leap_second_data = None if leap_second_data is not None: - print(f"{location}: Last updated {leap_second_data.last_updated}") - print(f"{location}: Valid until {leap_second_data.valid_until}") + print(f" Last updated {leap_second_data.last_updated}") + print(f" Valid until {leap_second_data.valid_until}") + print(f" {len(leap_second_data.leap_seconds)} leap seconds") else: - print(f"{location}: Could not be read") + print(" Could not be read") if __name__ == "__main__": # pragma no cover diff --git a/testleapseconddata.py b/testleapseconddata.py index b1541e8..f9ffe2f 100644 --- a/testleapseconddata.py +++ b/testleapseconddata.py @@ -18,10 +18,16 @@ import unittest import leapseconddata import leapseconddata.__main__ -db = leapseconddata.LeapSecondData.from_standard_source() +db = leapseconddata.LeapSecondData.from_standard_source(timeout=8) GMT1 = datetime.timezone(datetime.timedelta(seconds=3600), "GMT1") +bad_sources = [ + "data:text/plain;base64,SGVsbG8sIFdvcmxkIQ==", + "data:text/plain,%23h%099dac5845%208acd32c0%202947d462%20daf4a943%20f58d9391%0A", + "file:///doesnotexist", +] + class LeapSecondDataTest(unittest.TestCase): def run_main(self, *args: str) -> None: @@ -45,7 +51,8 @@ class LeapSecondDataTest(unittest.TestCase): self.run_main("next-leapsecond", "2100-2-2") self.run_main("previous-leapsecond", "2009-2-2") self.run_main("previous-leapsecond", "1960-2-2") - self.run_main("sources") + self.run_main("sources", "--timeout", "8") + self.run_main("sources", *bad_sources) def test_corrupt(self) -> None: self.assertRaises( @@ -98,14 +105,7 @@ class LeapSecondDataTest(unittest.TestCase): def test_invalid2(self) -> None: when = datetime.datetime(datetime.MAXYEAR, 1, 1, tzinfo=datetime.timezone.utc) - datetime.timedelta(seconds=1) with self.assertRaises(leapseconddata.ValidityError): - leapseconddata.LeapSecondData.from_standard_source( - when, - custom_sources=[ - "data:text/plain;base64,SGVsbG8sIFdvcmxkIQ==", - "data:text/plain,%23h%099dac5845%208acd32c0%202947d462%20daf4a943%20f58d9391%0A", - "file:///doesnotexist", - ], - ) + leapseconddata.LeapSecondData.from_standard_source(when, custom_sources=bad_sources, timeout=8) def test_tz(self) -> None: when = datetime.datetime(1999, 1, 1, tzinfo=datetime.timezone.utc) - datetime.timedelta(seconds=1)