jepler/leapseconddata
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #30 from jepler/sources-timeout

Browse source
This commit is contained in:
Jeff Epler 2025-04-25 08:52:34 +02:00 committed by GitHub
commit b7ae9ccca6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 36 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
leapseconddata/__init__.py
View file

@ -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

33
leapseconddata/__main__.py
View file

@ -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

20
testleapseconddata.py
View file

@ -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)