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
|
#!/usr/bin/env python
## Copyright (C) 2007, 2008, 2009 Red Hat, Inc.
## Copyright (C) 2008 Novell, Inc.
## Authors: Tim Waugh <twaugh@redhat.com>, Vincent Untz
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.
## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
import cups
import gobject
import gtk
import os
import asyncipp
import asyncpk1
import authconn
import config
from debug import *
import debug
def set_gettext_function (x):
asyncipp.set_gettext_function (x)
######
###### A class to keep track of what we're trying to achieve in order
###### to display that to the user if authentication is required.
######
class SemanticOperations(object):
def __init__ (self):
self._operation_stack = []
def _begin_operation (self, operation):
self._operation_stack.append (operation)
def _end_operation (self):
self._operation_stack.pop ()
def current_operation (self):
try:
return self._operation_stack[0]
except IndexError:
return None
######
###### An asynchronous libcups API using IPP or PolicyKit as
###### appropriate.
######
class Connection(SemanticOperations):
def __init__ (self, reply_handler=None, error_handler=None,
auth_handler=None, host=None, port=None, encryption=None,
parent=None, try_as_root=True, prompt_allowed=True):
super (Connection, self).__init__ ()
# Decide whether to use direct IPP or PolicyKit.
if host == None:
host = cups.getServer()
use_pk = ((host.startswith ('/') or host == 'localhost') and
os.getuid () != 0)
if use_pk and try_as_root:
debugprint ("Using polkit-1 connection class")
c = asyncpk1.PK1Connection (reply_handler=reply_handler,
error_handler=error_handler,
host=host, port=port,
encryption=encryption,
parent=parent)
self._conn = c
else:
debugprint ("Using IPP connection class")
c = asyncipp.IPPAuthConnection (reply_handler=reply_handler,
error_handler=error_handler,
auth_handler=auth_handler,
host=host, port=port,
encryption=encryption,
parent=parent,
try_as_root=try_as_root,
prompt_allowed=prompt_allowed,
semantic=self)
self._conn = c
methodtype = type (self._conn.getPrinters)
instancemethodtype = type (self._conn.getDevices)
bindings = []
for fname in dir (self._conn):
if fname.startswith ('_'):
continue
fn = getattr (self._conn, fname)
if type (fn) != methodtype and type (fn) != instancemethodtype:
continue
if not hasattr (self, fname):
setattr (self, fname, self._make_binding (fn))
bindings.append (fname)
self._bindings = bindings
debugprint ("+%s" % self)
def __del__ (self):
debug.debugprint ("-%s" % self)
def __eq__ (self, other):
# We want to be able to be compared as equal to our captured
# connection class.
return self._conn == other
def __ne__ (self, other):
return self._conn != other
def destroy (self):
debugprint ("DESTROY: %s" % self)
try:
self._conn.destroy ()
except AttributeError:
pass
for binding in self._bindings:
delattr (self, binding)
def _make_binding (self, fn):
return lambda *args, **kwds: fn (*args, **kwds)
def set_auth_info (self, password):
"""Call this from your auth_handler function."""
self.thread.set_auth_info (password)
if __name__ == "__main__":
# Demo
set_debugging (True)
gobject.threads_init ()
class Test:
def __init__ (self):
self._conn = Connection ()
debugprint ("+%s" % self)
def __del__ (self):
debug.debugprint ("-%s" % self)
def destroy (self):
debugprint ("DESTROY: %s" % self)
self._conn.destroy ()
loop.quit ()
def getDevices (self):
self._conn.getDevices (reply_handler=self.getDevices_reply,
error_handler=self.getDevices_error)
def getDevices_reply (self, conn, result):
print result
self.destroy ()
def getDevices_error (self, conn, exc):
print repr (exc)
self.destroy ()
t = Test ()
loop = gobject.MainLoop ()
t.getDevices ()
loop.run ()
|