def get_current_local_nfs_mounts()

in src/watchdog/__init__.py [0:0]

• 51 lines of code
• 16 McCabe index (conditional complexity)

def get_current_local_nfs_mounts(mount_file="/proc/mounts"):
    """
    Return a dict of the current NFS mounts for servers running on localhost, keyed by the mountpoint and port as it
    appears in EFS watchdog state files.
    """
    mounts = []

    if not check_if_running_on_macos():
        with open(mount_file) as f:
            for mount in f:
                try:
                    mounts.append(Mount._make(mount.strip().split()))
                except Exception as e:
                    # Make sure nfs mounts being skipped are made apparent
                    if " nfs4 " in mount:
                        logging.warning(
                            'Watchdog ignoring malformed nfs4 mount "%s": %s', mount, e
                        )
                    else:
                        logging.debug(
                            'Watchdog ignoring malformed mount "%s": %s', mount, e
                        )
    else:
        # stat command on MacOS does not have '--file-system' option to verify the filesystem type of a mount point,
        # traverse all the mounts, and find if current mount point is already mounted
        process = subprocess.run(
            ["mount", "-t", "nfs"],
            check=True,
            stdout=subprocess.PIPE,
            universal_newlines=True,
        )
        stdout = process.stdout
        if stdout:
            output = stdout.split("\n")
            for mount in output:
                _mount = mount.split()
                if len(_mount) >= 4:
                    mount_ops = get_nfs_mount_options_on_macos(_mount[2], _mount[0])
                    # Sample output: 127.0.0.1:/ on /Users/ec2-user/efs (nfs)
                    mounts.append(
                        Mount._make(
                            [
                                _mount[0],
                                _mount[2],
                                _mount[3],
                                mount_ops if mount_ops else "",
                                0,
                                0,
                            ]
                        )
                    )
        else:
            logging.warning("No nfs mounts found")

    mounts = [m for m in mounts if m.server.startswith("127.0.0.1") and "nfs" in m.type]

    mount_dict = {}
    for m in mounts:
        safe_mnt = get_file_safe_mountpoint(m)
        if safe_mnt:
            mount_dict[safe_mnt] = m

    return mount_dict