1# 2# Copyright OpenEmbedded Contributors 3# 4# SPDX-License-Identifier: MIT 5# 6import os 7import sys 8import tempfile 9import contextlib 10import socket 11from oeqa.utils.commands import bitbake, get_bb_var, Command 12from oeqa.utils.network import get_free_port 13 14@contextlib.contextmanager 15def unfs_server(directory, logger = None, udp = True): 16 unfs_sysroot = get_bb_var("RECIPE_SYSROOT_NATIVE", "unfs3-native") 17 if not os.path.exists(os.path.join(unfs_sysroot, "usr", "bin", "unfsd")): 18 # build native tool 19 bitbake("unfs3-native -c addto_recipe_sysroot") 20 21 exports = None 22 cmd = None 23 try: 24 # create the exports file 25 with tempfile.NamedTemporaryFile(delete = False) as exports: 26 exports.write("{0} (rw,no_root_squash,no_all_squash,insecure)\n".format(directory).encode()) 27 28 # find some ports for the server 29 nfsport, mountport = get_free_port(udp), get_free_port(udp) 30 31 nenv = dict(os.environ) 32 nenv['PATH'] = "{0}/sbin:{0}/usr/sbin:{0}/usr/bin:".format(unfs_sysroot) + nenv.get('PATH', '') 33 cmd = Command(["unfsd", "-d", "-p", "-e", exports.name, "-n", str(nfsport), "-m", str(mountport)], 34 bg = True, env = nenv, output_log = logger) 35 cmd.run() 36 yield nfsport, mountport 37 finally: 38 if cmd is not None: 39 cmd.stop() 40 if exports is not None: 41 # clean up exports file 42 os.unlink(exports.name) 43 44