1################################################################ 2# 3# Licensed to the Apache Software Foundation (ASF) under one 4# or more contributor license agreements. See the NOTICE file 5# distributed with this work for additional information 6# regarding copyright ownership. The ASF licenses this file 7# to you under the Apache License, Version 2.0 (the 8# "License"); you may not use this file except in compliance 9# with the License. You may obtain a copy of the License at 10# 11# http://www.apache.org/licenses/LICENSE-2.0 12# 13# Unless required by applicable law or agreed to in writing, 14# software distributed under the License is distributed on an 15# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 16# KIND, either express or implied. See the License for the 17# specific language governing permissions and limitations 18# under the License. 19# 20################################################################ 21 22"""fetch.py: checksum-verified downloads into a cache, offline mode, and the lookup in tools.py. No network: the downloader is faked.""" 23import hashlib 24import io 25import json 26import re 27 28import pytest 29 30from aootest import fetch, tools 31 32pytestmark = pytest.mark.selftest 33 34PAYLOAD = b"<grammar/>" 35ITEM = {"name": "x", "group": "g", "url": "https://example.invalid/x.rng", "sha256": hashlib.sha256(PAYLOAD).hexdigest(), "path": "g/x.rng", "license": "test"} 36 37 38class Opener: 39 def __init__(self, data=PAYLOAD): 40 self.data, self.calls = data, [] 41 42 def __call__(self, req): 43 self.calls.append(req.full_url) 44 return io.BytesIO(self.data) 45 46 47def test_download_verify_and_reuse(tmp_path): 48 op = Opener() 49 assert fetch.fetch(cache=tmp_path, opener=op, items=[ITEM]) == {"x": "downloaded"} 50 assert (tmp_path / "g/x.rng").read_bytes() == PAYLOAD 51 assert fetch.fetch(cache=tmp_path, opener=op, items=[ITEM]) == {"x": "present"} 52 assert len(op.calls) == 1 53 54 55def test_checksum_mismatch_leaves_nothing_behind(tmp_path): 56 res = fetch.fetch(cache=tmp_path, opener=Opener(b"tampered"), items=[ITEM]) 57 assert res["x"].startswith("failed: checksum mismatch") 58 assert list((tmp_path / "g").iterdir()) == [] 59 60 61def test_corrupt_cached_file_is_replaced_only_by_a_verified_download(tmp_path): 62 (tmp_path / "g").mkdir() 63 (tmp_path / "g/x.rng").write_bytes(b"corrupt") 64 assert fetch.status(ITEM, tmp_path) == "bad" 65 assert fetch.fetch(cache=tmp_path, opener=Opener(b"still wrong"), items=[ITEM])["x"].startswith("failed") 66 assert (tmp_path / "g/x.rng").read_bytes() == b"corrupt" 67 assert fetch.fetch(cache=tmp_path, opener=Opener(), items=[ITEM]) == {"x": "downloaded"} 68 assert fetch.status(ITEM, tmp_path) == "ok" 69 70 71def test_offline_downloads_nothing(tmp_path): 72 op = Opener() 73 assert fetch.fetch(offline=True, cache=tmp_path, opener=op, items=[ITEM]) == {"x": "missing"} 74 assert op.calls == [] 75 76 77def test_network_error_is_reported_not_raised(tmp_path): 78 def broken(req): 79 raise OSError("no route") 80 assert fetch.fetch(cache=tmp_path, opener=broken, items=[ITEM]) == {"x": "failed: no route"} 81 82 83def test_only_https_is_fetched(tmp_path): 84 http = dict(ITEM, url="http://example.invalid/x.rng") 85 op = Opener() 86 assert fetch.fetch(cache=tmp_path, opener=op, items=[http])["x"].startswith("failed: only https") 87 assert op.calls == [] 88 89 90def test_cache_dir_override(monkeypatch, tmp_path): 91 monkeypatch.setenv("AOO_TEST_CACHE", str(tmp_path)) 92 assert fetch.cache_dir() == tmp_path 93 94 95def test_fetch_json_is_well_formed(): 96 items = fetch.load_items() 97 assert {it["group"] for it in items} == {"schemas", "jing", "odfvalidator"} 98 assert len({it["name"] for it in items}) == len({it["path"] for it in items}) == len(items) 99 for it in items: 100 assert it["url"].startswith("https://") 101 assert re.fullmatch(r"[0-9a-f]{64}", it["sha256"]) 102 assert not it["path"].startswith("/") and ".." not in it["path"].split("/") 103 assert it["license"] 104 assert sum(it["group"] == "schemas" for it in items) == 8 105 106 107def test_tools_prefer_the_environment_then_the_cache(monkeypatch, tmp_path): 108 jar = tmp_path / "cache" / "jing" / "jing-20181222.jar" 109 jar.parent.mkdir(parents=True) 110 monkeypatch.setenv("AOO_TEST_CACHE", str(tmp_path / "cache")) 111 monkeypatch.delenv("AOO_JING", raising=False) 112 assert tools.jing() is None # nothing cached yet 113 jar.write_bytes(b"not the pinned jar") 114 assert tools.jing() is None # a file with the wrong checksum is not used 115 pinned = next(it for it in fetch.load_items() if it["name"] == "jing") 116 monkeypatch.setattr(fetch, "load_items", lambda path=None: [dict(pinned, sha256=hashlib.sha256(b"not the pinned jar").hexdigest())]) 117 assert tools.jing() == str(jar) # verified cache hit 118 own = tmp_path / "own.jar" 119 own.write_bytes(b"x") 120 monkeypatch.setenv("AOO_JING", str(own)) 121 assert tools.jing() == str(own) # the environment wins 122