Gecko [ www.utparral.edu.mx ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxrwxr-x
base.py	3.49 KB	-rw-r--r--
CheckLocalServerPublishing.py	3.1 KB	-rw-r--r--
CheckNetworkServerSanity.py	7.93 KB	-rw-r--r--
CheckPPDSanity.py	6.39 KB	-rw-r--r--
CheckPrinterSanity.py	5.59 KB	-rw-r--r--
CheckSELinux.py	2.76 KB	-rw-r--r--
CheckUSBPermissions.py	5.99 KB	-rw-r--r--
ChooseNetworkPrinter.py	6.05 KB	-rw-r--r--
ChoosePrinter.py	5.74 KB	-rw-r--r--
DeviceListed.py	6.55 KB	-rw-r--r--
ErrorLogCheckpoint.py	8.92 KB	-rw-r--r--
ErrorLogFetch.py	6.33 KB	-rw-r--r--
ErrorLogParse.py	2.23 KB	-rw-r--r--
__init__.py	12.2 KB	-rwxr-xr-x
Locale.py	4.67 KB	-rw-r--r--
LocalOrRemote.py	1.46 KB	-rw-r--r--
NetworkCUPSPrinterShared.py	2.87 KB	-rw-r--r--
PrinterStateReasons.py	4.36 KB	-rw-r--r--
PrintTestPage.py	20.13 KB	-rw-r--r--
QueueNotEnabled.py	2.84 KB	-rw-r--r--
QueueRejectingJobs.py	2.91 KB	-rw-r--r--
RemoteAddress.py	2.4 KB	-rw-r--r--
SchedulerNotRunning.py	2.23 KB	-rw-r--r--
ServerFirewalled.py	2.12 KB	-rw-r--r--
Shrug.py	4.21 KB	-rw-r--r--
VerifyPackages.py	2.64 KB	-rw-r--r--
Welcome.py	2.69 KB	-rw-r--r--

Code Editor : PrintTestPage.py

#!/usr/bin/python3

## Printing troubleshooter

## Copyright (C) 2008, 2009, 2010, 2012 Red Hat, Inc.
## Authors:
##  Tim Waugh <twaugh@redhat.com>

## 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., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.

import cups
import dbus
import dbus.glib
from gi.repository import GLib
import os
from gi.repository import Gdk
from gi.repository import Gtk
from gi.repository import Pango
import tempfile
import time
from timedops import TimedOperation, OperationCanceled

from .base import *

import errordialogs
from errordialogs import *

DBUS_PATH="/com/redhat/PrinterSpooler"
DBUS_IFACE="com.redhat.PrinterSpooler"
class PrintTestPage(Question):
    STATE = { cups.IPP_JOB_PENDING: _("Pending"),
              cups.IPP_JOB_HELD: _("Held"),
              cups.IPP_JOB_PROCESSING: _("Processing"),
              cups.IPP_JOB_STOPPED: _("Stopped"),
              cups.IPP_JOB_CANCELED: _("Canceled"),
              cups.IPP_JOB_ABORTED: _("Aborted"),
              cups.IPP_JOB_COMPLETED: _("Completed") }

def __init__ (self, troubleshooter):
        Question.__init__ (self, troubleshooter, "Print test page")
        page = Gtk.VBox ()
        page.set_spacing (12)
        page.set_border_width (12)

label = Gtk.Label ()
        label.set_alignment (0, 0)
        label.set_use_markup (True)
        label.set_line_wrap (True)
        page.pack_start (label, False, False, 0)
        self.main_label = label
        self.main_label_text = ('<span weight="bold" size="larger">' +
                                _("Test Page") + '</span>\n\n' +
                                _("Now print a test page.  If you are having "
                                  "problems printing a specific document, "
                                  "print that document now and mark the print "
                                  "job below."))

hbox = Gtk.HButtonBox ()
        hbox.set_border_width (0)
        hbox.set_spacing (3)
        hbox.set_layout (Gtk.ButtonBoxStyle.START)
        self.print_button = Gtk.Button.new_with_label (_("Print Test Page"))
        hbox.pack_start (self.print_button, False, False, 0)

self.cancel_button = Gtk.Button.new_with_label (_("Cancel All Jobs"))
        hbox.pack_start (self.cancel_button, False, False, 0)
        page.pack_start (hbox, False, False, 0)

tv = Gtk.TreeView ()
        test_cell = Gtk.CellRendererToggle ()
        test = Gtk.TreeViewColumn (_("Test"), test_cell, active=0)
        job = Gtk.TreeViewColumn (_("Job"), Gtk.CellRendererText (), text=1)
        printer_cell = Gtk.CellRendererText ()
        printer = Gtk.TreeViewColumn (_("Printer"), printer_cell, text=2)
        name_cell = Gtk.CellRendererText ()
        name = Gtk.TreeViewColumn (_("Document"), name_cell, text=3)
        status = Gtk.TreeViewColumn (_("Status"), Gtk.CellRendererText (),
                                     text=4)
        test_cell.set_radio (False)
        self.test_cell = test_cell
        printer.set_resizable (True)
        printer_cell.set_property ("ellipsize", Pango.EllipsizeMode.END)
        printer_cell.set_property ("width-chars", 20)
        name.set_resizable (True)
        name_cell.set_property ("ellipsize", Pango.EllipsizeMode.END)
        name_cell.set_property ("width-chars", 20)
        status.set_resizable (True)
        tv.append_column (test)
        tv.append_column (job)
        tv.append_column (printer)
        tv.append_column (name)
        tv.append_column (status)
        tv.set_rules_hint (True)
        sw = Gtk.ScrolledWindow ()
        sw.set_policy (Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
        sw.set_shadow_type (Gtk.ShadowType.IN)
        sw.add (tv)
        self.treeview = tv
        page.pack_start (sw, False, False, 0)

label = Gtk.Label(label=_("Did the marked print jobs print correctly?"))
        label.set_line_wrap (True)
        label.set_alignment (0, 0)
        page.pack_start (label, False, False, 0)

vbox = Gtk.VBox ()
        vbox.set_spacing (6)
        self.yes = Gtk.RadioButton (label=_("Yes"))
        no = Gtk.RadioButton.new_with_label_from_widget (self.yes, _("No"))
        vbox.pack_start (self.yes, False, False, 0)
        vbox.pack_start (no, False, False, 0)
        page.pack_start (vbox, False, False, 0)
        self.persistent_answers = {}
        troubleshooter.new_page (page, self)

def display (self):
        answers = self.troubleshooter.answers
        if 'cups_queue' not in answers:
            return False

parent = self.troubleshooter.get_window ()
        self.authconn = answers['_authenticated_connection']
        mediatype = None
        defaults = answers.get ('cups_printer_ppd_defaults', {})
        for opts in defaults.values ():
            for opt, value in opts.items ():
                if opt == "MediaType":
                    mediatype = value
                    break

if mediatype is not None:
            mediatype_string = '\n\n' + (_("Remember to load paper of type "
                                           "'%s' into the printer first.") %
                                         mediatype)
        else:
            mediatype_string = ""

label_text = self.main_label_text + mediatype_string
        self.main_label.set_markup (label_text)

model = Gtk.ListStore (bool,
                               int,
                               str,
                               str,
                               str)
        self.treeview.set_model (model)
        self.job_to_iter = {}

test_jobs = self.persistent_answers.get ('test_page_job_id', [])
        def get_jobs ():
            c = self.authconn
            try:
                r = ["job-id",
                     "job-name",
                     "job-state",
                     "job-printer-uri",
                     "printer-name"]
                jobs_dict = c.getJobs (which_jobs='not-completed',
                                       my_jobs=False,
                                       requested_attributes=r)
                completed_jobs_dict = c.getJobs (which_jobs='completed',
                                                 requested_attributes=r)
            except TypeError:
                # requested_attributes requires pycups 1.9.50
                jobs_dict = c.getJobs (which_jobs='not-completed',
                                       my_jobs=False)
                completed_jobs_dict = c.getJobs (which_jobs='completed')
            return (jobs_dict, completed_jobs_dict)

self.op = TimedOperation (get_jobs, parent=parent)
        try:
            (jobs_dict, completed_jobs_dict) = self.op.run ()
        except (OperationCanceled, cups.IPPError):
            return False

# We want to display the jobs in the queue for this printer...
        try:
            queue_uri_ending = "/" + self.troubleshooter.answers['cups_queue']
            jobs_on_this_printer = [x for x in jobs_dict.keys () if \
                                            jobs_dict[x]['job-printer-uri']. \
                                            endswith (queue_uri_ending)]
        except:
            jobs_on_this_printer = []

# ...as well as any other jobs we've previous submitted as test pages.
        jobs = list (set(test_jobs).union (set (jobs_on_this_printer)))

completed_jobs_dict = None
        for job in jobs:
            try:
                j = jobs_dict[job]
            except KeyError:
                try:
                    j = completed_jobs_dict[job]
                except KeyError:
                    continue

iter = model.append (None)
            self.job_to_iter[job] = iter
            model.set_value (iter, 0, job in test_jobs)
            model.set_value (iter, 1, job)
            self.update_job (job, j)

return True

def connect_signals (self, handler):
        self.print_sigid = self.print_button.connect ("clicked",
                                                      self.print_clicked)
        self.cancel_sigid = self.cancel_button.connect ("clicked",
                                                        self.cancel_clicked)
        self.test_sigid = self.test_cell.connect ('toggled',
                                                  self.test_toggled)

def create_subscription ():
            c = self.authconn
            sub_id = c.createSubscription ("/",
                                           events=["job-created",
                                                   "job-completed",
                                                   "job-stopped",
                                                   "job-progress",
                                                   "job-state-changed"])
            return sub_id

parent = self.troubleshooter.get_window ()
        self.op = TimedOperation (create_subscription, parent=parent)
        try:
            self.sub_id = self.op.run ()
        except (OperationCanceled, cups.IPPError):
            pass

try:
            bus = dbus.SystemBus ()
        except:
            bus = None

self.bus = bus
        if bus:
            bus.add_signal_receiver (self.handle_dbus_signal,
                                     path=DBUS_PATH,
                                     dbus_interface=DBUS_IFACE)

self.timer = GLib.timeout_add_seconds (1, self.update_jobs_list)

def disconnect_signals (self):
        if self.bus:
            self.bus.remove_signal_receiver (self.handle_dbus_signal,
                                             path=DBUS_PATH,
                                             dbus_interface=DBUS_IFACE)
                                             
        self.print_button.disconnect (self.print_sigid)
        self.cancel_button.disconnect (self.cancel_sigid)
        self.test_cell.disconnect (self.test_sigid)

def cancel_subscription (sub_id):
            c = self.authconn
            c.cancelSubscription (sub_id)

parent = self.troubleshooter.get_window ()
        self.op = TimedOperation (cancel_subscription,
                                  (self.sub_id,),
                                  parent=parent)
        try:
            self.op.run ()
        except (OperationCanceled, cups.IPPError):
            pass

try:
            del self.sub_seq
        except:
            pass

GLib.source_remove (self.timer)

def collect_answer (self):
        if not self.displayed:
            return {}

self.answers = self.persistent_answers.copy ()
        parent = self.troubleshooter.get_window ()
        success = self.yes.get_active ()
        self.answers['test_page_successful'] = success

class collect_jobs:
            def __init__ (self, model):
                self.jobs = []
                model.foreach (self.each, None)

def each (self, model, path, iter, user_data):
                self.jobs.append (model.get (iter, 0, 1, 2, 3, 4))

model = self.treeview.get_model ()
        jobs = collect_jobs (model).jobs
        def collect_attributes (jobs):
            job_attrs = None
            c = self.authconn
            with_attrs = []
            for (test, jobid, printer, doc, status) in jobs:
                attrs = None
                if test:
                    try:
                        attrs = c.getJobAttributes (jobid)
                    except AttributeError:
                        # getJobAttributes was introduced in pycups 1.9.35.
                        if job_attrs is None:
                            job_attrs = c.getJobs (which_jobs='all')

attrs = self.job_attrs[jobid]

with_attrs.append ((test, jobid, printer, doc, status, attrs))

return with_attrs

self.op = TimedOperation (collect_attributes,
                                  (jobs,),
                                  parent=parent)
        try:
            with_attrs = self.op.run ()
            self.answers['test_page_job_status'] = with_attrs
        except (OperationCanceled, cups.IPPError):
            pass

return self.answers

def cancel_operation (self):
        self.op.cancel ()

# Abandon the CUPS connection and make another.
        answers = self.troubleshooter.answers
        factory = answers['_authenticated_connection_factory']
        self.authconn = factory.get_connection ()
        self.answers['_authenticated_connection'] = self.authconn

def handle_dbus_signal (self, *args):
        debugprint ("D-Bus signal caught: updating jobs list soon")
        GLib.source_remove (self.timer)
        self.timer = GLib.timeout_add (200, self.update_jobs_list)

def update_job (self, jobid, job_dict):
        iter = self.job_to_iter[jobid]
        model = self.treeview.get_model ()
        try:
            printer_name = job_dict['printer-name']
        except KeyError:
            try:
                uri = job_dict['job-printer-uri']
                r = uri.rfind ('/')
                printer_name = uri[r + 1:]
            except KeyError:
                printer_name = None

if printer_name is not None:
            model.set_value (iter, 2, printer_name)

model.set_value (iter, 3, job_dict['job-name'])
        model.set_value (iter, 4, self.STATE[job_dict['job-state']])

def print_clicked (self, widget):
        now = time.time ()
        tt = time.localtime (now)
        when = time.strftime ("%d/%b/%Y:%T %z", tt)
        self.persistent_answers['test_page_attempted'] = when
        answers = self.troubleshooter.answers
        parent = self.troubleshooter.get_window ()

def print_test_page (*args, **kwargs):
            factory = answers['_authenticated_connection_factory']
            c = factory.get_connection ()
            return c.printTestPage (*args, **kwargs)

tmpfname = None
        mimetypes = [None, 'text/plain']
        for mimetype in mimetypes:
            try:
                if mimetype is None:
                    # Default test page.
                    self.op = TimedOperation (print_test_page,
                                              (answers['cups_queue'],),
                                              parent=parent)
                    jobid = self.op.run ()
                elif mimetype == 'text/plain':
                    (tmpfd, tmpfname) = tempfile.mkstemp ()
                    os.write (tmpfd, b"This is a test page.\n")
                    os.close (tmpfd)
                    self.op = TimedOperation (print_test_page,
                                              (answers['cups_queue'],),
                                              kwargs={'file': tmpfname,
                                                      'format': mimetype},
                                              parent=parent)
                    jobid = self.op.run ()
                    try:
                        os.unlink (tmpfname)
                    except OSError:
                        pass

tmpfname = None

jobs = self.persistent_answers.get ('test_page_job_id', [])
                jobs.append (jobid)
                self.persistent_answers['test_page_job_id'] = jobs
                break
            except OperationCanceled:
                self.persistent_answers['test_page_submit_failure'] = 'cancel'
                break
            except RuntimeError:
                self.persistent_answers['test_page_submit_failure'] = 'connect'
                break
            except cups.IPPError as e:
                (e, s) = e.args
                if (e == cups.IPP_DOCUMENT_FORMAT and
                    mimetypes.index (mimetype) < (len (mimetypes) - 1)):
                    # Try next format.
                    if tmpfname is not None:
                        os.unlink (tmpfname)
                        tmpfname = None
                    continue

self.persistent_answers['test_page_submit_failure'] = (e, s)
                show_error_dialog (_("Error submitting test page"),
                                   _("There was an error during the CUPS "
                                     "operation: '%s'.") % s,
                                   self.troubleshooter.get_window ())
                break

def cancel_clicked (self, widget):
        self.persistent_answers['test_page_jobs_cancelled'] = True
        jobids = []
        for jobid, iter in self.job_to_iter.items ():
            jobids.append (jobid)

def cancel_jobs (jobids):
            c = self.authconn
            for jobid in jobids:
                try:
                    c.cancelJob (jobid)
                except cups.IPPError as e:
                    (e, s) = e.args
                    if e != cups.IPP_NOT_POSSIBLE:
                        self.persistent_answers['test_page_cancel_failure'] = (e, s)

self.op = TimedOperation (cancel_jobs,
                                  (jobids,),
                                  parent=self.troubleshooter.get_window ())
        try:
            self.op.run ()
        except (OperationCanceled, cups.IPPError):
            pass

def test_toggled (self, cell, path):
        model = self.treeview.get_model ()
        iter = model.get_iter (path)
        active = model.get_value (iter, 0)
        model.set_value (iter, 0, not active)

def update_jobs_list (self):
        def get_notifications (self):
            c = self.authconn
            try:
                notifications = c.getNotifications ([self.sub_id],
                                                    [self.sub_seq + 1])
            except AttributeError:
                notifications = c.getNotifications ([self.sub_id])

return notifications

# Enter the GDK lock.  We need to do this because we were
        # called from a timeout.
        Gdk.threads_enter ()

parent = self.troubleshooter.get_window ()
        self.op = TimedOperation (get_notifications,
                                  (self,),
                                  parent=parent)
        try:
            notifications = self.op.run ()
        except (OperationCanceled, cups.IPPError):
            Gdk.threads_leave ()
            return True

answers = self.troubleshooter.answers
        model = self.treeview.get_model ()
        queue = answers['cups_queue']
        test_jobs = self.persistent_answers.get('test_page_job_id', [])
        for event in notifications['events']:
            seq = event['notify-sequence-number']
            self.sub_seq = seq

job = event['notify-job-id']

nse = event['notify-subscribed-event']
            if nse == 'job-created':
                if (job in test_jobs or
                    event['printer-name'] == queue):
                    iter = model.append (None)
                    self.job_to_iter[job] = iter
                    model.set_value (iter, 0, True)
                    model.set_value (iter, 1, job)
                else:
                    continue
            elif job not in self.job_to_iter:
                continue

if (job in test_jobs and
                nse in ["job-stopped", "job-completed"]):
                comp = self.persistent_answers.get ('test_page_completions', [])
                comp.append ((job, event['notify-text']))
                self.persistent_answers['test_page_completions'] = comp

self.update_job (job, event)

# Update again when we're told to. (But we might update sooner if
        # there is a D-Bus signal.)
        GLib.source_remove (self.timer)
        self.timer = GLib.timeout_add_seconds (
            notifications['notify-get-interval'],
            self.update_jobs_list)
        debugprint ("Update again in %ds" %
                    notifications['notify-get-interval'])
        Gdk.threads_leave ()
        return False

${this.title}

Code Editor : PrintTestPage.py