~rootmos/fetch

f03ffa028e3b17cabf24e7ffb068431d77e44525 — Gustav Behm 2 months ago 18666f7
Simplify tests by assuming cwd == temporary directory
3 files changed, 35 insertions(+), 26 deletions(-)

M tests/common.py
M tests/fresh.py
M tests/test_basic.py
M tests/common.py => tests/common.py +14 -1
@@ 19,6 19,7 @@ class TestCase(unittest.TestCase):
    def __init__(self, methodName=""):
        super().__init__(methodName)
        self.logger = None
        self._root = None

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


@@ 38,6 39,18 @@ class TestCase(unittest.TestCase):

        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 = []


@@ 65,7 78,7 @@ class TestCase(unittest.TestCase):
            stdin = asyncio.subprocess.DEVNULL,
            stdout = asyncio.subprocess.PIPE,
            stderr = asyncio.subprocess.PIPE,
            env = e, cwd = cwd)
            env = e, cwd = cwd or self.root)
        self.logger.info(f"running ({p.pid}): {cmd}")

        try:

M tests/fresh.py => tests/fresh.py +3 -0
@@ 16,6 16,7 @@ class Bytes:
    def __init__(self, bs):
        self.bs = bs
        self._sha256 = None
        self.url = None

    def __bytes__(self):
        return self.bs


@@ 38,12 39,14 @@ class Bytes:
    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):

M tests/test_basic.py => tests/test_basic.py +18 -25
@@ 13,34 13,27 @@ example_sha256 = example_json[0]["sha256"]

class BasicTests(TestCase):
    def test_add(self):
        with self.tempdir("root") as root:
            target = fresh.alphanum(prefix="t")
            stdout = self.run_exe("add", example_url, target, cwd=root)
            self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])
            self.assertEqual(fresh.Bytes.read(os.path.join(root, target)).sha256, example_sha256)
            self.assertSHA256(os.path.join(root, target), example_sha256)
        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):
        with self.tempdir("root") as root:
            target = fresh.alphanum(prefix="t")
            stdout = self.run_exe("add", "--sha256="+example_sha256, example_url, target, cwd=root)
            self.assertEqual(stdout.splitlines(), [target.encode("UTF-8")])
            self.assertEqual(fresh.Bytes.read(os.path.join(root, target)).sha256, example_sha256)
            self.assertSHA256(os.path.join(root, target), example_sha256)
        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):
        with self.tempdir("root") as root:
            p = os.path.join(root, "a")
            fresh.Bytes.fresh().write(p)
            url = f"file://{p}"
            sha256 = fresh.Bytes.zero.sha256
            self.run_exe("add", "--sha256="+sha256, url, "b", cwd=root, returncode=1)
        b = fresh.Bytes.fresh().write(self.cwd("a"))
        sha256 = fresh.Bytes.zero.sha256
        self.run_exe("add", "--sha256="+sha256, b.url, "b", returncode=1)

    def test_download(self):
        with self.tempdir("root") as root:
            shutil.copy(example_json_path, os.path.join(root, ".fetch.json"))

            downloaded = self.run_exe("download", cwd=root).splitlines()
            for i in example_json:
                self.assertIn(i["target"].encode("UTF-8"), downloaded)
                self.assertSHA256(os.path.join(root, i["target"]), i["sha256"])
        shutil.copy(example_json_path, self.cwd(".fetch.json"))
        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"])