Umbau auf server

2026-07-13 20:00:07 +02:00
parent 97357a88e0
commit 3193397dc7
1036 changed files with 365042 additions and 3 deletions
@@ -0,0 +1,159 @@
+"""Test whether Skyfield handles file download and file age correctly."""
+
+import os
+import shutil
+import tempfile
+from contextlib import contextmanager
+from threading import Thread
+try:
+    from Queue import Queue
+except ImportError:
+    from queue import Queue
+try:
+    from unittest.mock import patch
+except ImportError:
+    from mock import patch
+
+from assay import assert_raises
+from skyfield import api
+
+leap_second_text = (
+    b'#  File expires on 28 June 2021\n'
+    b'    41317.0    1  1 1972       10\n'
+    b'    41499.0    1  7 1972       11\n'
+)
+
+def load():
+    path = tempfile.mkdtemp()
+    try:
+        yield api.Loader(path)
+    finally:
+        shutil.rmtree(path)
+
+def save_file(load, path, content):
+    with open(load.path_to(path), 'wb') as f:
+        f.write(content)
+
+def file_contents(load, path):
+    with open(load.path_to(path), 'rb') as f:
+        return f.read()
+
+@contextmanager
+def fake_download(load, filename, content):
+    download = lambda *args, **kw: save_file(load, filename, content)
+    with patch('skyfield.iokit.download', download):
+        yield
+
+# Simple tests.
+
+def test_build_url(load):
+    url = 'https://ssd.jpl.nasa.gov/ftp/eph/planets/bsp/de421.bsp'
+    assert load.build_url('de421.bsp') == url
+    with assert_raises(ValueError, 'know the URL'):
+        load.build_url('unknown.kind.of.file')
+
+def test_open_in_main_directory(load):
+    with open(os.path.join(load.directory, 'file.tle'), 'wb') as f:
+        f.write(b'example text\n')
+    data = load.open('file.tle').read()
+    assert data == b'example text\n'
+
+def test_open_in_subdirectory(load):
+    os.mkdir(os.path.join(load.directory, 'folder'))
+    with open(os.path.join(load.directory, 'folder', 'file.tle'), 'wb') as f:
+        f.write(b'example text\n')
+    data = load.open('folder/file.tle').read()
+    assert data == b'example text\n'
+
+def test_missing_file_gets_downloaded(load):
+    with fake_download(load, 'Leap_Second.dat', leap_second_text):
+        data = load('Leap_Second.dat')
+    assert file_contents(load, 'Leap_Second.dat') == leap_second_text
+    assert list(data[1][1]) == [10, 10, 10, 11]
+
+def test_builtin_timescale_uses_recent_IERS_data(load):
+    ts = load.timescale()
+    # DUT1 cut and pasted from "20 1 1" line of "finals2000A.all":
+    assert abs(ts.utc(2020, 1, 1).dut1 - (-0.1771554)) < 1e-8
+
+def test_non_builtin_timescale_prefers_USNO_files(load):
+    save_file(load, 'deltat.data',
+              b' 1973  2  1  111.1\n'
+              b' 1973  3  1  222.2\n'
+              b' 1973  4  1  333.3\n')
+    save_file(load, 'deltat.preds',
+              b'   MJD        YEAR    TT-UT Pred  UT1-UTC Pred  ERROR\n'
+              b'   58484.000  2019.00   69.34      -0.152       0.117\n'
+              b'   58575.000  2019.25   69.48      -0.295       0.162\n')
+    save_file(load, 'Leap_Second.dat', leap_second_text)
+    save_file(load, 'finals2000A.all', b'invalid data')
+
+    ts = load.timescale(builtin=False)
+    assert abs(ts.utc(1973, 3, 1).delta_t - 222.2) < 1e-2
+
+    # Did we correctly convert the old leap second table to the new
+    # format? The offset prior to the first leap second should be +10.
+    t = ts.tai(1970, 1, 1)
+    assert t.utc == (1969, 12, 31, 23, 59, 50)
+
+def test_non_builtin_timescale_tries_to_load_finals2000A_all(load):
+    save_file(load, 'finals2000A.all', b'invalid data')
+    with assert_raises(IndexError):
+        load.timescale(builtin=False)
+
+def test_close_of_ephemeris(load):
+    eph = api.load('de421.bsp')
+    assert eph.segments
+    assert eph.codes
+    eph.close()
+    assert eph.segments == []
+    assert eph.codes == set()
+
+# Impressive tests: synchronize threads to reproduce concurrency bugs.
+
+class FakeConnection():
+    def __init__(self):
+        self.control = Queue()
+        self.blocks = Queue()
+        self.headers = {}
+
+    def __call__(self, *args, **kw):  # Intercept urlopen()
+        return self
+
+    def read(self, blocksize):  # Intercept connection.read()
+        self.control.put('Read requested')
+        return self.blocks.get()
+
+def test_concurrent_downloads(load):
+    # While downloads in a single process are protecte
+
+    eof = b''
+    data = b' 1973  2  1  43.4724\n'
+
+    c1 = FakeConnection()
+    c2 = FakeConnection()
+    t1 = Thread(target=load, args=('Leap_Second.dat',))
+    t2 = Thread(target=load, args=('Leap_Second.dat',))
+    t1.daemon = True
+    t2.daemon = True
+
+    with patch('skyfield.iokit.urlopen', c1):
+        t1.start()
+        c1.control.get()  # has reached its first read()
+    with patch('skyfield.iokit.urlopen', c2):
+        t2.start()
+        c2.control.get()  # has reached its first read()
+
+    assert sorted(os.listdir(load.directory)) == [
+        'Leap_Second.dat.download', 'Leap_Second.dat.download2',
+    ]
+    c1.blocks.put(data)
+    c1.blocks.put(eof)
+    t1.join()
+    assert sorted(os.listdir(load.directory)) == [
+        'Leap_Second.dat', 'Leap_Second.dat.download2',
+    ]
+    c2.blocks.put(data)
+    c2.blocks.put(eof)
+    t2.join()
+    assert sorted(os.listdir(load.directory)) == ['Leap_Second.dat']