~rootmos/fetch

47107ea210b364fdaa97c803161937b604cd2b66 — Gustav Behm 2 months ago 439a01f + 9e6798a
Merge branch 'tests'
A .build.yml => .build.yml +20 -0
@@ 0,0 1,20 @@
image: alpine/3.18
packages:
  - py3-pip
tasks:
  - check-doc: |
      cd fetch
      tools/is-clean --make --root=doc README.md
  - prepare: |
      pip install "pipenv >= 2023.11.15"
      echo 'export PATH="$HOME/.local/bin:$PATH"' >> "$HOME/.buildenv"
  - install: |
      cd fetch
      pipenv install .[test]
  - tests: |
      cd fetch
      pipenv run pytest -v
triggers:
  - action: email
    condition: always
    to: builds@rootmos.io

M .gitignore => .gitignore +3 -1
@@ 1,1 1,3 @@
__pycache__
.*.flag

*.egg-info

A Makefile => Makefile +15 -0
@@ 0,0 1,15 @@
.PHONY: test
test: .editable.flag
	pipenv run pytest -v

.PHONY: doc
doc:
	$(MAKE) -C doc

.PHONY: shell
shell: .editable.flag
	pipenv shell

.editable.flag: pyproject.toml
	pipenv install --editable .[test]
	touch $@

A Pipfile => Pipfile +12 -0
@@ 0,0 1,12 @@
[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
fetch = {extras = ["test"], file = ".", editable = true}

[dev-packages]

[requires]
python_version = "3.11"

A Pipfile.lock => Pipfile.lock +59 -0
@@ 0,0 1,59 @@
{
    "_meta": {
        "hash": {
            "sha256": "7df9e13bcbbd84ded92361f4e8d72af0508d96f9d55bf5d1cfd9354fd636518f"
        },
        "pipfile-spec": 6,
        "requires": {
            "python_version": "3.11"
        },
        "sources": [
            {
                "name": "pypi",
                "url": "https://pypi.org/simple",
                "verify_ssl": true
            }
        ]
    },
    "default": {
        "fetch": {
            "editable": true,
            "extras": [
                "test"
            ],
            "file": "."
        },
        "iniconfig": {
            "hashes": [
                "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3",
                "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"
            ],
            "markers": "python_version >= '3.7'",
            "version": "==2.0.0"
        },
        "packaging": {
            "hashes": [
                "sha256:048fb0e9405036518eaaf48a55953c750c11e1a1b68e0dd1a9d62ed0c092cfc5",
                "sha256:8c491190033a9af7e1d931d0b5dacc2ef47509b34dd0de67ed209b5203fc88c7"
            ],
            "markers": "python_version >= '3.7'",
            "version": "==23.2"
        },
        "pluggy": {
            "hashes": [
                "sha256:cf61ae8f126ac6f7c451172cf30e3e43d3ca77615509771b3a984a0730651e12",
                "sha256:d89c696a773f8bd377d18e5ecda92b7a3793cbe66c87060a6fb58c7b6e1061f7"
            ],
            "markers": "python_version >= '3.8'",
            "version": "==1.3.0"
        },
        "pytest": {
            "hashes": [
                "sha256:0d009c083ea859a71b76adf7c1d502e4bc170b80a8ef002da5806527b9591fac",
                "sha256:d989d136982de4e3b29dabcc838ad581c64e8ed52c11fbe86ddebd9da0818cd5"
            ],
            "version": "==7.4.3"
        }
    },
    "develop": {}
}

M README.md => README.md +2 -0
@@ 1,6 1,8 @@
File download helpers
=====================

[![builds.sr.ht status](https://builds.sr.ht/~rootmos/fetch.svg)](https://builds.sr.ht/~rootmos/fetch?)

## Usage
```
usage: fetch [-h] [--log LOG] [--log-file FILE] [--root ROOT]

M doc/README.in.md => doc/README.in.md +2 -0
@@ 1,6 1,8 @@
File download helpers
=====================

[![builds.sr.ht status](https://builds.sr.ht/~rootmos/fetch.svg)](https://builds.sr.ht/~rootmos/fetch?)

## Usage
@include "usage.fetch"


M fetch => fetch +22 -7
@@ 25,7 25,7 @@ def parse_args():
    parser.add_argument("--log-file", metavar="FILE", default=env("LOG_FILE"), help="redirect stdout and stderr to FILE")

    default_root = env("ROOT", ".")
    default_manifest = env("MANIFEST", os.path.join(default_root, ".fetch.json"))
    default_manifest = env("MANIFEST", os.path.join(default_root, f".{whoami}.json"))
    parser.add_argument("--root", metavar="ROOT", default=default_root, help="act relative the directory ROOT")
    parser.add_argument("--manifest", metavar="FILE", default=default_manifest, help="load manifest from DIR")



@@ 124,7 124,7 @@ class Manifest:
    def save(self):
        logger.debug(f"saving manifest: {self.path}")
        with open(self.path, "w") as f:
            json.dump(manifest.to_dict(), f, sort_keys=True, indent=4)
            json.dump(self.to_dict(), f, sort_keys=True, indent=4)

class Item:
    def __init__(self, url, target, sha256=None, timestamp=None):


@@ 144,9 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:
            self.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
        return self

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


@@ 183,7 195,7 @@ class Item:
            timestamp = datetime.datetime.fromisoformat(d["timestamp"]),
        )

if __name__ == "__main__":
def main():
    args = parse_args()
    if args.log_file is not None:
        sys.stderr = sys.stdout = open(args.log_file, "a")


@@ 200,9 212,9 @@ if __name__ == "__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(i.local, start=root))
        print(os.path.relpath(item.local, start=root))
        manifest.save()
    elif args.cmd == None or args.cmd == "download":
        items = set()


@@ 221,3 233,6 @@ if __name__ == "__main__":
            print(os.path.relpath(i.local, start=root))
    else:
        raise RuntimeError(f"unexpected command: {args.cmd}")

if __name__ == "__main__":
    main()

A pyproject.toml => pyproject.toml +16 -0
@@ 0,0 1,16 @@
[build-system]
requires = ["setuptools >= 69.0.2, < 70"]
build-backend = "setuptools.build_meta"

[project]
name = "fetch"
version = "0.0.1"
requires-python = ">=3.11"

[project.optional-dependencies]
test = [
    "pytest >= 7.4.3, < 7.5",
]

[project.scripts]
fetch = "fetch.cli:main"

A src/fetch/__init__.py => src/fetch/__init__.py +0 -0
A src/fetch/cli.py => src/fetch/cli.py +1 -0
@@ 0,0 1,1 @@
../../fetch
\ No newline at end of file

A tests/__init__.py => tests/__init__.py +5 -0
@@ 0,0 1,5 @@
package_name = __name__

def package_data(*f):
    import importlib.resources
    return importlib.resources.files(package_name).joinpath(*f)

A tests/common.py => tests/common.py +130 -0
@@ 0,0 1,130 @@
import asyncio
import datetime
import hashlib
import logging
import os
import subprocess
import tempfile
import unittest

from contextlib import contextmanager

whoami = "fetch"
env_prefix = f"{whoami.upper()}_"

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)
        self.logger = None
        self._root = None

    def setUp(self):
        super().setUp()

        level = os.environ.get("TESTS_LOG_LEVEL", "DEBUG")

        self.logger = logging.getLogger(self.id())
        self.logger.setLevel(level)

        ch = logging.StreamHandler()
        ch.setLevel(level)

        f = logging.Formatter(
            fmt='%(asctime)s:%(name)s:%(levelname)s %(message)s',
            datefmt='%Y-%m-%dT%H:%M:%S%z')
        ch.setFormatter(f)

        self.logger.addHandler(ch)

        self._root = tempfile.TemporaryDirectory(prefix=f"{whoami}-tests-{self.id()}-")

    def tearDown(self):
        self._root.cleanup()

    @property
    def root(self):
        return self._root.name

    def cwd(self, *j):
        return os.path.join(self.root, *j)

    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():
                bs = await pipe.readline()
                if len(bs):
                    rope.append(bs)
                    if logger is not None:
                        line = bs.decode("UTF-8", errors="replace").strip()
                        logger.info(line)
            return b"".join(rope)

        tails = []
        cmd = [exe] + list(args)
        self.logger.debug(f"spawning: {cmd}")

        e = { **os.environ }
        if env:
            e |= env
        n = env_prefix + "LOG_LEVEL"
        if n not in e:
            e[n] = "DEBUG"

        p = await asyncio.create_subprocess_exec(exe, *args,
            stdin = asyncio.subprocess.DEVNULL,
            stdout = asyncio.subprocess.PIPE,
            stderr = asyncio.subprocess.PIPE,
            env = e, cwd = cwd or self.root)
        self.logger.info(f"running ({p.pid}): {cmd}")

        try:
            stdout = asyncio.create_task(tail(p.stdout), name=f"{p.pid}.stdout")
            tails.append(stdout)
            stderr = asyncio.create_task(tail(p.stderr, self.logger.getChild(f"{p.pid}.stderr")), name=f"{p.pid}.stderr")
            tails.append(stderr)

            rc = await p.wait()
            if rc != returncode:
                raise subprocess.CalledProcessError(
                    returncode = rc,
                    cmd = cmd,
                    output = await stdout,
                    stderr = await stderr,
                )
            return await stdout
        except asyncio.CancelledError:
            self.logger.info(f"killing {p.pid}")
            p.kill()
            await p.wait()
        finally:
            for t in tails:
                t.cancel()

    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):
        tmp = tempfile.TemporaryDirectory(prefix=f"{whoami}-tests-{what}-")
        try:
            yield tmp.name
        finally:
            tmp.cleanup()

    def assertSHA256(self, path, expected):
        with open(path, "rb") as f:
            m = hashlib.sha256()
            while True:
                bs = f.read(4096)
                if len(bs) == 0:
                    break
                m.update(bs)
        self.assertEqual(m.hexdigest(), expected, msg=f"checksum failed: {path}")

A tests/data/example.json => tests/data/example.json +8 -0
@@ 0,0 1,8 @@
[
    {
        "sha256": "88832934c42c9ff0f0066da44bec48b31e45149ce02f6f3bd11b60cfb54af183",
        "target": "fetch",
        "timestamp": "2023-12-04T18:17:35+01:00",
        "url": "https://git.sr.ht/~rootmos/fetch/blob/439a01fc257bb5ffdcebf50d1fa95344bd7b4590/fetch"
    }
]
\ No newline at end of file

A tests/fresh.py => tests/fresh.py +56 -0
@@ 0,0 1,56 @@
import random
import string
import hashlib

def bool():
    return random.choice([True, False])

def alphanum(N=12, prefix=""):
    symbols = string.ascii_letters + string.digits
    return prefix + ''.join(random.choice(symbols) for i in range(N))

def bytestring(N=12):
    return random.randbytes(N)

class Bytes:
    def __init__(self, bs):
        self.bs = bs
        self._sha256 = None
        self.url = None

    def __bytes__(self):
        return self.bs

    def __len__(self):
        return len(self.bs)

    def __getitem__(self, i):
        return self.bs[i]

    def __eq__(self, other):
        return self.bs == bytes(other)

    @property
    def sha256(self):
        if self._sha256 is None:
            self._sha256 = hashlib.sha256(self.bs).hexdigest()
        return self._sha256

    def write(self, fn):
        with open(fn, "wb") as f:
            f.write(bytes(self))
        self.url = f"file://{fn}"
        return self

    @staticmethod
    def read(fn):
        with open(fn, "rb") as f:
            return Bytes(f.read())
        self.url = f"file://{fn}"

    @classmethod
    def fresh(cls, N=None):
        return cls(bytestring(N or 1024))

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

A tests/test_basic.py => tests/test_basic.py +98 -0
@@ 0,0 1,98 @@
import json
import shutil

from . import package_data, fresh
from .common import *

example_json_path = package_data("data", "example.json")
with open(example_json_path) as f:
    example_json = json.load(f)

example_url = example_json[0]["url"]
example_sha256 = example_json[0]["sha256"]

manifest_fn = ".fetch.json"

class BasicTests(TestCase):
    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_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")])
        self.assertEqual(fresh.Bytes.read(self.cwd(target)).sha256, example_sha256)
        self.assertSHA256(self.cwd(target), example_sha256)

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

    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)