~rootmos/fetch

b336ed6572adbe00db8db9fb9e5fcfde50037cb4 — Gustav Behm 5 months ago e86b4f4
Check SHA256 when adding files
4 files changed, 30 insertions(+), 8 deletions(-)

M fetch
M tests/common.py
M tests/fresh.py
M tests/test_basic.py
M fetch => fetch +6 -2
@@ 145,7 145,11 @@ class Item:
        self.timestamp = datetime.datetime.now().astimezone()
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "xb") as f:
            self.sha256 = download(self.url, f)
            sha256 = download(self.url, f)
        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



@@ 200,7 204,7 @@ def main():
        if target in manifest:
            item = manifest[target].verify(url=args.url, sha256=args.sha256, root=root)
        else:
            item = Item(url=args.url, target=target).download(root=root)
            item = Item(url=args.url, sha256=args.sha256, target=target).download(root=root)
            manifest.add(item)
        print(os.path.relpath(item.local, start=root))
        manifest.save()

M tests/common.py => tests/common.py +6 -6
@@ 38,7 38,7 @@ class TestCase(unittest.TestCase):

        self.logger.addHandler(ch)

    async def arun(self, *args, logger=None, env=None, cwd=None):
    async def arun(self, *args, logger=None, env=None, cwd=None, returncode=0):
        async def tail(pipe, logger=None):
            rope = []
            while not pipe.at_eof():


@@ 74,10 74,10 @@ class TestCase(unittest.TestCase):
            stderr = asyncio.create_task(tail(p.stderr, self.logger.getChild(f"{p.pid}.stderr")), name=f"{p.pid}.stderr")
            tails.append(stderr)

            returncode = await p.wait()
            if returncode != 0:
            rc = await p.wait()
            if rc != returncode:
                raise subprocess.CalledProcessError(
                    returncode = returncode,
                    returncode = rc,
                    cmd = cmd,
                    output = await stdout,
                    stderr = await stderr,


@@ 91,8 91,8 @@ class TestCase(unittest.TestCase):
            for t in tails:
                t.cancel()

    def run_exe(self, *args, env=None, cwd=None):
        return asyncio.run(self.arun(*args, env=env, cwd=cwd, logger=self.logger))
    def run_exe(self, *args, env=None, cwd=None, returncode=0):
        return asyncio.run(self.arun(*args, env=env, cwd=cwd, logger=self.logger, returncode=returncode))

    @contextmanager
    def tempdir(self, what):

M tests/fresh.py => tests/fresh.py +1 -0
@@ 50,3 50,4 @@ class Bytes:
        return cls(bytestring(N or 1024))

Bytes.foo = Bytes("foo".encode("UTF-8"))
Bytes.zero = Bytes(bytes())

M tests/test_basic.py => tests/test_basic.py +17 -0
@@ 12,11 12,28 @@ class BasicTests(TestCase):
            url = "https://git.sr.ht/~rootmos/fetch/blob/439a01fc257bb5ffdcebf50d1fa95344bd7b4590/fetch"
            sha256 = "88832934c42c9ff0f0066da44bec48b31e45149ce02f6f3bd11b60cfb54af183"
            target = fresh.alphanum(prefix="t")
            stdout = self.run_exe("add", url, target, cwd=root)
            self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])
            self.assertEqual(fresh.Bytes.read(os.path.join(root, target)).sha256, sha256)
            self.assertSHA256(os.path.join(root, target), sha256)

    def test_add_with_sha256(self):
        with self.tempdir("root") as root:
            url = "https://git.sr.ht/~rootmos/fetch/blob/439a01fc257bb5ffdcebf50d1fa95344bd7b4590/fetch"
            sha256 = "88832934c42c9ff0f0066da44bec48b31e45149ce02f6f3bd11b60cfb54af183"
            target = fresh.alphanum(prefix="t")
            stdout = self.run_exe("add", "--sha256="+sha256, url, target, cwd=root)
            self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])
            self.assertEqual(fresh.Bytes.read(os.path.join(root, target)).sha256, sha256)
            self.assertSHA256(os.path.join(root, target), sha256)

    def test_add_with_invalid_sha256(self):
        with self.tempdir("root") as root:
            url = "https://git.sr.ht/~rootmos/fetch/blob/439a01fc257bb5ffdcebf50d1fa95344bd7b4590/fetch"
            sha256 = fresh.Bytes.zero.sha256
            target = fresh.alphanum(prefix="t")
            self.run_exe("add", "--sha256="+sha256, url, target, cwd=root, returncode=1)

    def test_download(self):
        p = package_data("data", "example.json")
        with open(p) as f: