audrius/report-print-send
1
0
Fork 0
mirror of https://github.com/OCA/report-print-send.git synced 2025-02-16 07:11:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

[9.0] Merge syleam printers module into base_report_to_printer (#60)

Browse Source 
* [IMP] Updated unit tests

* [FIX] Fixed renamed attributes

* [FIX] Remove deleted fields

* [IMP] Add printing.server and printing.job models

* [IMP] Allow to cancel all jobs, enable, and disable printers

* [IMP] Split the cups part of print_document into a new print_file method

* [IMP] Updated cron job to run the action_update_jobs method

* [ADD] Add a migration script to create a printing server from configuration
This commit is contained in:
Sylvain Garancher
2016-11-09 00:02:46 +01:00
committed by Dave Lasley
parent f53f22b46f
commit 903802caf9
21 changed files with 1079 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
base_report_to_printer/README.rst
View File

@@ -90,6 +90,7 @@ Contributors
* Lionel Sausin <ls@numerigraphe.com>
* Guewen Baconnier <guewen.baconnier@camptocamp.com>
* Dave Lasley <dave@laslabs.com>
* Sylvain Garancher <sylvain.garancher@syleam.fr>
Maintainer
----------

4
base_report_to_printer/__openerp__.py
View File

@@ -8,7 +8,7 @@
{
'name': "Report to printer",
'version': '9.0.1.0.0',
'version': '9.0.2.0.0',
'category': 'Generic Modules/Base',
'author': "Agile Business Group & Domsense, Pegueroles SCP, NaN,"
" LasLabs, Odoo Community Association (OCA)",
@@ -20,6 +20,8 @@
'security/security.xml',
'views/assets.xml',
'views/printing_printer_view.xml',
'views/printing_server.xml',
'views/printing_job.xml',
'views/printing_report_view.xml',
'views/res_users_view.xml',
'views/ir_actions_report_xml_view.xml',

6
base_report_to_printer/data/printing_data.xml
View File

@@ -18,15 +18,15 @@
</record>
<record forcecreate="True" id="ir_cron_update_printers" model="ir.cron">
<field name="name">Update Printers Status</field>
<field name="name">Update Printers Jobs</field>
<field eval="True" name="active"/>
<field name="user_id" ref="base.user_root"/>
<field name="interval_number">1</field>
<field name="interval_type">minutes</field>
<field name="numbercall">-1</field>
<field eval="False" name="doall"/>
<field eval="'printing.printer'" name="model"/>
<field eval="'update_printers_status'" name="function"/>
<field eval="'printing.server'" name="model"/>
<field eval="'action_update_jobs'" name="function"/>
<field eval="'()'" name="args"/>
</record>

19
base_report_to_printer/migrations/9.0.2.0.0/post-10-create_server_record.py Normal file
View File

@@ -0,0 +1,19 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016 SYLEAM (<http://www.syleam.fr>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from openerp import SUPERUSER_ID, api
from openerp.tools.config import config
__name__ = 'Create a printing.server record from previous configuration'
def migrate(cr, v):
with api.Environment.manage():
uid = SUPERUSER_ID
env = api.Environment(cr, uid, {})
env['printing.server'].create({
'name': config.get('cups_host', 'localhost'),
'address': config.get('cups_host', 'localhost'),
'port': config.get('cups_port', 631),
})

2
base_report_to_printer/models/__init__.py
View File

@@ -2,7 +2,9 @@
from . import ir_actions_report_xml
from . import printing_action
from . import printing_job
from . import printing_printer
from . import printing_server
from . import printing_report_xml_action
from . import report
from . import res_users

107
base_report_to_printer/models/printing_job.py Normal file
View File

@@ -0,0 +1,107 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016 SYLEAM (<http://www.syleam.fr>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import logging
from openerp import models, fields, api
_logger = logging.getLogger(__name__)
class PrintingJob(models.Model):
_name = 'printing.job'
_description = 'Printing Job'
_order = 'job_id_cups DESC'
name = fields.Char(help='Job name.')
active = fields.Boolean(
default=True, help='Unchecked if the job is purged from cups.')
job_id_cups = fields.Integer(
string='Job ID', required=True,
help='CUPS id for this job.')
server_id = fields.Many2one(
comodel_name='printing.server', string='Server',
related='printer_id.server_id', store=True,
help='Server which hosts this job.')
printer_id = fields.Many2one(
comodel_name='printing.printer', string='Printer', required=True,
ondelete='cascade', help='Printer used for this job.')
job_media_progress = fields.Integer(
string='Media Progress', required=True,
help='Percentage of progress for this job.')
time_at_creation = fields.Datetime(
required=True, help='Date and time of creation for this job.')
time_at_processing = fields.Datetime(
help='Date and time of process for this job.')
time_at_completed = fields.Datetime(
help='Date and time of completion for this job.')
job_state = fields.Selection(selection=[
('pending', 'Pending'),
('pending held', 'Pending Held'),
('processing', 'Processing'),
('processing stopped', 'Processing Stopped'),
('canceled', 'Canceled'),
('aborted', 'Aborted'),
('completed', 'Completed'),
('unknown', 'Unknown'),
], string='State', help='Current state of the job.')
job_state_reason = fields.Selection(selection=[
('none', 'No reason'),
('aborted-by-system', 'Aborted by the system'),
('compression-error', 'Error in the compressed data'),
('document-access-error', 'The URI cannot be accessed'),
('document-format-error', 'Error in the document'),
('job-canceled-at-device', 'Cancelled at the device'),
('job-canceled-by-operator', 'Cancelled by the printer operator'),
('job-canceled-by-user', 'Cancelled by the user'),
('job-completed-successfully', 'Completed successfully'),
('job-completed-with-errors', 'Completed with some errors'),
('job-completed(with-warnings', 'Completed with some warnings'),
('job-data-insufficient', 'No data has been received'),
('job-hold-until-specified', 'Currently held'),
('job-incoming', 'Files are currently being received'),
('job-interpreting', 'Currently being interpreted'),
('job-outgoing', 'Currently being sent to the printer'),
('job-printing', 'Currently printing'),
('job-queued', 'Queued for printing'),
('job-queued-for-marker', 'Printer needs ink/marker/toner'),
('job-restartable', 'Can be restarted'),
('job-transforming', 'Being transformed into a different format'),
('printer-stopped', 'Printer is stopped'),
('printer-stopped-partly',
'Printer state reason set to \'stopped-partly\''),
('processing-to-stop-point',
'Cancelled, but printing already processed pages'),
('queued-in-device', 'Queued at the output device'),
('resources-are-not-ready',
'Resources not available to print the job'),
('service-off-line', 'Held because the printer is offline'),
('submission-interrupted', 'Files were not received in full'),
('unsupported-compression', 'Compressed using an unknown algorithm'),
('unsupported-document-format', 'Unsupported format'),
], string='State Reason', help='Reason for the current job state.')
_sql_constraints = [
('job_id_cups_unique', 'UNIQUE(job_id_cups, server_id)',
'The id of the job must be unique per server !'),
]
@api.multi
def action_cancel(self):
self.ensure_one()
return self.cancel()
@api.multi
def cancel(self, purge_job=False):
for job in self:
connection = job.server_id._open_connection()
if not connection:
continue
connection.cancelJob(job.job_id_cups, purge_job=purge_job)
# Update jobs' states info Odoo
self.mapped('server_id').update_jobs(
which='all', first_job_id=job.job_id_cups)
return True

125
base_report_to_printer/models/printing_printer.py
View File

@@ -4,6 +4,7 @@
# Copyright (C) 2011 Agile Business Group sagl (<http://www.agilebg.com>)
# Copyright (C) 2011 Domsense srl (<http://www.domsense.com>)
# Copyright (C) 2013-2014 Camptocamp (<http://www.camptocamp.com>)
# Copyright (C) 2016 SYLEAM (<http://www.syleam.fr>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import logging
@@ -11,24 +12,12 @@ import logging
import os
from tempfile import mkstemp
from openerp import models, fields, api, _
from openerp.exceptions import UserError
from openerp.tools.config import config
from openerp import models, fields, api
_logger = logging.getLogger(__name__)
try:
import cups
except ImportError:
_logger.debug('Cannot `import cups`.')
CUPS_HOST = config.get('cups_host', 'localhost')
CUPS_PORT = int(config.get('cups_port', 631)) # config.get returns a string
class PrintingPrinter(models.Model):
"""
Printers
@@ -38,8 +27,14 @@ class PrintingPrinter(models.Model):
_description = 'Printer'
_order = 'name'
name = fields.Char(required=True, select=True)
system_name = fields.Char(required=True, select=True)
name = fields.Char(required=True, index=True)
server_id = fields.Many2one(
comodel_name='printing.server', string='Server', required=True,
help='Server used to access this printer.')
job_ids = fields.One2many(
comodel_name='printing.job', inverse_name='printer_id', string='Jobs',
help='Jobs printed on this printer.')
system_name = fields.Char(required=True, index=True)
default = fields.Boolean(readonly=True)
status = fields.Selection([('unavailable', 'Unavailable'),
('printing', 'Printing'),
@@ -57,23 +52,13 @@ class PrintingPrinter(models.Model):
@api.model
def update_printers_status(self, domain=None):
_logger.warning(
'Deprecated : "printing.printer".update_printers_status has been '
'replaced by "printing.server".update_printers')
if domain is None:
domain = []
printer_recs = self.search(domain)
try:
connection = cups.Connection(CUPS_HOST, CUPS_PORT)
printers = connection.getPrinters()
except:
printer_recs.write({'status': 'server-error'})
else:
for printer in printer_recs:
cups_printer = printers.get(printer.system_name)
if cups_printer:
printer.update_from_cups(connection, cups_printer)
else:
# not in cups list
printer.status = 'unavailable'
return True
return printer_recs.mapped('server_id').update_printers()
@api.multi
def _prepare_update_from_cups(self, cups_connection, cups_printer):
@@ -83,10 +68,13 @@ class PrintingPrinter(models.Model):
5: 'error'
}
vals = {
'name': cups_printer['printer-info'],
'model': cups_printer.get('printer-make-and-model', False),
'location': cups_printer.get('printer-location', False),
'uri': cups_printer.get('device-uri', False),
'status': mapping.get(cups_printer['printer-state'], 'unknown'),
'status': mapping.get(cups_printer.get(
'printer-state'), 'unknown'),
'status_message': cups_printer.get('printer-state-message', ''),
}
return vals
@@ -99,13 +87,14 @@ class PrintingPrinter(models.Model):
:param cups_printer: dict of information returned by CUPS for the
current printer
"""
_logger.warning(
'Deprecated : "printing.printer".update_from_cups has been '
'replaced by "printing.server".update_printers')
self.ensure_one()
vals = self._prepare_update_from_cups(cups_connection, cups_printer)
if any(self[name] != value for name, value in vals.iteritems()):
self.write(vals)
return self.server_id.update_printers()
@api.multi
def print_options(self, report, format, copies=1):
def print_options(self, report=None, format=None, copies=1):
""" Hook to set print options """
options = {}
if format == 'raw':
@@ -128,29 +117,28 @@ class PrintingPrinter(models.Model):
finally:
os.close(fd)
try:
_logger.debug(
'Starting to connect to CUPS on %s:%s'
% (CUPS_HOST, CUPS_PORT))
connection = cups.Connection(CUPS_HOST, CUPS_PORT)
_logger.debug('Connection to CUPS successfull')
except:
raise UserError(
_("Failed to connect to the CUPS server on %s:%s. "
"Check that the CUPS server is running and that "
"you can reach it from the Odoo server.")
% (CUPS_HOST, CUPS_PORT))
return self.print_file(file_name, report=report, copies=copies)
options = self.print_options(report, format, copies)
@api.multi
def print_file(self, file_name, report=None, copies=1):
""" Print a file """
self.ensure_one()
connection = self.server_id._open_connection(raise_on_error=True)
options = self.print_options(
report=report, format=format, copies=copies)
_logger.debug(
'Sending job to CUPS printer %s on %s'
% (self.system_name, CUPS_HOST))
% (self.system_name, self.server_id.address))
connection.printFile(self.system_name,
file_name,
file_name,
options=options)
_logger.info("Printing job: '%s' on %s" % (file_name, CUPS_HOST))
_logger.info("Printing job: '%s' on %s" % (
file_name,
self.server_id.address,
))
return True
@api.multi
@@ -166,3 +154,42 @@ class PrintingPrinter(models.Model):
@api.multi
def get_default(self):
return self.search([('default', '=', True)], limit=1)
@api.multi
def action_cancel_all_jobs(self):
self.ensure_one()
return self.cancel_all_jobs()
@api.multi
def cancel_all_jobs(self, purge_jobs=False):
for printer in self:
connection = printer.server_id._open_connection()
connection.cancelAllJobs(
name=printer.system_name, purge_jobs=purge_jobs)
# Update jobs' states into Odoo
self.mapped('server_id').update_jobs(which='completed')
return True
@api.multi
def enable(self):
for printer in self:
connection = printer.server_id._open_connection()
connection.enablePrinter(printer.system_name)
# Update printers' stats into Odoo
self.mapped('server_id').update_printers()
return True
@api.multi
def disable(self):
for printer in self:
connection = printer.server_id._open_connection()
connection.disablePrinter(printer.system_name)
# Update printers' stats into Odoo
self.mapped('server_id').update_printers()
return True

224
base_report_to_printer/models/printing_server.py Normal file
View File

@@ -0,0 +1,224 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016 SYLEAM (<http://www.syleam.fr>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import logging
from datetime import datetime
from openerp import models, fields, api, exceptions, _
_logger = logging.getLogger(__name__)
try:
import cups
except ImportError:
_logger.debug('Cannot `import cups`.')
class PrintingServer(models.Model):
_name = 'printing.server'
_description = 'Printing server'
name = fields.Char(
default='Localhost', required=True, help='Name of the server.')
address = fields.Char(
default='localhost', required=True,
help='IP address or hostname of the server')
port = fields.Integer(
default=631, required=True, help='Port of the server.')
active = fields.Boolean(
default=True, help='If checked, this server is useable.')
printer_ids = fields.One2many(
comodel_name='printing.printer', inverse_name='server_id',
string='Printers List',
help='List of printers available on this server.')
@api.multi
def _open_connection(self, raise_on_error=False):
self.ensure_one()
connection = False
try:
connection = cups.Connection(host=self.address, port=self.port)
except:
message = _("Failed to connect to the CUPS server on %s:%s. "
"Check that the CUPS server is running and that "
"you can reach it from the Odoo server.") % (
self.address, self.port)
_logger.warning(message)
if raise_on_error:
raise exceptions.UserError(message)
return connection
@api.multi
def action_update_printers(self):
return self.update_printers()
@api.multi
def update_printers(self, domain=None, raise_on_error=False):
if domain is None:
domain = []
servers = self
if not self:
servers = self.search(domain)
res = True
for server in servers:
connection = server._open_connection(raise_on_error=raise_on_error)
if not connection:
server.printer_ids.write({'status': 'server-error'})
res = False
continue
# Update Printers
printers = connection.getPrinters()
existing_printers = dict([
(printer.system_name, printer)
for printer in server.printer_ids
])
updated_printers = []
for name, printer_info in printers.iteritems():
printer = self.env['printing.printer']
if name in existing_printers:
printer = existing_printers[name]
printer_values = printer._prepare_update_from_cups(
connection, printer_info)
printer_values.update(
system_name=name,
server_id=server.id,
)
updated_printers.append(name)
if not printer:
printer.create(printer_values)
else:
printer.write(printer_values)
# Set printers not found as unavailable
server.printer_ids.filtered(
lambda record: record.system_name not in updated_printers)\
.write({'status': 'unavailable'})
return res
@api.multi
def action_update_jobs(self):
if not self:
self = self.search([])
return self.update_jobs()
@api.multi
def update_jobs(self, which='all', first_job_id=-1):
job_obj = self.env['printing.job']
printer_obj = self.env['printing.printer']
mapping = {
3: 'pending',
4: 'pending held',
5: 'processing',
6: 'processing stopped',
7: 'canceled',
8: 'aborted',
9: 'completed',
}
# Update printers list, to ensure that jobs printers will be in Odoo
self.update_printers()
for server in self:
connection = server._open_connection()
if not connection:
continue
# Retrieve asked job data
jobs_data = connection.getJobs(
which_jobs=which, first_job_id=first_job_id,
requested_attributes=[
'job-name',
'job-id',
'printer-uri',
'job-media-progress',
'time-at-creation',
'job-state',
'job-state-reasons',
'time-at-processing',
'time-at-completed',
])
# Retrieve known uncompleted jobs data to update them
if which == 'not-completed':
oldest_uncompleted_job = job_obj.search([
('job_state', 'not in', (
'canceled',
'aborted',
'completed',
)),
], limit=1, order='job_id_cups')
if oldest_uncompleted_job:
jobs_data.update(connection.getJobs(
which_jobs='completed',
first_job_id=oldest_uncompleted_job.job_id_cups,
requested_attributes=[
'job-name',
'job-id',
'printer-uri',
'job-media-progress',
'time-at-creation',
'job-state',
'job-state-reasons',
'time-at-processing',
'time-at-completed',
]))
all_cups_job_ids = set()
for cups_job_id, job_data in jobs_data.items():
all_cups_job_ids.add(cups_job_id)
jobs = job_obj.with_context(active_test=False).search([
('job_id_cups', '=', cups_job_id),
('server_id', '=', server.id),
])
job_values = {
'name': job_data.get('job-name', ''),
'active': True,
'job_id_cups': cups_job_id,
'job_media_progress': job_data.get(
'job-media-progress', 0),
'job_state': mapping.get(
job_data.get('job-state'), 'unknown'),
'job_state_reason': job_data.get('job-state-reasons', ''),
'time_at_creation': fields.Datetime.to_string(
datetime.fromtimestamp(job_data.get(
'time-at-creation', 0))),
'time_at_processing': job_data.get(
'time-at-processing', 0) and fields.Datetime.to_string(
datetime.fromtimestamp(job_data.get(
'time-at-processing', 0))),
'time_at_completed': job_data.get(
'time-at-completed', 0) and fields.Datetime.to_string(
datetime.fromtimestamp(job_data.get(
'time-at-completed', 0))),
}
# Search for the printer in OpenERP
printer_uri = job_data['printer-uri']
printer_system_name = printer_uri[printer_uri.rfind('/') + 1:]
printer = printer_obj.search([
('server_id', '=', server.id),
('system_name', '=', printer_system_name),
], limit=1)
job_values['printer_id'] = printer.id
if jobs:
jobs.write(job_values)
else:
job_obj.create(job_values)
# Deactive purged jobs
if which == 'all' and first_job_id == -1:
purged_jobs = job_obj.search([
('job_id_cups', 'not in', list(all_cups_job_ids)),
])
purged_jobs.write({'active': False})
return True

27
base_report_to_printer/security/security.xml
View File

@@ -8,6 +8,15 @@
<record id="printing_group_user" model="res.groups">
<field name="name">Printing / Print User</field>
</record>
<record id="printing_server_group_manager" model="ir.model.access">
<field name="name">Printing Server Manager</field>
<field name="model_id" ref="model_printing_server"/>
<field name="group_id" ref="printing_group_manager"/>
<field eval="1" name="perm_read"/>
<field eval="1" name="perm_unlink"/>
<field eval="1" name="perm_write"/>
<field eval="1" name="perm_create"/>
</record>
<record id="printing_printer_group_manager" model="ir.model.access">
<field name="name">Printing Printer Manager</field>
<field name="model_id" ref="model_printing_printer"/>
@@ -38,6 +47,15 @@
</data>
<data>
<record id="printing_server_group_user" model="ir.model.access">
<field name="name">Printing Server User</field>
<field name="model_id" ref="model_printing_server"/>
<field name="group_id" ref="printing_group_user"/>
<field eval="1" name="perm_read"/>
<field eval="0" name="perm_unlink"/>
<field eval="0" name="perm_write"/>
<field eval="0" name="perm_create"/>
</record>
<record id="printing_printer_group_user" model="ir.model.access">
<field name="name">Printing Printer User</field>
<field name="model_id" ref="model_printing_printer"/>
@@ -47,6 +65,15 @@
<field eval="0" name="perm_write"/>
<field eval="0" name="perm_create"/>
</record>
<record id="printing_job_group_user" model="ir.model.access">
<field name="name">Printing Job User</field>
<field name="model_id" ref="model_printing_job"/>
<field name="group_id" ref="printing_group_user"/>
<field eval="1" name="perm_read"/>
<field eval="0" name="perm_unlink"/>
<field eval="0" name="perm_write"/>
<field eval="0" name="perm_create"/>
</record>
<record id="printing_action_group_user" model="ir.model.access">
<field name="name">Printing Action User</field>
<field name="model_id" ref="model_printing_action"/>

2
base_report_to_printer/tests/__init__.py
View File

@@ -2,7 +2,9 @@
# Copyright 2016 LasLabs Inc.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from . import test_printing_job
from . import test_printing_printer
from . import test_printing_server
from . import test_report
from . import test_res_users
from . import test_ir_actions_report_xml

3
base_report_to_printer/tests/test_ir_actions_report_xml.py
View File

@@ -14,9 +14,6 @@ class TestIrActionsReportXml(TransactionCase):
self.Model = self.env['ir.actions.report.xml']
self.vals = {}
def new_record(self):
return self.Model.create(self.vals)
def test_print_action_for_report_name_gets_report(self):
""" It should get report by name """
with mock.patch.object(self.Model, 'env') as mk:

66
base_report_to_printer/tests/test_printing_job.py Normal file
View File

@@ -0,0 +1,66 @@
# -*- coding: utf-8 -*-
# Copyright 2016 LasLabs Inc.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import mock
from openerp import fields
from openerp.tests.common import TransactionCase
model = 'openerp.addons.base_report_to_printer.models.printing_server'
class TestPrintingJob(TransactionCase):
def setUp(self):
super(TestPrintingJob, self).setUp()
self.Model = self.env['printing.server']
self.server = self.Model.create({})
self.printer_vals = {
'name': 'Printer',
'server_id': self.server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
}
self.job_vals = {
'server_id': self.server.id,
'job_id_cups': 1,
'job_media_progress': 0,
'time_at_creation': fields.Datetime.now(),
}
def new_printer(self):
return self.env['printing.printer'].create(self.printer_vals)
def new_job(self, printer, vals=None):
values = self.job_vals
if vals is not None:
values.update(vals)
values['printer_id'] = printer.id
return self.env['printing.job'].create(values)
@mock.patch('%s.cups' % model)
def test_cancel_job_error(self, cups):
""" It should catch any exception from CUPS and update status """
cups.Connection.side_effect = Exception
printer = self.new_printer()
job = self.new_job(printer, {'job_id_cups': 2})
job.action_cancel()
cups.Connection.side_effect = None
self.assertEquals(cups.Connection().cancelJob.call_count, 0)
@mock.patch('%s.cups' % model)
def test_cancel_job(self, cups):
""" It should catch any exception from CUPS and update status """
printer = self.new_printer()
job = self.new_job(printer)
job.cancel()
cups.Connection().cancelJob.assert_called_once_with(
job.job_id_cups, purge_job=False,
)

154
base_report_to_printer/tests/test_printing_printer.py
View File

@@ -2,17 +2,15 @@
# Copyright 2016 LasLabs Inc.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import tempfile
import mock
from openerp.exceptions import UserError
from openerp.tests.common import TransactionCase
from openerp.addons.base_report_to_printer.models.printing_printer import (
CUPS_HOST,
CUPS_PORT,
)
model = 'openerp.addons.base_report_to_printer.models.printing_printer'
server_model = 'openerp.addons.base_report_to_printer.models.printing_server'
class TestPrintingPrinter(TransactionCase):
@@ -20,8 +18,11 @@ class TestPrintingPrinter(TransactionCase):
def setUp(self):
super(TestPrintingPrinter, self).setUp()
self.Model = self.env['printing.printer']
self.ServerModel = self.env['printing.server']
self.server = self.env['printing.server'].create({})
self.printer_vals = {
'name': 'Printer',
'server_id': self.server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
@@ -34,7 +35,7 @@ class TestPrintingPrinter(TransactionCase):
def new_record(self):
return self.Model.create(self.printer_vals)
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_error(self, cups):
""" It should catch any exception from CUPS and update status """
cups.Connection.side_effect = Exception
@@ -44,39 +45,37 @@ class TestPrintingPrinter(TransactionCase):
'server-error', rec_id.status,
)
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_inits_cups(self, cups):
""" It should init CUPS connection """
self.new_record()
self.Model.update_printers_status()
cups.Connection.assert_called_once_with(
CUPS_HOST, CUPS_PORT,
host=self.server.address, port=self.server.port,
)
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_gets_all_printers(self, cups):
""" It should get all printers from CUPS server """
self.new_record()
self.Model.update_printers_status()
cups.Connection().getPrinters.assert_called_once_with()
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_gets_printer(self, cups):
""" It should get printer from CUPS by system_name """
rec_id = self.new_record()
self.new_record()
self.Model.update_printers_status()
cups.Connection().getPrinters().get.assert_called_once_with(
rec_id.system_name,
)
cups.Connection().getPrinters.assert_called_once_with()
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_search(self, cups):
""" It should search all when no domain """
with mock.patch.object(self.Model, 'search') as search:
self.Model.update_printers_status()
search.assert_called_once_with([])
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_search_domain(self, cups):
""" It should use specific domain for search """
with mock.patch.object(self.Model, 'search') as search:
@@ -84,19 +83,14 @@ class TestPrintingPrinter(TransactionCase):
self.Model.update_printers_status(expect)
search.assert_called_once_with(expect)
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_update_printer(self, cups):
""" It should update from CUPS when printer identified """
with mock.patch.object(self.Model, 'search') as search:
printer_mk = mock.MagicMock()
search.return_value = [printer_mk]
self.Model.update_printers_status()
printer_mk.update_from_cups.assert_called_once_with(
cups.Connection(),
cups.Connection().getPrinters().get(),
)
printer = self.new_record()
printer.update_from_cups(None, None)
cups.Connection().getPrinters.assert_called_once_with()
@mock.patch('%s.cups' % model)
@mock.patch('%s.cups' % server_model)
def test_update_printers_status_update_unavailable(self, cups):
""" It should update status when printer is unavailable """
rec_id = self.new_record()
@@ -105,3 +99,111 @@ class TestPrintingPrinter(TransactionCase):
self.assertEqual(
'unavailable', rec_id.status,
)
def test_printing_options(self):
""" It should generate the right options dictionnary """
self.assertEquals(self.Model.print_options('report', 'raw'), {
'raw': 'True',
})
self.assertEquals(self.Model.print_options('report', 'pdf', 2), {
'copies': '2',
})
self.assertEquals(self.Model.print_options('report', 'raw', 2), {
'raw': 'True',
'copies': '2',
})
@mock.patch('%s.cups' % server_model)
def test_print_report(self, cups):
""" It should print a report through CUPS """
fd, file_name = tempfile.mkstemp()
with mock.patch('%s.mkstemp' % model) as mkstemp:
mkstemp.return_value = fd, file_name
printer = self.new_record()
printer.print_document('report_name', 'content to print', 'pdf')
cups.Connection().printFile.assert_called_once_with(
printer.system_name,
file_name,
file_name,
options={})
@mock.patch('%s.cups' % server_model)
def test_print_report_error(self, cups):
""" It should print a report through CUPS """
cups.Connection.side_effect = Exception
fd, file_name = tempfile.mkstemp()
with mock.patch('%s.mkstemp' % model) as mkstemp:
mkstemp.return_value = fd, file_name
printer = self.new_record()
with self.assertRaises(UserError):
printer.print_document(
'report_name', 'content to print', 'pdf')
@mock.patch('%s.cups' % server_model)
def test_print_file(self, cups):
""" It should print a file through CUPS """
file_name = 'file_name'
printer = self.new_record()
printer.print_file(file_name, 'pdf')
cups.Connection().printFile.assert_called_once_with(
printer.system_name,
file_name,
file_name,
options={})
@mock.patch('%s.cups' % server_model)
def test_print_file_error(self, cups):
""" It should print a file through CUPS """
cups.Connection.side_effect = Exception
file_name = 'file_name'
printer = self.new_record()
with self.assertRaises(UserError):
printer.print_file(file_name)
def test_set_default(self):
""" It should set a single record as default """
printer = self.new_record()
self.assertTrue(printer.default)
other_printer = self.new_record()
other_printer.set_default()
self.assertFalse(printer.default)
self.assertTrue(other_printer.default)
# Check that calling the method on an empty recordset does nothing
self.Model.set_default()
self.assertEquals(other_printer, self.Model.get_default())
@mock.patch('%s.cups' % server_model)
def test_cancel_all_jobs(self, cups):
""" It should cancel all jobs """
printer = self.new_record()
printer.action_cancel_all_jobs()
cups.Connection().cancelAllJobs.assert_called_once_with(
name=printer.system_name,
purge_jobs=False,
)
@mock.patch('%s.cups' % server_model)
def test_cancel_and_purge_all_jobs(self, cups):
""" It should cancel all jobs """
printer = self.new_record()
printer.cancel_all_jobs(purge_jobs=True)
cups.Connection().cancelAllJobs.assert_called_once_with(
name=printer.system_name,
purge_jobs=True,
)
@mock.patch('%s.cups' % server_model)
def test_enable_printer(self, cups):
""" It should enable the printer """
printer = self.new_record()
printer.enable()
cups.Connection().enablePrinter.assert_called_once_with(
printer.system_name)
@mock.patch('%s.cups' % server_model)
def test_disable_printer(self, cups):
""" It should disable the printer """
printer = self.new_record()
printer.disable()
cups.Connection().disablePrinter.assert_called_once_with(
printer.system_name)

22
base_report_to_printer/tests/test_printing_printer_wizard.py
View File

@@ -7,14 +7,8 @@ import mock
from openerp.tests.common import TransactionCase
from openerp.exceptions import UserError
from openerp.addons.base_report_to_printer.models.printing_printer import (
CUPS_HOST,
CUPS_PORT,
)
model = '%s.%s' % ('openerp.addons.base_report_to_printer.wizards',
'printing_printer_update_wizard')
model = 'openerp.addons.base_report_to_printer.models.printing_server'
class StopTest(Exception):
@@ -26,6 +20,7 @@ class TestPrintingPrinterWizard(TransactionCase):
def setUp(self):
super(TestPrintingPrinterWizard, self).setUp()
self.Model = self.env['printing.printer.update.wizard']
self.server = self.env['printing.server'].create({})
self.printer_vals = {
'printer-info': 'Info',
'printer-make-and-model': 'Make and Model',
@@ -36,6 +31,7 @@ class TestPrintingPrinterWizard(TransactionCase):
def _record_vals(self, sys_name='sys_name'):
return {
'name': self.printer_vals['printer-info'],
'server_id': self.server.id,
'system_name': sys_name,
'model': self.printer_vals['printer-make-and-model'],
'location': self.printer_vals['printer-location'],
@@ -45,12 +41,9 @@ class TestPrintingPrinterWizard(TransactionCase):
@mock.patch('%s.cups' % model)
def test_action_ok_inits_connection(self, cups):
""" It should initialize CUPS connection """
try:
self.Model.action_ok()
except:
pass
self.Model.action_ok()
cups.Connection.assert_called_once_with(
CUPS_HOST, CUPS_PORT,
host=self.server.address, port=self.server.port,
)
@mock.patch('%s.cups' % model)
@@ -83,8 +76,11 @@ class TestPrintingPrinterWizard(TransactionCase):
)
self.assertTrue(rec_id)
for key, val in self._record_vals().iteritems():
if rec_id._fields[key].type == 'many2one':
val = self.env[rec_id._fields[key].comodel_name].browse(val)
self.assertEqual(
val, getattr(rec_id, key),
val, rec_id[key],
)
@mock.patch('%s.cups' % model)

213
base_report_to_printer/tests/test_printing_server.py Normal file
View File

@@ -0,0 +1,213 @@
# -*- coding: utf-8 -*-
# Copyright 2016 LasLabs Inc.
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import mock
from openerp import fields
from openerp.tests.common import TransactionCase
model = 'openerp.addons.base_report_to_printer.models.printing_server'
class TestPrintingServer(TransactionCase):
def setUp(self):
super(TestPrintingServer, self).setUp()
self.Model = self.env['printing.server']
self.server = self.Model.create({})
self.printer_vals = {
'name': 'Printer',
'server_id': self.server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
}
self.job_vals = {
'server_id': self.server.id,
'job_id_cups': 1,
'job_media_progress': 0,
'time_at_creation': fields.Datetime.now(),
}
def new_printer(self):
return self.env['printing.printer'].create(self.printer_vals)
def new_job(self, printer, vals=None):
values = self.job_vals
if vals is not None:
values.update(vals)
values['printer_id'] = printer.id
return self.env['printing.job'].create(values)
@mock.patch('%s.cups' % model)
def test_update_printers_error(self, cups):
""" It should catch any exception from CUPS and update status """
cups.Connection.side_effect = Exception
rec_id = self.new_printer()
self.Model.update_printers()
self.assertEqual(
'server-error', rec_id.status,
)
@mock.patch('%s.cups' % model)
def test_update_printers_inits_cups(self, cups):
""" It should init CUPS connection """
self.new_printer()
self.Model.update_printers()
cups.Connection.assert_called_once_with(
host=self.server.address, port=self.server.port,
)
@mock.patch('%s.cups' % model)
def test_update_printers_gets_all_printers(self, cups):
""" It should get all printers from CUPS server """
self.new_printer()
self.Model.update_printers()
cups.Connection().getPrinters.assert_called_once_with()
@mock.patch('%s.cups' % model)
def test_update_printers_search(self, cups):
""" It should search all when no domain """
with mock.patch.object(self.Model, 'search') as search:
self.Model.update_printers()
search.assert_called_once_with([])
@mock.patch('%s.cups' % model)
def test_update_printers_search_domain(self, cups):
""" It should use specific domain for search """
with mock.patch.object(self.Model, 'search') as search:
expect = [('id', '>', 0)]
self.Model.update_printers(expect)
search.assert_called_once_with(expect)
@mock.patch('%s.cups' % model)
def test_update_printers_update_unavailable(self, cups):
""" It should update status when printer is unavailable """
rec_id = self.new_printer()
cups.Connection().getPrinters().get.return_value = False
self.Model.action_update_printers()
self.assertEqual(
'unavailable', rec_id.status,
)
@mock.patch('%s.cups' % model)
def test_update_jobs_cron(self, cups):
""" It should get all jobs from CUPS server """
self.new_printer()
self.Model.action_update_jobs()
cups.Connection().getPrinters.assert_called_once_with()
cups.Connection().getJobs.assert_called_once_with(
which_jobs='all',
first_job_id=-1,
requested_attributes=[
'job-name',
'job-id',
'printer-uri',
'job-media-progress',
'time-at-creation',
'job-state',
'job-state-reasons',
'time-at-processing',
'time-at-completed',
],
)
@mock.patch('%s.cups' % model)
def test_update_jobs_button(self, cups):
""" It should get all jobs from CUPS server """
self.new_printer()
self.server.action_update_jobs()
cups.Connection().getPrinters.assert_called_once_with()
cups.Connection().getJobs.assert_called_once_with(
which_jobs='all',
first_job_id=-1,
requested_attributes=[
'job-name',
'job-id',
'printer-uri',
'job-media-progress',
'time-at-creation',
'job-state',
'job-state-reasons',
'time-at-processing',
'time-at-completed',
],
)
@mock.patch('%s.cups' % model)
def test_update_jobs_error(self, cups):
""" It should catch any exception from CUPS and update status """
cups.Connection.side_effect = Exception
self.new_printer()
self.server.update_jobs()
cups.Connection.assert_called_with(
host=self.server.address, port=self.server.port,
)
@mock.patch('%s.cups' % model)
def test_update_jobs_uncompleted(self, cups):
"""
It should search which jobs have been completed since last update
"""
printer = self.new_printer()
self.new_job(printer, vals={'job_state': 'completed'})
self.new_job(printer, vals={
'job_id_cups': 2,
'job_state': 'processing',
})
self.server.update_jobs(which='not-completed')
cups.Connection().getJobs.assert_any_call(
which_jobs='completed', first_job_id=2,
requested_attributes=[
'job-name',
'job-id',
'printer-uri',
'job-media-progress',
'time-at-creation',
'job-state',
'job-state-reasons',
'time-at-processing',
'time-at-completed',
],
)
@mock.patch('%s.cups' % model)
def test_update_jobs(self, cups):
"""
It should update all jobs, known or not
"""
printer = self.new_printer()
printer_uri = 'hostname:port/' + printer.system_name
cups.Connection().getJobs.return_value = {
1: {
'printer-uri': printer_uri,
},
2: {
'printer-uri': printer_uri,
'job-state': 9,
},
4: {
'printer-uri': printer_uri,
'job-state': 5,
},
}
self.new_job(printer, vals={'job_state': 'completed'})
completed_job = self.new_job(printer, vals={
'job_id_cups': 2,
'job_state': 'processing',
})
purged_job = self.new_job(printer, vals={
'job_id_cups': 3,
'job_state': 'processing',
})
self.server.update_jobs()
new_job = self.env['printing.job'].search([('job_id_cups', '=', 4)])
self.assertEqual(completed_job.job_state, 'completed')
self.assertEqual(purged_job.active, False)
self.assertEqual(new_job.job_state, 'processing')

44
base_report_to_printer/views/printing_job.xml Normal file
View File

@@ -0,0 +1,44 @@
<?xml version="1.0"?>
<odoo>
<record model="ir.ui.view" id="printing_job_view_form">
<field name="name">printing.job.form</field>
<field name="model">printing.job</field>
<field name="arch" type="xml">
<form string="Job">
<sheet>
<header>
<button name="action_cancel" type="object" string="Cancel" attrs="{'invisible': [('job_state', 'in', ('canceled', 'aborted', 'completed'))]}"/>
<field name="job_state" widget="statusbar"/>
</header>
<group>
<group>
<field name="name"/>
<field name="job_id_cups"/>
<field name="job_media_progress" widget="progressbar"/>
<field name="job_state_reason"/>
</group>
<group>
<field name="time_at_creation"/>
<field name="time_at_processing"/>
<field name="time_at_completed"/>
</group>
</group>
</sheet>
</form>
</field>
</record>
<record model="ir.ui.view" id="printing_job_view_tree">
<field name="name">printing.job.tree</field>
<field name="model">printing.job</field>
<field name="arch" type="xml">
<tree string="Job">
<field name="name"/>
<field name="job_id_cups"/>
<field name="job_state"/>
</tree>
</field>
</record>
</odoo>

151
base_report_to_printer/views/printing_printer_view.xml
View File

@@ -1,81 +1,90 @@
<?xml version="1.0"?>
<odoo>
<menuitem name="Printing"
id="printing_menu"
parent="base.menu_administration"
groups="printing_group_manager"
/>
<menuitem name="Printing"
id="printing_menu"
parent="base.menu_administration"
groups="printing_group_manager"
/>
<record model="ir.ui.view" id="printing_printer_view_form">
<field name="name">printing.printer.form</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<form string="Printers">
<sheet>
<div class="oe_title">
<h1>
<label for="name"/>
<field name="name"/>
</h1>
</div>
<group>
<field name="system_name"/>
</group>
<group col="3" colspan="4">
<field name="default"/>
<button name="set_default" string="Set Default" type="object"/>
</group>
<group>
<field name="uri"/>
<field name="model"/>
<field name="location"/>
<field name="status"/>
<field name="status_message"/>
</group>
</sheet>
</form>
</field>
</record>
<record model="ir.ui.view" id="printing_printer_view_form">
<field name="name">printing.printer.form</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<form string="Printers">
<sheet>
<header>
<button name="enable" type="object" string="Enable" attrs="{'invisible': [('status', 'in', ('available', 'printing'))]}"/>
<button name="disable" type="object" string="Disable" attrs="{'invisible': [('status', '=', 'unavailable')]}"/>
<button name="action_cancel_all_jobs" type="object" string="Cancel all running jobs"/>
</header>
<div class="oe_title">
<h1>
<label for="name"/>
<field name="name"/>
</h1>
</div>
<group>
<field name="system_name"/>
</group>
<group col="3" colspan="4">
<field name="default"/>
<button name="set_default" string="Set Default" type="object"/>
</group>
<group>
<field name="uri"/>
<field name="model"/>
<field name="location"/>
<field name="status"/>
<field name="status_message"/>
</group>
<group>
<separator string="Jobs" colspan="2"/>
<field name="job_ids" nolabel="1"/>
</group>
</sheet>
</form>
</field>
</record>
<record model="ir.ui.view" id="printing_printer_view_tree">
<field name="name">printing.printer.tree</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<tree string="Printers" colors="green:status=='available'; orange:status=='printing'; red:1;">
<field name="default"/>
<field name="name"/>
<field name="system_name"/>
<field name="status"/>
</tree>
</field>
</record>
<record model="ir.ui.view" id="printing_printer_view_tree">
<field name="name">printing.printer.tree</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<tree string="Printers" colors="green:status=='available'; orange:status=='printing'; red:1;">
<field name="default"/>
<field name="name"/>
<field name="system_name"/>
<field name="status"/>
</tree>
</field>
</record>
<record model="ir.ui.view" id="printing_printer_view_search">
<field name="name">printing.printer.search</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<search string="Printers">
<field name="name"/>
<field name="system_name"/>
<field name="location"/>
<field name="status"/>
</search>
</field>
</record>
<record model="ir.ui.view" id="printing_printer_view_search">
<field name="name">printing.printer.search</field>
<field name="model">printing.printer</field>
<field name="arch" type="xml">
<search string="Printers">
<field name="name"/>
<field name="system_name"/>
<field name="location"/>
<field name="status"/>
</search>
</field>
</record>
<record model="ir.actions.act_window" id="printing_printer_action">
<field name="name">Show Printers</field>
<field name="type">ir.actions.act_window</field>
<field name="res_model">printing.printer</field>
<field name="view_type">form</field>
<field name="view_mode">tree,form</field>
<field name="auto_refresh">20</field>
</record>
<record model="ir.actions.act_window" id="printing_printer_action">
<field name="name">Show Printers</field>
<field name="type">ir.actions.act_window</field>
<field name="res_model">printing.printer</field>
<field name="view_type">form</field>
<field name="view_mode">tree,form</field>
</record>
<menuitem name="Printers"
id="printing_printer_menu"
parent="printing_menu"
action="printing_printer_action"/>
<menuitem name="Printers"
sequence="20"
id="printing_printer_menu"
parent="printing_menu"
action="printing_printer_action"/>
</odoo>

1
base_report_to_printer/views/printing_report_view.xml
View File

@@ -28,6 +28,7 @@
<!-- Add a shorcut to "Actions/Report" in the Printing menu -->
<menuitem id="printing_report_xml_action_menu"
sequence="30"
parent="printing_menu"
action="base.ir_action_report_xml"/>

70
base_report_to_printer/views/printing_server.xml Normal file
View File

@@ -0,0 +1,70 @@
<?xml version="1.0"?>
<odoo>
<record model="ir.ui.view" id="printing_server_view_form">
<field name="name">printing.server.form</field>
<field name="model">printing.server</field>
<field name="arch" type="xml">
<form string="Servers">
<sheet>
<header>
<button name="action_update_printers" type="object" string="Update Printers"/>
<button name="action_update_jobs" type="object" string="Update Jobs"/>
</header>
<div class="oe_title">
<h1>
<field name="name"/>
</h1>
</div>
<group>
<field name="address"/>
<field name="port"/>
</group>
<group>
<separator string="Printers" colspan="2"/>
<field name="printer_ids" nolabel="1"/>
</group>
</sheet>
</form>
</field>
</record>
<record model="ir.ui.view" id="printing_server_view_tree">
<field name="name">printing.server.tree</field>
<field name="model">printing.server</field>
<field name="arch" type="xml">
<tree string="Servers">
<field name="name"/>
<field name="address"/>
<field name="port"/>
</tree>
</field>
</record>
<record model="ir.ui.view" id="printing_server_view_search">
<field name="name">printing.server.search</field>
<field name="model">printing.server</field>
<field name="arch" type="xml">
<search string="Servers">
<field name="name"/>
<field name="address"/>
<field name="port"/>
</search>
</field>
</record>
<record model="ir.actions.act_window" id="printing_server_action">
<field name="name">Servers</field>
<field name="type">ir.actions.act_window</field>
<field name="res_model">printing.server</field>
<field name="view_type">form</field>
<field name="view_mode">tree,form</field>
</record>
<menuitem name="Servers"
sequence="10"
id="printing_server_menu"
parent="printing_menu"
action="printing_server_action"/>
</odoo>

45
base_report_to_printer/wizards/printing_printer_update_wizard.py
View File

@@ -7,16 +7,7 @@
import logging
from openerp.exceptions import UserError
from openerp import models, api, _
from ..models.printing_printer import CUPS_HOST, CUPS_PORT
try:
import cups
except ImportError:
_logger = logging.getLogger(__name__)
_logger.debug('Cannot `import cups`.')
from openerp import models, api
_logger = logging.getLogger(__name__)
@@ -28,38 +19,8 @@ class PrintingPrinterUpdateWizard(models.TransientModel):
@api.multi
def action_ok(self):
# Update Printers
printer_obj = self.env['printing.printer']
try:
_logger.info('Trying to get list of printers')
connection = cups.Connection(CUPS_HOST, CUPS_PORT)
printers = connection.getPrinters()
_logger.info('Printers found: %s' % ','.join(printers.keys()))
except:
raise UserError(
_('Could not get the list of printers from the CUPS server '
'(%s:%s)') % (CUPS_HOST, CUPS_PORT))
printer_recs = printer_obj.search(
[('system_name', 'in', printers.keys())]
)
for printer in printer_recs:
del printers[printer.system_name]
_logger.info(
'Printer %s was already created' % printer.system_name)
for name, printer in printers.iteritems():
values = {
'name': printer['printer-info'],
'system_name': name,
'model': printer.get('printer-make-and-model', False),
'location': printer.get('printer-location', False),
'uri': printer.get('device-uri', False),
}
printer_obj.create(values)
_logger.info(
'Created new printer %s with URI %s'
% (values['name'], values['uri']))
self.env['printing.server'].search([]) \
.update_printers(raise_on_error=True)
return {
'name': 'Printers',

2
base_report_to_printer/wizards/printing_printer_update_wizard_view.xml
View File

@@ -23,6 +23,6 @@
<field name="view_mode">form</field>
<field name="target">new</field>
</record>
<menuitem action="action_printer_update_wizard" id="menu_printer_update_wizard" parent="printing_menu"/>
<menuitem action="action_printer_update_wizard" sequence="40" id="menu_printer_update_wizard" parent="printing_menu"/>
</data>
</openerp>