bundlewrap/hooks/test_zfs_consistency.py

from bundlewrap.exceptions import BundleError
from bundlewrap.utils.text import bold, green
from bundlewrap.utils.ui import io

# https://github.com/bundlewrap/bundlewrap/commit/a261304b583b78a16ad44def5508cbc27d5f1c3f
# https://github.com/bundlewrap/bundlewrap/commit/76647590d684446f146cedb89518458f18dd229e

def test_node(repo, node, **kwargs):
    if not node.has_bundle('zfs'):
        return

    zfs_pools = set(node.metadata.get('zfs/pools', {}).keys())
    zfs_mountpoints = {}

    for name, attrs in node.metadata.get('zfs/datasets', {}).items():
        if attrs.get('mountpoint'):
            if attrs['mountpoint'] in zfs_mountpoints:
                raise BundleError('{n}  zfs_dataset:{ds}  mountpoint is already taken by {o}'.format(
                    n=node.name,
                    ds=name,
                    o=zfs_mountpoints[attrs['mountpoint']],
                ))

            zfs_mountpoints[attrs['mountpoint']] = name

        pool_name = name.split('/', 1)[0]

        if pool_name not in zfs_pools and node.os != 'arch':
            raise BundleError('{n}  zfs_dataset:{ds} wants zfs_pool:{pool}, which wasn\'t found'.format(
                n=node.name,
                ds=name,
                pool=pool_name,
            ))

    io.stdout('{x} {node}  zfs metadata is consistent'.format(
        x=green("✓"),
        node=bold(node.name),
    ))