-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathxcall.py
192 lines (145 loc) · 6.59 KB
/
xcall.py
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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
# encoding: utf-8
#
# Copyright (c) 2016 Rob Walton <https://github.com/robwalton>
#
# MIT Licence. See http://opensource.org/licenses/MIT
#
# Created on 2017-04-17
"""
A Python x-callback-url client used to communicate with an application's
x-callback-url scheme registered with macOS.
Uses `xcall`. `xcall` is command line macOS application providing generic
access to applications with x-callback-url schemes:
https://github.com/martinfinke/xcall
Call to this module are _probably_ not thread/process safe. An ettempt is made
to ensure that `xcall` is not running, but there is 20-30ms window in which
multiple calls to this module will result in multiple xcall processes running;
and the chance of replies being mixed up.
"""
import json
import urllib
import logging
import os
import subprocess
__all__ = ['XCallClient', 'xcall', 'XCallbackError']
XCALL_PATH = (os.path.dirname(os.path.abspath(__file__)) +
'/lib/xcall.app/Contents/MacOS/xcall')
logging.basicConfig(format='%(asctime)s %(levelname)s:%(message)s',
level=logging.WARNING)
logger = logging.getLogger(__name__)
def enable_verbose_logging():
logger.setLevel(logging.DEBUG)
class XCallbackError(Exception):
"""Exception representing an x-error callback from xcall.
"""
def __init__(self, *args, **kwargs):
Exception.__init__(self, *args, **kwargs)
def default_xerror_handler(xerror, requested_url):
"""Handle an x-error callback by raising a generic XCallbackError
xerror -- utf-8 un-encoded and then url unquoted x-error reply
requested_url -- the encoded url sent to application
(Note: this doc forms part of XCallClient API)
"""
msg = "x-error callback: '%s'" % xerror
if requested_url:
msg += " (in response to url: '%s')" % requested_url
raise XCallbackError(msg)
def xcall(scheme, action, action_parameters={},
activate_app=False):
"""Perform action and return un-marshalled result.
scheme -- scheme name of application to target
action -- the name of the application action to perform
action_parameters -- dictionary of parameters to pass with call. None
entries will be removed before sending. Values
will be utf-8 encoded and then url quoted.
activate_app -- bring target application to foreground if True
An x-success reply will be utf-8 un-encoded, then url unquoted,
and then unmarshalled using json into python objects before being
returned.
An x-error reply will result in an XCallbackError being raised.
"""
client = XCallClient(scheme)
return client.xcall(action, action_parameters, activate_app)
class XCallClient(object):
"""A client used for communicating with a particular application.
"""
def __init__(self, scheme_name, on_xerror_handler=default_xerror_handler,
json_decode_success=True):
"""Create an xcall client for a particular application.
scheme_name -- the url scheme name, as registered with macOS
on_xerror_handler -- callable to handle x-error callbacks.
See xcall.default_xerror_handler
json_decode_success -- unmarshal x-success calls if True
"""
self.scheme_name = scheme_name
self.on_xerror_handler = on_xerror_handler
self.json_decode_success = json_decode_success
def xcall(self, action, action_parameters={}, activate_app=False):
"""Perform action and return result across xcall.
action -- the name of the application action to perform
action_parameters -- dictionary of parameters to pass with call. None
entries will be removed before sending. Values
will be utf-8 encoded and then url quoted.
activate_app -- bring target application to foreground if True
An x-success reply will be utf-8 un-encoded, then url unquoted,
and then (if configured) unmarshalled using json into python objects
before being returned.
An x-error reply will result in a call to the configured
on_xerror_handler.
"""
for key in list(action_parameters):
if action_parameters[key] is None:
del action_parameters[key]
pid_list = get_pid_of_running_xcall_processes()
if pid_list:
raise AssertionError('xcall processe(s) already running. pid(s): ' + str(pid_list))
cmdurl = self._build_url(action, action_parameters)
logger.debug('--> ' + cmdurl)
result = self._xcall(cmdurl, activate_app)
logger.debug('<-- ' + str(result) + '\n')
return result
__call__ = xcall
def _build_url(self, action, action_parameter_dict):
url = '%s://x-callback-url/%s' % (self.scheme_name, action)
if action_parameter_dict:
par_list = []
for k, v in action_parameter_dict.items():
par_list.append(
k + '=' + urllib.parse.quote(str(v).encode('utf8')))
url = url + '?' + '&'.join(par_list)
return url
def _xcall(self, url, activate_app):
args = [XCALL_PATH, '-url', '"%s"' % url]
if activate_app:
args += ['-activateApp', 'YES']
logger.info('Making bash call: "%s"' % ' '.join(args))
p = subprocess.Popen(
args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
# Assert that reply had output on one, and only one of stdout and stderr
if (len(stdout) > 0) and (len(stderr) > 0):
raise AssertionError(
'xcall utility replied unexpectedly on *both* stdout and stderr.'
'\nstdout: "%s"\nstderr: "%s"\n'
'Try xcall directly from terminal with: "%s" ' % (stdout, stderr, ' '.join(args)))
if (stdout == '') and (stderr == ''):
raise AssertionError(
'xcall utility unexpectedly replied on *neither* stdout nor stderr'
'Try xcall directly from terminal with: "%s"' % ' '.join(args))
if stdout:
response = urllib.parse.unquote(stdout)
if self.json_decode_success:
return json.loads(response)
else:
return response
elif stderr:
self.on_xerror_handler(stderr, url)
def get_pid_of_running_xcall_processes():
try:
reply = subprocess.check_output(['pgrep', 'xcall'])
except subprocess.CalledProcessError:
return []
pid_list = reply.strip().split('\n')
if '' in pid_list:
pid_list.remove('')
return pid_list