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