jw-pkg/test/unit/python/jw/pkg/lib/util/test.py

import asyncio
import os
import stat
import tempfile

from jw.pkg.lib.ec.Local import Local
from jw.pkg.lib.util import copy

def uri(path: str) -> str:
    return 'local://localhost' + path

async def run_tests() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        # -- Source tree
        #
        #   src/a/f1
        #   src/a/sub/f2
        #   src/a/sub/subsub/f3
        #   src/a/empty
        #   src/single
        for rel, content in [
            ('src/a/f1', 'one\n'),
            ('src/a/sub/f2', 'two\n'),
            ('src/a/sub/subsub/f3', 'three\n'),
            ('src/single', 'solo\n'),
        ]:
            path = os.path.join(tmp, rel)
            os.makedirs(os.path.dirname(path), exist_ok = True)
            with open(path, 'w') as f:
                f.write(content)
        os.makedirs(os.path.join(tmp, 'src/a/empty'))

        # -- A single file to a nonexistent destination path; like cp,
        #    the destination's parent must exist
        os.makedirs(os.path.join(tmp, 'dst'))
        ret = await copy(
            uri(os.path.join(tmp, 'src/single')), uri(os.path.join(tmp, 'dst/single'))
        )
        assert ret == os.path.join(tmp, 'dst/single'), ret
        with open(os.path.join(tmp, 'dst/single')) as f:
            assert f.read() == 'solo\n'

        # -- A single file into an existing directory gets the basename
        os.makedirs(os.path.join(tmp, 'dst/dir'))
        ret = await copy(
            uri(os.path.join(tmp, 'src/single')), uri(os.path.join(tmp, 'dst/dir'))
        )
        assert ret == os.path.join(tmp, 'dst/dir/single'), ret
        with open(os.path.join(tmp, 'dst/dir/single')) as f:
            assert f.read() == 'solo\n'

        # -- A file with recursive=True behaves like a plain file copy
        ret = await copy(
            uri(os.path.join(tmp, 'src/single')),
            uri(os.path.join(tmp, 'dst/recursive-file')),
            recursive = True,
        )
        assert ret == os.path.join(tmp, 'dst/recursive-file'), ret

        # -- A tree to a nonexistent destination: the destination becomes
        #    the tree, parent directories are created, empty directories
        #    survive, and the returned paths are the copied leaves
        dst_tree = os.path.join(tmp, 'dst', 'tree')
        ret = await copy(
            uri(os.path.join(tmp, 'src/a')), uri(dst_tree), recursive = True
        )
        assert isinstance(ret, list), ret
        assert sorted(ret) == sorted([
            os.path.join(dst_tree, 'f1'),
            os.path.join(dst_tree, 'sub/f2'),
            os.path.join(dst_tree, 'sub/subsub/f3'),
        ]), ret
        for rel, content in [
            ('f1', 'one\n'),
            ('sub/f2', 'two\n'),
            ('sub/subsub/f3', 'three\n'),
        ]:
            with open(os.path.join(dst_tree, rel)) as f:
                assert f.read() == content, rel
        assert os.path.isdir(os.path.join(dst_tree, 'empty'))

        # -- A tree into an existing directory is copied into it under
        #    the source basename
        os.makedirs(os.path.join(tmp, 'dst', 'into'))
        ret = await copy(
            uri(os.path.join(tmp, 'src/a')),
            uri(os.path.join(tmp, 'dst/into')),
            recursive = True,
        )
        assert os.path.isfile(os.path.join(tmp, 'dst/into/a/sub/f2')), ret

        # -- A mode applies to every copied leaf
        dst_mode = os.path.join(tmp, 'dst', 'mode')
        await copy(
            uri(os.path.join(tmp, 'src/a')),
            uri(dst_mode),
            mode = 0o640,
            recursive = True
        )
        st = os.stat(os.path.join(dst_mode, 'sub/subsub/f3'))
        assert stat.S_IMODE(st.st_mode) == 0o640, st

        # -- A chrooted context source: the empty address is the
        #    context's root, and landing into an existing directory
        #    takes the root's basename
        os.makedirs(os.path.join(tmp, 'dst/chroot'))
        src_fc = Local(uri(os.path.join(tmp, 'src/a')), chroot = True)
        ret = await copy(src_fc, uri(os.path.join(tmp, 'dst/chroot')), recursive = True)
        assert os.path.isfile(os.path.join(tmp, 'dst/chroot/a/f1')), ret
        assert os.path.isfile(os.path.join(tmp, 'dst/chroot/a/sub/f2')), ret

        # -- A trailing slash does not change the source's name
        os.makedirs(os.path.join(tmp, 'dst/slash'))
        ret = await copy(
            uri(os.path.join(tmp, 'src/a/')),
            uri(os.path.join(tmp, 'dst/slash')),
            recursive = True,
        )
        assert isinstance(ret, list), ret
        assert os.path.join(tmp, 'dst/slash/a/f1') in ret, ret
        src_fc = Local(uri(os.path.join(tmp, 'src/a/')), chroot = True)
        ret = await copy(src_fc, uri(os.path.join(tmp, 'dst/slash')), recursive = True)
        assert isinstance(ret, list), ret
        assert os.path.join(tmp, 'dst/slash/a/f1') in ret, ret

        # -- parents lands the object at its source path below the
        #    destination root, like cp --parents
        dst_parents = os.path.join(tmp, 'dst/parents')
        os.makedirs(dst_parents)
        src_single = os.path.join(tmp, 'src/single')
        ret = await copy(uri(src_single), uri(dst_parents), parents = True)
        assert ret == dst_parents + src_single, ret
        with open(dst_parents + src_single) as f:
            assert f.read() == 'solo\n'

        # -- parents on a tree lands the tree at its source path
        dst_ptree = os.path.join(tmp, 'dst/ptree')
        os.makedirs(dst_ptree)
        src_tree = os.path.join(tmp, 'src/a')
        ret = await copy(
            uri(src_tree), uri(dst_ptree), recursive = True, parents = True
        )
        assert isinstance(ret, list), ret
        expected = dst_ptree + src_tree
        assert sorted(ret) == sorted([
            os.path.join(expected, 'f1'),
            os.path.join(expected, 'sub/f2'),
            os.path.join(expected, 'sub/subsub/f3'),
        ]), ret
        assert os.path.isdir(os.path.join(expected, 'empty')), ret

        # -- a chrooted context source lands at its URI path
        dst_pchroot = os.path.join(tmp, 'dst/pchroot')
        os.makedirs(dst_pchroot)
        src_fc = Local(uri(src_tree), chroot = True)
        ret = await copy(src_fc, uri(dst_pchroot), recursive = True, parents = True)
        assert isinstance(ret, list), ret
        assert os.path.isfile(os.path.join(dst_pchroot + src_tree, 'f1')), ret

        # -- re-copying with parents replaces content, not layout
        with open(dst_parents + src_single, 'w') as f:
            f.write('replaced\n')
        ret = await copy(uri(src_single), uri(dst_parents), parents = True)
        assert ret == dst_parents + src_single, ret
        with open(dst_parents + src_single) as f:
            assert f.read() == 'solo\n'

        # -- A missing source fails loudly
        try:
            await copy(
                uri(os.path.join(tmp, 'src/missing')), uri(os.path.join(tmp, 'dst/x'))
            )
            raise AssertionError('expected an exception')
        except Exception:
            pass

def main() -> None:
    asyncio.run(run_tests())
    print('All util copy tests passed')

main()