1#!/usr/bin/env python3

2

3 from subprocess import CalledProcessError, check_output, STDOUT

4 from tempfile import TemporaryDirectory

5 import argparse

6 import logging

7 import os

8 import re

9 import sys

10 import time

11

12 OBEX_RESPONSE_CODE = {

13 0x10: "Continue",

14 0x20: "OK, Success",

15 0x21: "Created",

16 0x22: "Accepted",

17 0x23: "Non-Authoritative Information",

18 0x24: "No Content",

19 0x25: "Reset Content",

20 0x26: "Partial Content",

21 0x30: "Multiple Choices",

22 0x31: "Moved Permanently",

23 0x32: "Moved temporarily",

24 0x33: "See Other",

25 0x34: "Not modified",

26 0x35: "Use Proxy",

27 0x40: "Bad Request - server couldn't understand request",

28 0x41: "Unauthorized",

29 0x42: "Payment required",

30 0x43: "Forbidden - operation is understood but refused",

31 0x44: "Not Found",

32 0x45: "Method not allowed",

33 0x46: "Not Acceptable",

34 0x47: "Proxy Authentication required",

35 0x48: "Request Time Out",

36 0x49: "Conflict",

37 0x4A: "Gone",

38 0x4B: "Length Required",

39 0x4C: "Precondition failed",

40 0x4D: "Requested entity too large",

41 0x4E: "Request URL too large",

42 0x4F: "Unsupported media type",

43 0x50: "Internal Server Error",

44 0x51: "Not Implemented",

45 0x52: "Bad Gateway",

46 0x53: "Service Unavailable",

47 0x54: "Gateway Timeout",

48 0x55: "HTTP version not supported",

49 0x60: "Database Full",

50 0x61: "Database Locked",

51}

52

53

54 class ObexFTPTest:

55 def __init__(self, path, btaddr):

56 self._file = path

57 self._filename = os.path.basename(path)

58 self._filesize = os.path.getsize(path)

59 self._btaddr = btaddr

60

61 def _error_helper(self, pattern, **extra):

62 # obexftp 0.23 version returns 255 on success, see:

63 # http://bugs.debian.org/cgi-bin/bugreport.cgi?bug=549623

64 if "exception" in extra:

65 exception = extra.get("exception")

66 if re.search(pattern, exception.output):

67 logging.info("PASS")

68 return 0

69 else:

70 logging.error(exception.output.strip())

71 if exception.returncode in OBEX_RESPONSE_CODE:

72 logging.error(OBEX_RESPONSE_CODE.get(exception.returncode))

73 return exception.returncode

74 elif "output" in extra:

75 output = extra.get("output")

76 if re.search(pattern, output):

77 logging.info("PASS")

78 return 0

79 else:

80 logging.error(output)

81 return 1

82

83 def _run_command(self, command, expected_pattern, cwd=None):

84 try:

85 output = check_output(

86 command, stderr=STDOUT, universal_newlines=True, cwd=cwd

87 )

88 return self._error_helper(expected_pattern, output=output)

89 except OSError as e:

90 logging.error(e)

91 logging.error(

92 "Binary not found, " "maybe obexftp is not installed"

93 )

94 except CalledProcessError as e:

95 return self._error_helper(expected_pattern, exception=e)

96 finally:

97 # Let the Bluetooth stack enough time to close the connection

98 # before doing another test

99 time.sleep(5)

100

101 def send(self):

102 logging.info("[ Send test ]".center(80, "="))

103 logging.info("Using {} as a test file".format(self._filename))

104 logging.info("Sending {} to {}".format(self._file, self._btaddr))

105 return self._run_command(

106 ["obexput", "-b", self._btaddr, self._file], "Sending.*?done"

107 )

108

109 def browse(self):

110 logging.info("[ Browse test ]".center(80, "="))

111 logging.info("Checking {} for {}".format(self._btaddr, self._file))

112 logging.info("Will check for a filesize of {}".format(self._filesize))

113 return self._run_command(

114 ["obexftp", "-b", self._btaddr, "-l"],

115 '{}.*?size="{}"'.format(self._filename, self._filesize),

116 )

117

118 def remove(self):

119 logging.info("[ Remove test ]".center(80, "="))

120 logging.info(

121 "Removing {} from {}".format(self._filename, self._btaddr)

122 )

123 return self._run_command(

124 ["obexrm", "-b", self._btaddr, self._filename], "Sending.*?done"

125 )

126

127 def get(self):

128 with TemporaryDirectory() as tmpdirname:

129 logging.info("[ Get test ]".center(80, "="))

130 logging.info(

131 "Getting file {} from {}".format(self._filename, self._btaddr)

132 )

133 # Dont trust "get" returncode, it's always 0...

134 return self._run_command(

135 ["obexget", "-b", self._btaddr, self._filename],

136 "Receiving.*?done",

137 cwd=tmpdirname,

138 )

139

140

141 def main():

142 description = "Bluetooth tests using ObexFTP"

143 parser = argparse.ArgumentParser(description=description)

144 parser.add_argument("file", type=argparse.FileType("rb"))

145 parser.add_argument("btaddr", help="bluetooth mac address")

146 parser.add_argument("action", choices=["send", "browse", "remove", "get"])

147 args = parser.parse_args()

148 args.file.close()

149 # Configure logging as requested

150 logging.basicConfig(

151 level=logging.INFO,

152 stream=sys.stdout,

153 format="%(levelname)s: %(message)s",

154 )

155 return getattr(ObexFTPTest(args.file.name, args.btaddr), args.action)()

156

157

158 if __name__ == "__main__":

159 sys.exit(main())