• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2
3if __name__ == '__main__':
4    import sys
5
6    import pytest
7    sys.exit(pytest.main([__file__] + sys.argv[1:]))
8
9import os
10import socket
11import struct
12import subprocess
13import sys
14import time
15from os.path import join as pjoin
16
17import pytest
18
19from util import base_cmdline, basename
20
21FUSE_OP_INIT = 26
22
23FUSE_MAJOR_VERSION = 7
24FUSE_MINOR_VERSION = 38
25
26fuse_in_header_fmt = '<IIQQIIII'
27fuse_out_header_fmt = '<IiQ'
28
29fuse_init_in_fmt = '<IIIII44x'
30fuse_init_out_fmt = '<IIIIHHIIHHI28x'
31
32
33def sock_recvall(sock: socket.socket, bufsize: int) -> bytes:
34    buf = bytes()
35    while len(buf) < bufsize:
36        buf += sock.recv(bufsize - len(buf))
37    return buf
38
39
40def tst_init(sock: socket.socket):
41    unique_req = 10
42    dummy_init_req_header = struct.pack(
43        fuse_in_header_fmt, struct.calcsize(fuse_in_header_fmt) +
44        struct.calcsize(fuse_init_in_fmt), FUSE_OP_INIT, unique_req, 0, 0, 0,
45        0, 0)
46    dummy_init_req_payload = struct.pack(
47        fuse_init_in_fmt, FUSE_MAJOR_VERSION, FUSE_MINOR_VERSION, 0, 0, 0)
48    dummy_init_req = dummy_init_req_header + dummy_init_req_payload
49
50    sock.sendall(dummy_init_req)
51
52    response_header = sock_recvall(sock, struct.calcsize(fuse_out_header_fmt))
53    packet_len, _, unique_res = struct.unpack(
54        fuse_out_header_fmt, response_header)
55    assert unique_res == unique_req
56
57    response_payload = sock_recvall(sock, packet_len - len(response_header))
58    response_payload = struct.unpack(fuse_init_out_fmt, response_payload)
59    assert response_payload[0] == FUSE_MAJOR_VERSION
60
61
62def test_hello_uds(output_checker):
63    cmdline = base_cmdline + [pjoin(basename, 'example', 'hello_ll_uds')]
64    print(cmdline)
65    uds_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
66                                   stderr=output_checker.fd)
67    time.sleep(1)
68
69    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
70    sock.settimeout(1)
71    sock.connect("/tmp/libfuse-hello-ll.sock")
72
73    tst_init(sock)
74
75    sock.close()
76    uds_process.terminate()
77    try:
78        uds_process.wait(1)
79    except subprocess.TimeoutExpired:
80        uds_process.kill()
81    os.remove("/tmp/libfuse-hello-ll.sock")
82