~rootmos/fetch

9e6798aa1d928d954f61f53fc3c566db16f7aca6 — Gustav Behm 2 months ago f03ffa0 tests
More tests
3 files changed, 81 insertions(+), 10 deletions(-)

M fetch
M tests/common.py
M tests/test_basic.py
M fetch => fetch +11 -3
@@ 144,13 144,21 @@ class Item:
        path = self.path(root)
        self.timestamp = datetime.datetime.now().astimezone()
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "xb") as f:
            sha256 = download(self.url, f)
        self.local = path
        try:
            with open(path, "xb") as f:
                sha256 = download(self.url, f)
        except FileExistsError:
            sha256 = sha256_file(path)
            if self.sha256 is not None:
                if self.sha256 != sha256:
                    raise RuntimeError(f"checksum failed", path, self.url, sha256, self.sha256)
            else:
                raise
        if self.sha256 is not None:
            if self.sha256 != sha256:
                raise RuntimeError(f"checksum failed", self.url, sha256, self.sha256)
        self.sha256 = sha256
        self.local = path
        return self

    def verify(self, root=None, url=None, sha256=None):

M tests/common.py => tests/common.py +5 -1
@@ 1,10 1,11 @@
import asyncio
import datetime
import hashlib
import logging
import os
import subprocess
import tempfile
import unittest
import hashlib

from contextlib import contextmanager



@@ 15,6 16,9 @@ script_dir = os.path.dirname(os.path.realpath(__file__))
root = os.environ.get("ROOT", os.path.dirname(script_dir))
exe = os.environ.get("EXE", os.path.join(root, whoami))

def now():
    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")

class TestCase(unittest.TestCase):
    def __init__(self, methodName=""):
        super().__init__(methodName)

M tests/test_basic.py => tests/test_basic.py +65 -6
@@ 11,15 11,17 @@ with open(example_json_path) as f:
example_url = example_json[0]["url"]
example_sha256 = example_json[0]["sha256"]

manifest_fn = ".fetch.json"

class BasicTests(TestCase):
    def test_add(self):
    def test_example_add(self):
        target = fresh.alphanum(prefix="t")
        stdout = self.run_exe("add", example_url, target)
        self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])
        self.assertEqual(fresh.Bytes.read(self.cwd(target)).sha256, example_sha256)
        self.assertSHA256(self.cwd(target), example_sha256)

    def test_add_with_sha256(self):
    def test_example_add_with_sha256(self):
        target = fresh.alphanum(prefix="t")
        stdout = self.run_exe("add", "--sha256="+example_sha256, example_url, target)
        self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])


@@ 27,13 29,70 @@ class BasicTests(TestCase):
        self.assertSHA256(self.cwd(target), example_sha256)

    def test_add_with_invalid_sha256(self):
        b = fresh.Bytes.fresh().write(self.cwd("a"))
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        sha256 = fresh.Bytes.zero.sha256
        self.run_exe("add", "--sha256="+sha256, b.url, "b", returncode=1)
        self.run_exe("add", "--sha256="+sha256, a.url, "b", returncode=1)

    def test_download(self):
        shutil.copy(example_json_path, self.cwd(".fetch.json"))
    def test_add_quietly_accepts_existing_file(self):
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        self.run_exe("add", "--sha256="+a.sha256, a.url, "a", returncode=0)

    def test_add_refuse_overwriting_existing_differing_file(self):
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        sha256 = fresh.Bytes.zero.sha256
        self.run_exe("add", "--sha256="+sha256, a.url, "a", returncode=1)

    def test_example_download(self):
        shutil.copy(example_json_path, self.cwd(manifest_fn))
        downloaded = self.run_exe("download").splitlines()
        for i in example_json:
            self.assertIn(i["target"].encode("UTF-8"), downloaded)
            self.assertSHA256(self.cwd(i["target"]), i["sha256"])

    def test_download(self):
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        j = [ {
            "url": a.url,
            "target": "b",
            "timestamp": now(),
            "sha256": a.sha256,
        } ]

        with open(self.cwd(manifest_fn), "w") as f:
            json.dump(j, f)

        downloaded = self.run_exe("download").splitlines()
        for i in j:
            self.assertIn(i["target"].encode("UTF-8"), downloaded)
            self.assertSHA256(self.cwd(i["target"]), i["sha256"])

    def test_download_existing(self):
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        j = [ {
            "url": "file:///dev/null",
            "target": "a",
            "timestamp": now(),
            "sha256": a.sha256,
        } ]

        with open(self.cwd(manifest_fn), "w") as f:
            json.dump(j, f)

        downloaded = self.run_exe("download").splitlines()
        for i in j:
            self.assertIn(i["target"].encode("UTF-8"), downloaded)
            self.assertSHA256(self.cwd(i["target"]), i["sha256"])

    def test_download_invalid_sha256(self):
        a = fresh.Bytes.fresh().write(self.cwd("a"))
        j = [ {
            "url": a.url,
            "target": "b",
            "timestamp": now(),
            "sha256": fresh.Bytes.zero.sha256,
        } ]

        with open(self.cwd(manifest_fn), "w") as f:
            json.dump(j, f)

        self.run_exe("download", returncode=1)