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