[FIX] .travis.yml: Update definition

This commit is contained in:
Pedro M. Baeza
2019-10-01 15:26:55 +02:00
committed by Carlos Roca
parent b1103c7b6d
commit 226d4b2c31
25 changed files with 995 additions and 1043 deletions

View File

@@ -6,30 +6,28 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
{
'name': "Report to printer",
'version': '13.0.1.0.0',
'category': 'Generic Modules/Base',
'author': "Agile Business Group & Domsense, Pegueroles SCP, NaN,"
" LasLabs, Camptocamp, Odoo Community Association (OCA),"
" Open for Small Business Ltd",
'website': 'http://www.agilebg.com',
'license': 'AGPL-3',
"depends": ['web'],
'data': [
'data/printing_data.xml',
'security/security.xml',
'views/assets.xml',
'views/printing_printer.xml',
'views/printing_server.xml',
'views/printing_job.xml',
'views/printing_report.xml',
'views/res_users.xml',
'views/ir_actions_report.xml',
'wizards/printing_printer_update_wizard_view.xml',
"name": "Report to printer",
"version": "13.0.1.0.0",
"category": "Generic Modules/Base",
"author": "Agile Business Group & Domsense, Pegueroles SCP, NaN,"
" LasLabs, Camptocamp, Odoo Community Association (OCA),"
" Open for Small Business Ltd",
"website": "http://www.agilebg.com",
"license": "AGPL-3",
"depends": ["web"],
"data": [
"data/printing_data.xml",
"security/security.xml",
"views/assets.xml",
"views/printing_printer.xml",
"views/printing_server.xml",
"views/printing_job.xml",
"views/printing_report.xml",
"views/res_users.xml",
"views/ir_actions_report.xml",
"wizards/printing_printer_update_wizard_view.xml",
],
'installable': True,
'application': False,
'external_dependencies': {
'python': ['pycups'],
},
"installable": True,
"application": False,
"external_dependencies": {"python": ["pycups"]},
}

View File

@@ -1,4 +1,3 @@
from . import ir_actions_report
from . import printing_action
from . import printing_job

View File

@@ -5,35 +5,33 @@
# Copyright (C) 2013-2014 Camptocamp (<http://www.camptocamp.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import api, exceptions, fields, models, _
from odoo import _, api, exceptions, fields, models
class IrActionsReport(models.Model):
_inherit = 'ir.actions.report'
_inherit = "ir.actions.report"
property_printing_action_id = fields.Many2one(
comodel_name='printing.action',
string='Default Behaviour',
comodel_name="printing.action",
string="Default Behaviour",
company_dependent=True,
)
printing_printer_id = fields.Many2one(
comodel_name='printing.printer',
string='Default Printer'
comodel_name="printing.printer", string="Default Printer"
)
printer_tray_id = fields.Many2one(
comodel_name='printing.tray',
string='Paper Source',
comodel_name="printing.tray",
string="Paper Source",
domain="[('printer_id', '=', printing_printer_id)]",
)
printing_action_ids = fields.One2many(
comodel_name='printing.report.xml.action',
inverse_name='report_id',
string='Actions',
help='This field allows configuring action and printer on a per '
'user basis'
comodel_name="printing.report.xml.action",
inverse_name="report_id",
string="Actions",
help="This field allows configuring action and printer on a per " "user basis",
)
@api.onchange('printing_printer_id')
@api.onchange("printing_printer_id")
def onchange_printing_printer_id(self):
""" Reset the tray when the printer is changed """
self.printer_tray_id = False
@@ -49,68 +47,69 @@ class IrActionsReport(models.Model):
return {}
result = report.behaviour()
serializable_result = {
'action': result['action'],
'printer_name': result['printer'].name,
"action": result["action"],
"printer_name": result["printer"].name,
}
return serializable_result
def _get_user_default_print_behaviour(self):
printer_obj = self.env['printing.printer']
printer_obj = self.env["printing.printer"]
user = self.env.user
return dict(
action=user.printing_action or 'client',
action=user.printing_action or "client",
printer=user.printing_printer_id or printer_obj.get_default(),
tray=str(user.printer_tray_id.system_name) if
user.printer_tray_id else False
tray=str(user.printer_tray_id.system_name)
if user.printer_tray_id
else False,
)
def _get_report_default_print_behaviour(self):
result = {}
report_action = self.property_printing_action_id
if report_action and report_action.action_type != 'user_default':
result['action'] = report_action.action_type
if report_action and report_action.action_type != "user_default":
result["action"] = report_action.action_type
if self.printing_printer_id:
result['printer'] = self.printing_printer_id
result["printer"] = self.printing_printer_id
if self.printer_tray_id:
result['tray'] = self.printer_tray_id.system_name
result["tray"] = self.printer_tray_id.system_name
return result
def behaviour(self):
self.ensure_one()
printing_act_obj = self.env['printing.report.xml.action']
printing_act_obj = self.env["printing.report.xml.action"]
result = self._get_user_default_print_behaviour()
result.update(self._get_report_default_print_behaviour())
# Retrieve report-user specific values
print_action = printing_act_obj.search([
('report_id', '=', self.id),
('user_id', '=', self.env.uid),
('action', '!=', 'user_default'),
], limit=1)
print_action = printing_act_obj.search(
[
("report_id", "=", self.id),
("user_id", "=", self.env.uid),
("action", "!=", "user_default"),
],
limit=1,
)
if print_action:
# For some reason design takes report defaults over
# False action entries so we must allow for that here
result.update({k: v for k, v in
print_action.behaviour().items() if v})
result.update({k: v for k, v in print_action.behaviour().items() if v})
return result
def print_document(self, record_ids, data=None):
""" Print a document, do not return the document file """
document, doc_format = self.with_context(
must_skip_send_to_printer=True).render_qweb_pdf(
record_ids, data=data)
must_skip_send_to_printer=True
).render_qweb_pdf(record_ids, data=data)
behaviour = self.behaviour()
printer = behaviour.pop('printer', None)
printer = behaviour.pop("printer", None)
if not printer:
raise exceptions.Warning(
_('No printer configured to print this report.')
)
raise exceptions.Warning(_("No printer configured to print this report."))
# TODO should we use doc_format instead of report_type
return printer.print_document(self, document,
doc_format=self.report_type,
**behaviour)
return printer.print_document(
self, document, doc_format=self.report_type, **behaviour
)
def _can_print_report(self, behaviour, printer, document):
"""Predicate that decide if report can be sent to printer
@@ -118,16 +117,16 @@ class IrActionsReport(models.Model):
If you want to prevent `render_qweb_pdf` to send report you can set
the `must_skip_send_to_printer` key to True in the context
"""
if self.env.context.get('must_skip_send_to_printer'):
if self.env.context.get("must_skip_send_to_printer"):
return False
if behaviour['action'] == 'server' and printer and document:
if behaviour["action"] == "server" and printer and document:
return True
return False
def report_action(self, docids, data=None, config=True):
res = super().report_action(docids, data=data, config=config)
if not res.get('id'):
res['id'] = self.id
if not res.get("id"):
res["id"] = self.id
return res
def render_qweb_pdf(self, res_ids=None, data=None):
@@ -137,14 +136,16 @@ class IrActionsReport(models.Model):
generated document as well.
"""
document, doc_format = super(IrActionsReport, self).render_qweb_pdf(
res_ids=res_ids, data=data)
res_ids=res_ids, data=data
)
behaviour = self.behaviour()
printer = behaviour.pop('printer', None)
printer = behaviour.pop("printer", None)
can_print_report = self._can_print_report(behaviour, printer, document)
if can_print_report:
printer.print_document(self, document, doc_format=self.report_type,
**behaviour)
printer.print_document(
self, document, doc_format=self.report_type, **behaviour
)
return document, doc_format

View File

@@ -5,24 +5,22 @@
# Copyright (C) 2013-2014 Camptocamp (<http://www.camptocamp.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import models, fields, api
from odoo import api, fields, models
class PrintingAction(models.Model):
_name = 'printing.action'
_description = 'Print Job Action'
_name = "printing.action"
_description = "Print Job Action"
@api.model
def _available_action_types(self):
return [
('server', 'Send to Printer'),
('client', 'Send to Client'),
('user_default', "Use user's defaults"),
("server", "Send to Printer"),
("client", "Send to Client"),
("user_default", "Use user's defaults"),
]
name = fields.Char(required=True)
action_type = fields.Selection(
selection=_available_action_types,
string='Type',
required=True,
selection=_available_action_types, string="Type", required=True
)

View File

@@ -2,87 +2,110 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import logging
from odoo import models, fields
from odoo import fields, models
_logger = logging.getLogger(__name__)
class PrintingJob(models.Model):
_name = 'printing.job'
_description = 'Printing Job'
_order = 'job_id_cups DESC'
_name = "printing.job"
_description = "Printing Job"
_order = "job_id_cups DESC"
name = fields.Char(help='Job name.')
name = fields.Char(help="Job name.")
active = fields.Boolean(
default=True, help='Unchecked if the job is purged from cups.')
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.')
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.')
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.')
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.')
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.')
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.')
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 !'),
(
"job_id_cups_unique",
"UNIQUE(job_id_cups, server_id)",
"The id of the job must be unique per server !",
)
]
def action_cancel(self):
@@ -98,7 +121,6 @@ class PrintingJob(models.Model):
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)
self.mapped("server_id").update_jobs(which="all", first_job_id=job.job_id_cups)
return True

View File

@@ -11,15 +11,14 @@ import logging
import os
from tempfile import mkstemp
from odoo import models, fields
from odoo import fields, models
_logger = logging.getLogger(__name__)
try:
import cups
except ImportError:
_logger.debug('Cannot `import cups`.')
_logger.debug("Cannot `import cups`.")
class PrintingPrinter(models.Model):
@@ -27,63 +26,65 @@ class PrintingPrinter(models.Model):
Printers
"""
_name = 'printing.printer'
_description = 'Printer'
_order = 'name'
_name = "printing.printer"
_description = "Printer"
_order = "name"
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.')
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.')
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(
selection=[
('unavailable', 'Unavailable'),
('printing', 'Printing'),
('unknown', 'Unknown'),
('available', 'Available'),
('error', 'Error'),
('server-error', 'Server Error'),
("unavailable", "Unavailable"),
("printing", "Printing"),
("unknown", "Unknown"),
("available", "Available"),
("error", "Error"),
("server-error", "Server Error"),
],
required=True,
readonly=True,
default='unknown')
default="unknown",
)
status_message = fields.Char(readonly=True)
model = fields.Char(readonly=True)
location = fields.Char(readonly=True)
uri = fields.Char(string='URI', readonly=True)
tray_ids = fields.One2many(comodel_name='printing.tray',
inverse_name='printer_id',
string='Paper Sources')
uri = fields.Char(string="URI", readonly=True)
tray_ids = fields.One2many(
comodel_name="printing.tray", inverse_name="printer_id", string="Paper Sources"
)
def _prepare_update_from_cups(self, cups_connection, cups_printer):
mapping = {
3: 'available',
4: 'printing',
5: 'error'
}
mapping = {3: "available", 4: "printing", 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.get(
'printer-state'), 'unknown'),
'status_message': cups_printer.get('printer-state-message', ''),
"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.get("printer-state"), "unknown"),
"status_message": cups_printer.get("printer-state-message", ""),
}
printer_uri = cups_printer['printer-uri-supported']
printer_system_name = printer_uri[printer_uri.rfind('/') + 1:]
printer_uri = cups_printer["printer-uri-supported"]
printer_system_name = printer_uri[printer_uri.rfind("/") + 1 :]
ppd_info = cups_connection.getPPD3(printer_system_name)
ppd_path = ppd_info[2]
if not ppd_path:
return vals
ppd = cups.PPD(ppd_path)
option = ppd.findOption('InputSlot')
option = ppd.findOption("InputSlot")
try:
os.unlink(ppd_path)
except OSError as err:
@@ -94,25 +95,29 @@ class PrintingPrinter(models.Model):
if not option:
return vals
vals['tray_ids'] = []
vals["tray_ids"] = []
cups_trays = {
tray_option['choice']: tray_option['text']
for tray_option in option.choices
tray_option["choice"]: tray_option["text"] for tray_option in option.choices
}
# Add new trays
vals['tray_ids'].extend([
(0, 0, {'name': text, 'system_name': choice})
for choice, text in cups_trays.items()
if choice not in self.tray_ids.mapped('system_name')
])
vals["tray_ids"].extend(
[
(0, 0, {"name": text, "system_name": choice})
for choice, text in cups_trays.items()
if choice not in self.tray_ids.mapped("system_name")
]
)
# Remove deleted trays
vals['tray_ids'].extend([
(2, tray.id)
for tray in self.tray_ids.filtered(
lambda record: record.system_name not in cups_trays.keys())
])
vals["tray_ids"].extend(
[
(2, tray.id)
for tray in self.tray_ids.filtered(
lambda record: record.system_name not in cups_trays.keys()
)
]
)
return vals
def print_document(self, report, content, **print_opts):
@@ -126,12 +131,11 @@ class PrintingPrinter(models.Model):
finally:
os.close(fd)
return self.print_file(
file_name, report=report, **print_opts)
return self.print_file(file_name, report=report, **print_opts)
@staticmethod
def _set_option_doc_format(report, value):
return {'raw': 'True'} if value == 'raw' else {}
return {"raw": "True"} if value == "raw" else {}
# Backwards compatibility of builtin used as kwarg
_set_option_format = _set_option_doc_format
@@ -139,7 +143,7 @@ class PrintingPrinter(models.Model):
def _set_option_tray(self, report, value):
"""Note we use self here as some older PPD use tray
rather than InputSlot so we may need to query printer in override"""
return {'InputSlot': str(value)} if value else {}
return {"InputSlot": str(value)} if value else {}
@staticmethod
def _set_option_noop(report, value):
@@ -152,8 +156,7 @@ class PrintingPrinter(models.Model):
options = {}
for option, value in print_opts.items():
try:
options.update(getattr(
self, '_set_option_%s' % option)(report, value))
options.update(getattr(self, "_set_option_%s" % option)(report, value))
except AttributeError:
options[option] = str(value)
return options
@@ -165,33 +168,30 @@ class PrintingPrinter(models.Model):
options = self.print_options(report=report, **print_opts)
_logger.debug(
'Sending job to CUPS printer %s on %s'
% (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,
self.server_id.address,
))
"Sending job to CUPS printer %s on %s"
% (self.system_name, self.server_id.address)
)
connection.printFile(self.system_name, file_name, file_name, options=options)
_logger.info(
"Printing job: '{}' on {}".format(file_name, self.server_id.address)
)
return True
def set_default(self):
if not self:
return
self.ensure_one()
default_printers = self.search([('default', '=', True)])
default_printers = self.search([("default", "=", True)])
default_printers.unset_default()
self.write({'default': True})
self.write({"default": True})
return True
def unset_default(self):
self.write({'default': False})
self.write({"default": False})
return True
def get_default(self):
return self.search([('default', '=', True)], limit=1)
return self.search([("default", "=", True)], limit=1)
def action_cancel_all_jobs(self):
self.ensure_one()
@@ -200,11 +200,10 @@ class PrintingPrinter(models.Model):
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)
connection.cancelAllJobs(name=printer.system_name, purge_jobs=purge_jobs)
# Update jobs' states into Odoo
self.mapped('server_id').update_jobs(which='completed')
self.mapped("server_id").update_jobs(which="completed")
return True
@@ -214,7 +213,7 @@ class PrintingPrinter(models.Model):
connection.enablePrinter(printer.system_name)
# Update printers' stats into Odoo
self.mapped('server_id').update_printers()
self.mapped("server_id").update_printers()
return True
@@ -224,6 +223,6 @@ class PrintingPrinter(models.Model):
connection.disablePrinter(printer.system_name)
# Update printers' stats into Odoo
self.mapped('server_id').update_printers()
self.mapped("server_id").update_printers()
return True

View File

@@ -5,35 +5,35 @@
# Copyright (C) 2013-2014 Camptocamp (<http://www.camptocamp.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import models, fields, api
from odoo import api, fields, models
class PrintingReportXmlAction(models.Model):
_name = 'printing.report.xml.action'
_description = 'Printing Report Printing Actions'
_name = "printing.report.xml.action"
_description = "Printing Report Printing Actions"
report_id = fields.Many2one(comodel_name='ir.actions.report',
string='Report',
required=True,
ondelete='cascade')
user_id = fields.Many2one(comodel_name='res.users',
string='User',
required=True,
ondelete='cascade')
report_id = fields.Many2one(
comodel_name="ir.actions.report",
string="Report",
required=True,
ondelete="cascade",
)
user_id = fields.Many2one(
comodel_name="res.users", string="User", required=True, ondelete="cascade"
)
action = fields.Selection(
selection=lambda s: s.env['printing.action']._available_action_types(),
selection=lambda s: s.env["printing.action"]._available_action_types(),
required=True,
)
printer_id = fields.Many2one(comodel_name='printing.printer',
string='Printer')
printer_id = fields.Many2one(comodel_name="printing.printer", string="Printer")
printer_tray_id = fields.Many2one(
comodel_name='printing.tray',
string='Paper Source',
comodel_name="printing.tray",
string="Paper Source",
domain="[('printer_id', '=', printer_id)]",
)
@api.onchange('printer_id')
@api.onchange("printer_id")
def onchange_printer_id(self):
""" Reset the tray when the printer is changed """
self.printer_tray_id = False
@@ -42,7 +42,7 @@ class PrintingReportXmlAction(models.Model):
if not self:
return {}
return {
'action': self.action,
'printer': self.printer_id,
'tray': self.printer_tray_id.system_name
"action": self.action,
"printer": self.printer_id,
"tray": self.printer_tray_id.system_name,
}

View File

@@ -3,7 +3,8 @@
import logging
from datetime import datetime
from odoo import models, fields, exceptions, _
from odoo import _, exceptions, fields, models
_logger = logging.getLogger(__name__)
@@ -11,26 +12,25 @@ _logger = logging.getLogger(__name__)
try:
import cups
except ImportError:
_logger.debug('Cannot `import cups`.')
_logger.debug("Cannot `import cups`.")
class PrintingServer(models.Model):
_name = 'printing.server'
_description = 'Printing server'
_name = "printing.server"
_description = "Printing server"
name = fields.Char(
default='Localhost', required=True, help='Name of the 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.')
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.')
comodel_name="printing.printer",
inverse_name="server_id",
string="Printers List",
help="List of printers available on this server.",
)
def _open_connection(self, raise_on_error=False):
self.ensure_one()
@@ -38,10 +38,11 @@ class PrintingServer(models.Model):
try:
connection = cups.Connection(host=self.address, port=self.port)
except Exception:
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)
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)
@@ -63,28 +64,25 @@ class PrintingServer(models.Model):
for server in servers:
connection = server._open_connection(raise_on_error=raise_on_error)
if not connection:
server.printer_ids.write({'status': 'server-error'})
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
])
existing_printers = {
printer.system_name: printer for printer in server.printer_ids
}
updated_printers = []
for name, printer_info in printers.items():
printer = self.env['printing.printer']
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,
connection, printer_info
)
printer_values.update(system_name=name, server_id=server.id)
updated_printers.append(name)
if not printer:
printer.create(printer_values)
@@ -93,8 +91,8 @@ class PrintingServer(models.Model):
# Set printers not found as unavailable
server.printer_ids.filtered(
lambda record: record.system_name not in updated_printers)\
.write({'status': 'unavailable'})
lambda record: record.system_name not in updated_printers
).write({"status": "unavailable"})
return res
@@ -103,18 +101,18 @@ class PrintingServer(models.Model):
self = self.search([])
return self.update_jobs()
def update_jobs(self, which='all', first_job_id=-1):
job_obj = self.env['printing.job']
printer_obj = self.env['printing.printer']
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',
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
@@ -127,86 +125,91 @@ class PrintingServer(models.Model):
# Retrieve asked job data
jobs_data = connection.getJobs(
which_jobs=which, first_job_id=first_job_id,
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',
])
"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 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',
]))
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),
])
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', False),
'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', False))),
'time_at_processing': job_data.get(
'time-at-processing', False) and
fields.Datetime.to_string(datetime.fromtimestamp(
job_data.get('time-at-processing', False))),
'time_at_completed': job_data.get(
'time-at-completed', False) and
fields.Datetime.to_string(datetime.fromtimestamp(
job_data.get('time-at-completed', False))),
"name": job_data.get("job-name", ""),
"active": True,
"job_id_cups": cups_job_id,
"job_media_progress": job_data.get("job-media-progress", False),
"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", False))
),
"time_at_processing": job_data.get("time-at-processing", False)
and fields.Datetime.to_string(
datetime.fromtimestamp(
job_data.get("time-at-processing", False)
)
),
"time_at_completed": job_data.get("time-at-completed", False)
and fields.Datetime.to_string(
datetime.fromtimestamp(job_data.get("time-at-completed", False))
),
}
# Search for the printer in Odoo
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)
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,
)
# CUPS retains jobs for disconnected printers and also may
# leak jobs data for unshared printers, therefore we just
# discard here if not printer found
if not printer:
continue
job_values['printer_id'] = printer.id
job_values["printer_id"] = printer.id
if jobs:
jobs.write(job_values)
@@ -214,10 +217,10 @@ class PrintingServer(models.Model):
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})
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

View File

@@ -5,17 +5,17 @@ from odoo import fields, models
class PrinterTray(models.Model):
_name = 'printing.tray'
_description = 'Printer Tray'
_name = "printing.tray"
_description = "Printer Tray"
_order = 'name asc'
_order = "name asc"
name = fields.Char(required=True)
system_name = fields.Char(required=True, readonly=True)
printer_id = fields.Many2one(
comodel_name='printing.printer',
string='Printer',
comodel_name="printing.printer",
string="Printer",
required=True,
readonly=True,
ondelete='cascade',
ondelete="cascade",
)

View File

@@ -5,46 +5,38 @@
# Copyright (C) 2013-2014 Camptocamp (<http://www.camptocamp.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import models, fields, api
from odoo import api, fields, models
class ResUsers(models.Model):
_inherit = 'res.users'
_inherit = "res.users"
@api.model
def _user_available_action_types(self):
return [
(code, string)
for code, string
in self.env['printing.action']._available_action_types()
if code != 'user_default'
for code, string in self.env["printing.action"]._available_action_types()
if code != "user_default"
]
printing_action = fields.Selection(
selection=_user_available_action_types,
printing_action = fields.Selection(selection=_user_available_action_types)
printing_printer_id = fields.Many2one(
comodel_name="printing.printer", string="Default Printer"
)
printing_printer_id = fields.Many2one(comodel_name='printing.printer',
string='Default Printer')
@api.model
def _register_hook(self):
super()._register_hook()
self.SELF_WRITEABLE_FIELDS.extend([
'printing_action',
'printing_printer_id',
])
self.SELF_READABLE_FIELDS.extend([
'printing_action',
'printing_printer_id',
])
self.SELF_WRITEABLE_FIELDS.extend(["printing_action", "printing_printer_id"])
self.SELF_READABLE_FIELDS.extend(["printing_action", "printing_printer_id"])
printer_tray_id = fields.Many2one(
comodel_name='printing.tray',
string='Default Printer Paper Source',
comodel_name="printing.tray",
string="Default Printer Paper Source",
domain="[('printer_id', '=', printing_printer_id)]",
)
@api.onchange('printing_printer_id')
@api.onchange("printing_printer_id")
def onchange_printing_printer_id(self):
""" Reset the tray when the printer is changed """
self.printer_tray_id = False

View File

@@ -1,6 +1,6 @@
<?xml version="1.0" ?>
<odoo>
<record id="printing_group_user" model="res.groups">
<field name="name">Printing / Print User</field>
</record>

View File

@@ -32,8 +32,8 @@ odoo.define('base_report_to_printer.print', function (require) {
}, function () {
self.do_notify(_t('Report'),
_.str.sprintf(
_t('Error when sending the document\
to the printer '),
_t('Error when sending the document ' +
'to the printer '),
print_action.printer_name
)
);
@@ -48,4 +48,3 @@ odoo.define('base_report_to_printer.print', function (require) {
},
});
});

View File

@@ -8,79 +8,74 @@ from odoo.tests.common import TransactionCase
class TestIrActionsReportXml(TransactionCase):
def setUp(self):
super(TestIrActionsReportXml, self).setUp()
self.Model = self.env['ir.actions.report']
self.Model = self.env["ir.actions.report"]
self.vals = {}
self.report = self.env['ir.actions.report'].search([], limit=1)
self.server = self.env['printing.server'].create({})
self.report = self.env["ir.actions.report"].search([], limit=1)
self.server = self.env["printing.server"].create({})
def new_action(self):
return self.env['printing.action'].create({
'name': 'Printing Action',
'action_type': 'server',
})
return self.env["printing.action"].create(
{"name": "Printing Action", "action_type": "server"}
)
def new_printing_action(self):
return self.env['printing.report.xml.action'].create({
'report_id': self.report.id,
'user_id': self.env.ref('base.user_demo').id,
'action': 'server',
})
return self.env["printing.report.xml.action"].create(
{
"report_id": self.report.id,
"user_id": self.env.ref("base.user_demo").id,
"action": "server",
}
)
def new_printer(self):
return self.env['printing.printer'].create({
'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',
})
return self.env["printing.printer"].create(
{
"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",
}
)
def new_tray(self, vals=None, defaults=None):
values = dict(defaults)
if vals is not None:
values.update(vals)
return self.env['printing.tray'].create(values)
return self.env["printing.tray"].create(values)
def test_print_action_for_report_name_gets_report(self):
""" It should get report by name """
with mock.patch.object(self.Model, '_get_report_from_name') as mk:
expect = 'test'
with mock.patch.object(self.Model, "_get_report_from_name") as mk:
expect = "test"
self.Model.print_action_for_report_name(expect)
mk.assert_called_once_with(
expect
)
mk.assert_called_once_with(expect)
def test_print_action_for_report_name_returns_if_no_report(self):
""" It should return empty dict when no matching report """
with mock.patch.object(self.Model, '_get_report_from_name') as mk:
expect = 'test'
with mock.patch.object(self.Model, "_get_report_from_name") as mk:
expect = "test"
mk.return_value = False
res = self.Model.print_action_for_report_name(expect)
self.assertDictEqual(
{}, res,
)
self.assertDictEqual({}, res)
def test_print_action_for_report_name_returns_if_report(self):
""" It should return correct serializable result for behaviour """
with mock.patch.object(self.Model, '_get_report_from_name') as mk:
res = self.Model.print_action_for_report_name('test')
with mock.patch.object(self.Model, "_get_report_from_name") as mk:
res = self.Model.print_action_for_report_name("test")
behaviour = mk().behaviour()
expect = {
'action': behaviour['action'],
'printer_name': behaviour['printer'].name,
"action": behaviour["action"],
"printer_name": behaviour["printer"].name,
}
self.assertDictEqual(
expect, res,
'Expect %s, Got %s' % (expect, res),
)
self.assertDictEqual(expect, res, "Expect {}, Got {}".format(expect, res))
def test_behaviour_default_values(self):
""" It should return the default action and printer """
@@ -89,81 +84,81 @@ class TestIrActionsReportXml(TransactionCase):
self.env.user.printing_printer_id = False
report.property_printing_action_id = False
report.printing_printer_id = False
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': self.env['printing.printer'],
'tray': False,
},
self.assertEqual(
report.behaviour(),
{
"action": "client",
"printer": self.env["printing.printer"],
"tray": False,
},
)
def test_behaviour_user_values(self):
""" It should return the action and printer from user """
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
self.env.user.printing_printer_id = self.new_printer()
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': self.env.user.printing_printer_id,
'tray': False,
},
self.assertEqual(
report.behaviour(),
{
"action": "client",
"printer": self.env.user.printing_printer_id,
"tray": False,
},
)
def test_behaviour_report_values(self):
""" It should return the action and printer from report """
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
report.property_printing_action_id = self.new_action()
report.printing_printer_id = self.new_printer()
self.assertEqual(report.behaviour(), {
'action': report.property_printing_action_id.action_type,
'printer': report.printing_printer_id,
'tray': False,
},
self.assertEqual(
report.behaviour(),
{
"action": report.property_printing_action_id.action_type,
"printer": report.printing_printer_id,
"tray": False,
},
)
def test_behaviour_user_action(self):
""" It should return the action and printer from user action"""
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
report.property_printing_action_id.action_type = 'user_default'
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': report.printing_printer_id,
'tray': False,
},
self.env.user.printing_action = "client"
report.property_printing_action_id.action_type = "user_default"
self.assertEqual(
report.behaviour(),
{"action": "client", "printer": report.printing_printer_id, "tray": False},
)
def test_behaviour_printing_action_on_wrong_user(self):
""" It should return the action and printer ignoring printing action
"""
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
printing_action = self.new_printing_action()
printing_action.user_id = self.env['res.users'].search([
('id', '!=', self.env.user.id),
], limit=1)
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': report.printing_printer_id,
'tray': False,
},
printing_action.user_id = self.env["res.users"].search(
[("id", "!=", self.env.user.id)], limit=1
)
self.assertEqual(
report.behaviour(),
{"action": "client", "printer": report.printing_printer_id, "tray": False},
)
def test_behaviour_printing_action_on_wrong_report(self):
""" It should return the action and printer ignoring printing action
"""
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
printing_action = self.new_printing_action()
printing_action.user_id = self.env.user
printing_action.report_id = self.env['ir.actions.report'].search([
('id', '!=', report.id),
], limit=1)
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': report.printing_printer_id,
'tray': False,
},
printing_action.report_id = self.env["ir.actions.report"].search(
[("id", "!=", report.id)], limit=1
)
self.assertEqual(
report.behaviour(),
{"action": "client", "printer": report.printing_printer_id, "tray": False},
)
def test_behaviour_printing_action_with_no_printer(self):
@@ -171,29 +166,33 @@ class TestIrActionsReportXml(TransactionCase):
other
"""
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
printing_action = self.new_printing_action()
printing_action.user_id = self.env.user
printing_action.report_id = report
self.assertEqual(report.behaviour(), {
'action': printing_action.action,
'printer': report.printing_printer_id,
'tray': False,
},
self.assertEqual(
report.behaviour(),
{
"action": printing_action.action,
"printer": report.printing_printer_id,
"tray": False,
},
)
def test_behaviour_printing_action_with_printer(self):
""" It should return the action and printer from printing action """
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
printing_action = self.new_printing_action()
printing_action.user_id = self.env.user
printing_action.printer_id = self.new_printer()
self.assertEqual(report.behaviour(), {
'action': printing_action.action,
'printer': printing_action.printer_id,
'tray': False,
},
self.assertEqual(
report.behaviour(),
{
"action": printing_action.action,
"printer": printing_action.printer_id,
"tray": False,
},
)
def test_behaviour_printing_action_user_defaults(self):
@@ -201,106 +200,96 @@ class TestIrActionsReportXml(TransactionCase):
action
"""
report = self.Model.search([], limit=1)
self.env.user.printing_action = 'client'
self.env.user.printing_action = "client"
printing_action = self.new_printing_action()
printing_action.user_id = self.env.user
printing_action.action = 'user_default'
self.assertEqual(report.behaviour(), {
'action': 'client',
'printer': report.printing_printer_id,
'tray': False,
},
printing_action.action = "user_default"
self.assertEqual(
report.behaviour(),
{"action": "client", "printer": report.printing_printer_id, "tray": False},
)
def test_print_tray_behaviour(self):
"""
It should return the correct tray
"""
report = self.env['ir.actions.report'].search([], limit=1)
action = self.env['printing.report.xml.action'].create({
'user_id': self.env.user.id,
'report_id': report.id,
'action': 'server',
})
report = self.env["ir.actions.report"].search([], limit=1)
action = self.env["printing.report.xml.action"].create(
{"user_id": self.env.user.id, "report_id": report.id, "action": "server"}
)
printer = self.new_printer()
tray_vals = {
'name': 'Tray',
'system_name': 'Tray',
'printer_id': printer.id,
}
user_tray = self.new_tray({'system_name': 'User tray'}, tray_vals)
report_tray = self.new_tray({'system_name': 'Report tray'}, tray_vals)
action_tray = self.new_tray({'system_name': 'Action tray'}, tray_vals)
tray_vals = {"name": "Tray", "system_name": "Tray", "printer_id": printer.id}
user_tray = self.new_tray({"system_name": "User tray"}, tray_vals)
report_tray = self.new_tray({"system_name": "Report tray"}, tray_vals)
action_tray = self.new_tray({"system_name": "Action tray"}, tray_vals)
# No report passed
self.env.user.printer_tray_id = False
options = printer.print_options()
self.assertFalse('InputSlot' in options)
self.assertFalse("InputSlot" in options)
# No tray defined
self.env.user.printer_tray_id = False
report.printer_tray_id = False
action.printer_tray_id = False
options = report.behaviour()
self.assertTrue('tray' in options)
self.assertTrue("tray" in options)
# Only user tray is defined
self.env.user.printer_tray_id = user_tray
report.printer_tray_id = False
action.printer_tray_id = False
self.assertEqual('User tray', report.behaviour()['tray'])
self.assertEqual("User tray", report.behaviour()["tray"])
# Only report tray is defined
self.env.user.printer_tray_id = False
report.printer_tray_id = report_tray
action.printer_tray_id = False
self.assertEqual('Report tray', report.behaviour()['tray'])
self.assertEqual("Report tray", report.behaviour()["tray"])
# Only action tray is defined
self.env.user.printer_tray_id = False
report.printer_tray_id = False
action.printer_tray_id = action_tray
self.assertEqual('Action tray', report.behaviour()['tray'])
self.assertEqual("Action tray", report.behaviour()["tray"])
# User and report tray defined
self.env.user.printer_tray_id = user_tray
report.printer_tray_id = report_tray
action.printer_tray_id = False
self.assertEqual('Report tray', report.behaviour()['tray'])
self.assertEqual("Report tray", report.behaviour()["tray"])
# All trays are defined
self.env.user.printer_tray_id = user_tray
report.printer_tray_id = report_tray
action.printer_tray_id = action_tray
self.assertEqual('Action tray', report.behaviour()['tray'])
self.assertEqual("Action tray", report.behaviour()["tray"])
def test_onchange_printer_tray_id_empty(self):
action = self.env['ir.actions.report'].new(
{'printer_tray_id': False})
action = self.env["ir.actions.report"].new({"printer_tray_id": False})
action.onchange_printing_printer_id()
self.assertFalse(action.printer_tray_id)
def test_onchange_printer_tray_id_not_empty(self):
server = self.env['printing.server'].create({})
printer = self.env['printing.printer'].create({
'name': 'Printer',
'server_id': server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
})
tray = self.env['printing.tray'].create({
'name': 'Tray',
'system_name': 'TrayName',
'printer_id': printer.id,
})
server = self.env["printing.server"].create({})
printer = self.env["printing.printer"].create(
{
"name": "Printer",
"server_id": server.id,
"system_name": "Sys Name",
"default": True,
"status": "unknown",
"status_message": "Msg",
"model": "res.users",
"location": "Location",
"uri": "URI",
}
)
tray = self.env["printing.tray"].create(
{"name": "Tray", "system_name": "TrayName", "printer_id": printer.id}
)
action = self.env['ir.actions.report'].new(
{'printer_tray_id': tray.id})
action = self.env["ir.actions.report"].new({"printer_tray_id": tray.id})
self.assertEqual(action.printer_tray_id, tray)
action.onchange_printing_printer_id()
self.assertFalse(action.printer_tray_id)

View File

@@ -6,60 +6,58 @@ import mock
from odoo import fields
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.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.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',
"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(),
"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)
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)
values["printer_id"] = printer.id
return self.env["printing.job"].create(values)
@mock.patch('%s.cups' % model)
@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 = self.new_job(printer, {"job_id_cups": 2})
job.action_cancel()
cups.Connection.side_effect = None
self.assertEqual(cups.Connection().cancelJob.call_count, 0)
@mock.patch('%s.cups' % model)
@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,
job.job_id_cups, purge_job=False
)

View File

@@ -2,35 +2,34 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import tempfile
import mock
from odoo.exceptions import UserError
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_printer'
server_model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.addons.base_report_to_printer.models.printing_printer"
server_model = "odoo.addons.base_report_to_printer.models.printing_server"
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.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',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
"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.report = self.env['ir.actions.report'].search([], limit=1)
self.report = self.env["ir.actions.report"].search([], limit=1)
def new_record(self):
return self.Model.create(self.printer_vals)
@@ -39,94 +38,88 @@ class TestPrintingPrinter(TransactionCase):
"""
It should put the value in InputSlot
"""
self.assertEqual(self.Model._set_option_tray(None, 'Test Tray'),
{'InputSlot': 'Test Tray'})
self.assertEqual(self.Model._set_option_tray(None, False),
{})
self.assertEqual(
self.Model._set_option_tray(None, "Test Tray"), {"InputSlot": "Test Tray"}
)
self.assertEqual(self.Model._set_option_tray(None, False), {})
def test_option_noops(self):
"""
Noops should return an empty dict
"""
self.assertEqual(self.Model._set_option_action(None, 'printer'), {})
self.assertEqual(self.Model._set_option_action(None, "printer"), {})
self.assertEqual(self.Model._set_option_printer(None, self.Model), {})
def test_option_doc_format(self):
"""
Raw documents should set raw boolean.
"""
self.assertEqual(self.Model._set_option_doc_format(None, 'raw'),
{'raw': 'True'})
self.assertEqual(
self.Model._set_option_doc_format(None, "raw"), {"raw": "True"}
)
# Deprecate _set_option_format in v12.
self.assertEqual(self.Model._set_option_format(None, 'raw'),
{'raw': 'True'})
self.assertEqual(self.Model._set_option_format(None, "raw"), {"raw": "True"})
self.assertEqual(self.Model._set_option_doc_format(None, 'pdf'), {})
self.assertEqual(self.Model._set_option_doc_format(None, "pdf"), {})
# Deprecate _set_option_format in v12.
self.assertEqual(self.Model._set_option_format(None, 'pdf'), {})
self.assertEqual(self.Model._set_option_format(None, "pdf"), {})
def test_print_options(self):
""" It should generate the right options dictionnary """
# TODO: None here used as report - tests here should be merged
# with tests in test_printing_printer_tray from when modules merged
report = self.env['ir.actions.report'].search([], limit=1)
self.assertEqual(self.Model.print_options(
doc_format='raw'), {'raw': 'True'}
report = self.env["ir.actions.report"].search([], limit=1)
self.assertEqual(self.Model.print_options(doc_format="raw"), {"raw": "True"})
self.assertEqual(
self.Model.print_options(report, doc_format="pdf", copies=2),
{"copies": "2"},
)
self.assertEqual(self.Model.print_options(
report, doc_format='pdf', copies=2), {'copies': '2'}
self.assertEqual(
self.Model.print_options(report, doc_format="raw", copies=2),
{"raw": "True", "copies": "2"},
)
self.assertEqual(self.Model.print_options(
report, doc_format='raw', copies=2),
{'raw': 'True', 'copies': '2'}
)
self.assertTrue('InputSlot' in self.Model.print_options(report,
tray='Test'))
self.assertTrue("InputSlot" in self.Model.print_options(report, tray="Test"))
@mock.patch('%s.cups' % server_model)
@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:
with mock.patch("%s.mkstemp" % model) as mkstemp:
mkstemp.return_value = fd, file_name
printer = self.new_record()
printer.print_document(self.report, b'content to print',
doc_format='pdf')
printer.print_document(self.report, b"content to print", doc_format="pdf")
cups.Connection().printFile.assert_called_once_with(
printer.system_name,
file_name,
file_name,
options={})
printer.system_name, file_name, file_name, options={}
)
@mock.patch('%s.cups' % server_model)
@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:
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(
self.report, b'content to print', doc_format='pdf')
self.report, b"content to print", doc_format="pdf"
)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_print_file(self, cups):
""" It should print a file through CUPS """
file_name = 'file_name'
file_name = "file_name"
printer = self.new_record()
printer.print_file(file_name, 'pdf')
printer.print_file(file_name, "pdf")
cups.Connection().printFile.assert_called_once_with(
printer.system_name,
file_name,
file_name,
options={})
printer.system_name, file_name, file_name, options={}
)
@mock.patch('%s.cups' % server_model)
@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'
file_name = "file_name"
printer = self.new_record()
with self.assertRaises(UserError):
printer.print_file(file_name)
@@ -150,38 +143,34 @@ class TestPrintingPrinter(TransactionCase):
printer.unset_default()
self.assertFalse(printer.default)
@mock.patch('%s.cups' % server_model)
@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,
name=printer.system_name, purge_jobs=False
)
@mock.patch('%s.cups' % server_model)
@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,
name=printer.system_name, purge_jobs=True
)
@mock.patch('%s.cups' % server_model)
@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)
cups.Connection().enablePrinter.assert_called_once_with(printer.system_name)
@mock.patch('%s.cups' % server_model)
@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)
cups.Connection().disablePrinter.assert_called_once_with(printer.system_name)

View File

@@ -2,13 +2,14 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import errno
import mock
import tempfile
import mock
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_printer'
server_model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.addons.base_report_to_printer.models.printing_printer"
server_model = "odoo.addons.base_report_to_printer.models.printing_server"
ppd_header = '*PPD-Adobe: "4.3"'
ppd_input_slot_header = """
@@ -33,34 +34,35 @@ ppd_input_slot_footer = """
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 = self.env['printing.printer'].create({
'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.Model = self.env["printing.printer"]
self.ServerModel = self.env["printing.server"]
self.server = self.env["printing.server"].create({})
self.printer = self.env["printing.printer"].create(
{
"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.tray_vals = {
'name': 'Tray',
'system_name': 'TrayName',
'printer_id': self.printer.id,
"name": "Tray",
"system_name": "TrayName",
"printer_id": self.printer.id,
}
def new_tray(self, vals=None):
values = self.tray_vals
if vals is not None:
values.update(vals)
return self.env['printing.tray'].create(values)
return self.env["printing.tray"].create(values)
def build_ppd(self, input_slots=None):
"""
@@ -71,8 +73,7 @@ class TestPrintingPrinter(TransactionCase):
if input_slots is not None:
for input_slot in input_slots:
ppd_contents += ppd_input_slot_body.format(
name=input_slot['name'],
text=input_slot['text'],
name=input_slot["name"], text=input_slot["text"]
)
ppd_contents += ppd_input_slot_footer
@@ -88,29 +89,29 @@ class TestPrintingPrinter(TransactionCase):
if file_name:
ppd_contents = self.build_ppd(input_slots=input_slots)
with open(file_name, 'w') as fp:
with open(file_name, "w") as fp:
fp.write(ppd_contents)
cups.Connection().getPPD3.return_value = (200, 0, file_name)
cups.Connection().getPrinters.return_value = {
self.printer.system_name: {
'printer-info': 'info',
'printer-uri-supported': 'uri',
},
"printer-info": "info",
"printer-uri-supported": "uri",
}
}
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_update_printers(self, cups):
"""
Check that the update_printers method calls _prepare_update_from_cups
"""
self.mock_cups_ppd(cups, file_name=False)
self.assertEqual(self.printer.name, 'Printer')
self.assertEqual(self.printer.name, "Printer")
self.ServerModel.update_printers()
self.assertEqual(self.printer.name, 'info')
self.assertEqual(self.printer.name, "info")
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups_no_ppd(self, cups):
"""
Check that the tray_ids field has no value when no PPD is available
@@ -121,9 +122,9 @@ class TestPrintingPrinter(TransactionCase):
cups_printer = connection.getPrinters()[self.printer.system_name]
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertFalse('tray_ids' in vals)
self.assertFalse("tray_ids" in vals)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups_empty_ppd(self, cups):
"""
Check that the tray_ids field has no value when the PPD file has
@@ -132,23 +133,23 @@ class TestPrintingPrinter(TransactionCase):
fd, file_name = tempfile.mkstemp()
self.mock_cups_ppd(cups, file_name=file_name)
# Replace the ppd file's contents by an empty file
with open(file_name, 'w') as fp:
with open(file_name, "w") as fp:
fp.write(ppd_header)
connection = cups.Connection()
cups_printer = connection.getPrinters()[self.printer.system_name]
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertFalse('tray_ids' in vals)
self.assertFalse("tray_ids" in vals)
@mock.patch('%s.cups' % server_model)
@mock.patch('os.unlink')
@mock.patch("%s.cups" % server_model)
@mock.patch("os.unlink")
def test_prepare_update_from_cups_unlink_error(self, os_unlink, cups):
"""
When OSError other than ENOENT is encountered, the exception is raised
"""
# Break os.unlink
os_unlink.side_effect = OSError(errno.EIO, 'Error')
os_unlink.side_effect = OSError(errno.EIO, "Error")
self.mock_cups_ppd(cups)
@@ -158,17 +159,16 @@ class TestPrintingPrinter(TransactionCase):
with self.assertRaises(OSError):
self.printer._prepare_update_from_cups(connection, cups_printer)
@mock.patch('%s.cups' % server_model)
@mock.patch('os.unlink')
def test_prepare_update_from_cups_unlink_error_enoent(
self, os_unlink, cups):
@mock.patch("%s.cups" % server_model)
@mock.patch("os.unlink")
def test_prepare_update_from_cups_unlink_error_enoent(self, os_unlink, cups):
"""
When a ENOENT error is encountered, the file has already been unlinked
This is not an issue, as we were trying to delete the file.
The update can continue.
"""
# Break os.unlink
os_unlink.side_effect = OSError(errno.ENOENT, 'Error')
os_unlink.side_effect = OSError(errno.ENOENT, "Error")
self.mock_cups_ppd(cups)
@@ -176,12 +176,12 @@ class TestPrintingPrinter(TransactionCase):
cups_printer = connection.getPrinters()[self.printer.system_name]
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertEqual(vals['tray_ids'], [(0, 0, {
'name': 'Auto (Default)',
'system_name': 'Auto',
})])
self.assertEqual(
vals["tray_ids"],
[(0, 0, {"name": "Auto (Default)", "system_name": "Auto"})],
)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups(self, cups):
"""
Check the return value when adding a single tray
@@ -192,55 +192,51 @@ class TestPrintingPrinter(TransactionCase):
cups_printer = connection.getPrinters()[self.printer.system_name]
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertEqual(vals['tray_ids'], [(0, 0, {
'name': 'Auto (Default)',
'system_name': 'Auto',
})])
self.assertEqual(
vals["tray_ids"],
[(0, 0, {"name": "Auto (Default)", "system_name": "Auto"})],
)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups_with_multiple_trays(self, cups):
"""
Check the return value when adding multiple trays at once
"""
self.mock_cups_ppd(cups, input_slots=[
{'name': 'Tray1', 'text': 'Tray 1'},
])
self.mock_cups_ppd(cups, input_slots=[{"name": "Tray1", "text": "Tray 1"}])
connection = cups.Connection()
cups_printer = connection.getPrinters()[self.printer.system_name]
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertItemsEqual(vals['tray_ids'], [(0, 0, {
'name': 'Auto (Default)',
'system_name': 'Auto',
}), (0, 0, {
'name': 'Tray 1',
'system_name': 'Tray1',
})])
self.assertItemsEqual(
vals["tray_ids"],
[
(0, 0, {"name": "Auto (Default)", "system_name": "Auto"}),
(0, 0, {"name": "Tray 1", "system_name": "Tray1"}),
],
)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups_already_known_trays(self, cups):
"""
Check that calling the method twice doesn't create the trays multiple
times
"""
self.mock_cups_ppd(cups, input_slots=[
{'name': 'Tray1', 'text': 'Tray 1'},
])
self.mock_cups_ppd(cups, input_slots=[{"name": "Tray1", "text": "Tray 1"}])
connection = cups.Connection()
cups_printer = connection.getPrinters()[self.printer.system_name]
# Create a tray which is in the PPD file
self.new_tray({'system_name': 'Tray1'})
self.new_tray({"system_name": "Tray1"})
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertEqual(vals['tray_ids'], [(0, 0, {
'name': 'Auto (Default)',
'system_name': 'Auto',
})])
self.assertEqual(
vals["tray_ids"],
[(0, 0, {"name": "Auto (Default)", "system_name": "Auto"})],
)
@mock.patch('%s.cups' % server_model)
@mock.patch("%s.cups" % server_model)
def test_prepare_update_from_cups_unknown_trays(self, cups):
"""
Check that trays which are not in the PPD file are removed from Odoo
@@ -254,7 +250,7 @@ class TestPrintingPrinter(TransactionCase):
tray = self.new_tray()
vals = self.printer._prepare_update_from_cups(connection, cups_printer)
self.assertEqual(vals['tray_ids'], [(0, 0, {
'name': 'Auto (Default)',
'system_name': 'Auto',
}), (2, tray.id)])
self.assertEqual(
vals["tray_ids"],
[(0, 0, {"name": "Auto (Default)", "system_name": "Auto"}), (2, tray.id)],
)

View File

@@ -3,11 +3,10 @@
import mock
from odoo.tests.common import TransactionCase
from odoo.exceptions import UserError
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.addons.base_report_to_printer.models.printing_server"
class StopTest(Exception):
@@ -15,90 +14,75 @@ class StopTest(Exception):
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.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',
'printer-location': "location",
'device-uri': 'URI',
'printer-uri-supported': 'uri'
"printer-info": "Info",
"printer-make-and-model": "Make and Model",
"printer-location": "location",
"device-uri": "URI",
"printer-uri-supported": "uri",
}
def _record_vals(self, sys_name='sys_name'):
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'],
'uri': self.printer_vals['device-uri'],
"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"],
"uri": self.printer_vals["device-uri"],
}
@mock.patch('%s.cups' % model)
@mock.patch("%s.cups" % model)
def test_action_ok_inits_connection(self, cups):
""" It should initialize CUPS connection """
self.Model.action_ok()
cups.Connection.assert_called_once_with(
host=self.server.address, port=self.server.port,
host=self.server.address, port=self.server.port
)
@mock.patch('%s.cups' % model)
@mock.patch("%s.cups" % model)
def test_action_ok_gets_printers(self, cups):
""" It should get printers from CUPS """
cups.Connection().getPrinters.return_value = {
'sys_name': self.printer_vals,
}
cups.Connection().getPPD3.return_value = (200, 0, '')
cups.Connection().getPrinters.return_value = {"sys_name": self.printer_vals}
cups.Connection().getPPD3.return_value = (200, 0, "")
self.Model.action_ok()
cups.Connection().getPrinters.assert_called_once_with()
@mock.patch('%s.cups' % model)
@mock.patch("%s.cups" % model)
def test_action_ok_raises_warning_on_error(self, cups):
""" It should raise Warning on any error """
cups.Connection.side_effect = StopTest
with self.assertRaises(UserError):
self.Model.action_ok()
@mock.patch('%s.cups' % model)
@mock.patch("%s.cups" % model)
def test_action_ok_creates_new_printer(self, cups):
""" It should create new printer w/ proper vals """
cups.Connection().getPrinters.return_value = {
'sys_name': self.printer_vals,
}
cups.Connection().getPPD3.return_value = (200, 0, '')
cups.Connection().getPrinters.return_value = {"sys_name": self.printer_vals}
cups.Connection().getPPD3.return_value = (200, 0, "")
self.Model.action_ok()
rec_id = self.env['printing.printer'].search([
('system_name', '=', 'sys_name')
],
limit=1,
rec_id = self.env["printing.printer"].search(
[("system_name", "=", "sys_name")], limit=1
)
self.assertTrue(rec_id)
for key, val in self._record_vals().items():
if rec_id._fields[key].type == 'many2one':
if rec_id._fields[key].type == "many2one":
val = self.env[rec_id._fields[key].comodel_name].browse(val)
self.assertEqual(
val, rec_id[key],
)
self.assertEqual(val, rec_id[key])
@mock.patch('%s.cups' % model)
@mock.patch("%s.cups" % model)
def test_action_ok_skips_existing_printer(self, cups):
""" It should not recreate existing printers """
cups.Connection().getPrinters.return_value = {
'sys_name': self.printer_vals,
}
cups.Connection().getPPD3.return_value = (200, 0, '')
self.env['printing.printer'].create(
self._record_vals()
)
cups.Connection().getPrinters.return_value = {"sys_name": self.printer_vals}
cups.Connection().getPPD3.return_value = (200, 0, "")
self.env["printing.printer"].create(self._record_vals())
self.Model.action_ok()
res_ids = self.env['printing.printer'].search([
('system_name', '=', 'sys_name')
])
self.assertEqual(
1, len(res_ids),
res_ids = self.env["printing.printer"].search(
[("system_name", "=", "sys_name")]
)
self.assertEqual(1, len(res_ids))

View File

@@ -5,18 +5,17 @@ from odoo.tests.common import TransactionCase
class TestPrintingReportXmlAction(TransactionCase):
def setUp(self):
super(TestPrintingReportXmlAction, self).setUp()
self.Model = self.env['printing.report.xml.action']
self.Model = self.env["printing.report.xml.action"]
self.report = self.env['ir.actions.report'].search([], limit=1)
self.server = self.env['printing.server'].create({})
self.report = self.env["ir.actions.report"].search([], limit=1)
self.server = self.env["printing.server"].create({})
self.report_vals = {
'report_id': self.report.id,
'user_id': self.env.ref('base.user_demo').id,
'action': 'server',
"report_id": self.report.id,
"user_id": self.env.ref("base.user_demo").id,
"action": "server",
}
def new_record(self, vals=None):
@@ -27,65 +26,73 @@ class TestPrintingReportXmlAction(TransactionCase):
return self.Model.create(values)
def new_printer(self):
return self.env['printing.printer'].create({
'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',
})
return self.env["printing.printer"].create(
{
"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",
}
)
def test_behaviour(self):
""" It should return some action's data, unless called on empty
recordset
"""
xml_action = self.new_record()
self.assertEqual(xml_action.behaviour(), {
'action': xml_action.action,
'printer': xml_action.printer_id,
'tray': False,
})
self.assertEqual(
xml_action.behaviour(),
{
"action": xml_action.action,
"printer": xml_action.printer_id,
"tray": False,
},
)
xml_action = self.new_record({'printer_id': self.new_printer().id})
self.assertEqual(xml_action.behaviour(), {
'action': xml_action.action,
'printer': xml_action.printer_id,
'tray': False,
})
xml_action = self.new_record({"printer_id": self.new_printer().id})
self.assertEqual(
xml_action.behaviour(),
{
"action": xml_action.action,
"printer": xml_action.printer_id,
"tray": False,
},
)
self.assertEqual(self.Model.behaviour(), {})
def test_onchange_printer_tray_id_empty(self):
action = self.env['printing.report.xml.action'].new(
{'printer_tray_id': False})
action = self.env["printing.report.xml.action"].new({"printer_tray_id": False})
action.onchange_printer_id()
self.assertFalse(action.printer_tray_id)
def test_onchange_printer_tray_id_not_empty(self):
server = self.env['printing.server'].create({})
printer = self.env['printing.printer'].create({
'name': 'Printer',
'server_id': server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
})
tray = self.env['printing.tray'].create({
'name': 'Tray',
'system_name': 'TrayName',
'printer_id': printer.id,
})
server = self.env["printing.server"].create({})
printer = self.env["printing.printer"].create(
{
"name": "Printer",
"server_id": server.id,
"system_name": "Sys Name",
"default": True,
"status": "unknown",
"status_message": "Msg",
"model": "res.users",
"location": "Location",
"uri": "URI",
}
)
tray = self.env["printing.tray"].create(
{"name": "Tray", "system_name": "TrayName", "printer_id": printer.id}
)
action = self.env['printing.report.xml.action'].new(
{'printer_tray_id': tray.id})
action = self.env["printing.report.xml.action"].new(
{"printer_tray_id": tray.id}
)
self.assertEqual(action.printer_tray_id, tray)
action.onchange_printer_id()
self.assertFalse(action.printer_tray_id)

View File

@@ -6,207 +6,189 @@ import mock
from odoo import fields
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.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.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',
"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(),
"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)
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)
values["printer_id"] = printer.id
return self.env["printing.job"].create(values)
@mock.patch('%s.cups' % model)
@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,
)
self.assertEqual("server-error", rec_id.status)
@mock.patch('%s.cups' % model)
@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,
host=self.server.address, port=self.server.port
)
@mock.patch('%s.cups' % model)
@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)
@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:
with mock.patch.object(self.Model, "search") as search:
self.Model.update_printers()
search.assert_called_once_with([])
@mock.patch('%s.cups' % model)
@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)]
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)
@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,
)
self.assertEqual("unavailable", rec_id.status)
@mock.patch('%s.cups' % model)
@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',
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',
"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)
@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',
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',
"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)
@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,
host=self.server.address, port=self.server.port
)
@mock.patch('%s.cups' % model)
@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')
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,
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',
"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)
@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
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,
},
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.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')
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')
self.assertEqual(new_job.job_state, "processing")

View File

@@ -3,49 +3,47 @@
from odoo.tests.common import TransactionCase
model = 'odoo.addons.base_report_to_printer.models.printing_server'
model = "odoo.addons.base_report_to_printer.models.printing_server"
class TestPrintingTray(TransactionCase):
def setUp(self):
super(TestPrintingTray, self).setUp()
self.Model = self.env['printing.tray']
self.server = self.env['printing.server'].create({})
self.printer = self.env['printing.printer'].create({
'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.Model = self.env["printing.tray"]
self.server = self.env["printing.server"].create({})
self.printer = self.env["printing.printer"].create(
{
"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.tray_vals = {
'name': 'Tray',
'system_name': 'TrayName',
'printer_id': self.printer.id,
"name": "Tray",
"system_name": "TrayName",
"printer_id": self.printer.id,
}
def new_tray(self):
return self.env['printing.tray'].create(self.tray_vals)
return self.env["printing.tray"].create(self.tray_vals)
def test_report_behaviour(self):
""" It should add the selected tray in the report data """
ir_report = self.env['ir.actions.report'].search([], limit=1)
report = self.env['printing.report.xml.action'].create({
'user_id': self.env.user.id,
'report_id': ir_report.id,
'action': 'server',
})
ir_report = self.env["ir.actions.report"].search([], limit=1)
report = self.env["printing.report.xml.action"].create(
{"user_id": self.env.user.id, "report_id": ir_report.id, "action": "server"}
)
report.printer_tray_id = False
behaviour = report.behaviour()
self.assertEqual(behaviour['tray'], False)
self.assertEqual(behaviour["tray"], False)
# Check that we have te right value
report.printer_tray_id = self.new_tray()
behaviour = report.behaviour()
self.assertEqual(behaviour['tray'], report.printer_tray_id.system_name)
self.assertEqual(behaviour["tray"], report.printer_tray_id.system_name)

View File

@@ -3,8 +3,9 @@
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import mock
from odoo.tests import common
from odoo import exceptions
from odoo.tests import common
@common.at_install(False)
@@ -12,118 +13,124 @@ from odoo import exceptions
class TestReport(common.HttpCase):
def setUp(self):
super(TestReport, self).setUp()
self.Model = self.env['ir.actions.report']
self.server = self.env['printing.server'].create({})
self.Model = self.env["ir.actions.report"]
self.server = self.env["printing.server"].create({})
self.report_vals = {
'name': 'Test Report',
'model': 'ir.actions.report',
'report_name': 'Test Report',
"name": "Test Report",
"model": "ir.actions.report",
"report_name": "Test Report",
}
self.report_imd = self.env["ir.model.data"].create({
"name": "test",
"module": "base_report_to_printer",
"model": "ir.ui.view",
})
self.report_view = self.env["ir.ui.view"].create({
"name": "Test",
"type": "qweb",
"xml_id": "base_report_to_printer.test",
"model_data_id": self.report_imd.id,
"arch": """<t t-name="base_report_to_printer.test">
self.report_imd = self.env["ir.model.data"].create(
{"name": "test", "module": "base_report_to_printer", "model": "ir.ui.view"}
)
self.report_view = self.env["ir.ui.view"].create(
{
"name": "Test",
"type": "qweb",
"xml_id": "base_report_to_printer.test",
"model_data_id": self.report_imd.id,
"arch": """<t t-name="base_report_to_printer.test">
<div>Test</div>
</t>""",
})
}
)
self.report_imd.res_id = self.report_view.id
self.report = self.Model.create({
"name": "Test",
"report_type": "qweb-pdf",
"model": "res.partner",
"report_name": "base_report_to_printer.test",
})
self.report = self.Model.create(
{
"name": "Test",
"report_type": "qweb-pdf",
"model": "res.partner",
"report_name": "base_report_to_printer.test",
}
)
self.partners = self.env["res.partner"]
for n in range(5):
self.partners += self.env["res.partner"].create({
"name": "Test %d" % n,
})
self.partners += self.env["res.partner"].create({"name": "Test %d" % n})
def new_record(self):
return self.Model.create(self.report_vals)
def new_printer(self):
return self.env['printing.printer'].create({
'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',
})
return self.env["printing.printer"].create(
{
"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",
}
)
def test_can_print_report_context_skip(self):
""" It should return False based on context """
rec_id = self.new_record().with_context(
must_skip_send_to_printer=True
)
res = rec_id._can_print_report(
{'action': 'server'}, True, True
)
rec_id = self.new_record().with_context(must_skip_send_to_printer=True)
res = rec_id._can_print_report({"action": "server"}, True, True)
self.assertFalse(res)
def test_can_print_report_true(self):
""" It should return True when server print allowed """
res = self.new_record()._can_print_report(
{'action': 'server'}, True, True
)
res = self.new_record()._can_print_report({"action": "server"}, True, True)
self.assertTrue(res)
def test_can_print_report_false(self):
""" It should return False when server print not allowed """
res = self.new_record()._can_print_report(
{'action': 'server'}, True, False
)
res = self.new_record()._can_print_report({"action": "server"}, True, False)
self.assertFalse(res)
def test_render_qweb_pdf_not_printable(self):
""" It should print the report, only if it is printable
"""
with mock.patch('odoo.addons.base_report_to_printer.models.'
'printing_printer.PrintingPrinter.'
'print_document') as print_document:
with mock.patch(
"odoo.addons.base_report_to_printer.models."
"printing_printer.PrintingPrinter."
"print_document"
) as print_document:
self.report.render_qweb_pdf(self.partners.ids)
print_document.assert_not_called()
def test_render_qweb_pdf_printable(self):
""" It should print the report, only if it is printable
"""
with mock.patch('odoo.addons.base_report_to_printer.models.'
'printing_printer.PrintingPrinter.'
'print_document') as print_document:
self.report.property_printing_action_id.action_type = 'server'
with mock.patch(
"odoo.addons.base_report_to_printer.models."
"printing_printer.PrintingPrinter."
"print_document"
) as print_document:
self.report.property_printing_action_id.action_type = "server"
self.report.printing_printer_id = self.new_printer()
document = self.report.render_qweb_pdf(self.partners.ids)
print_document.assert_called_once_with(
self.report, document[0],
action='server', doc_format='qweb-pdf', tray=False)
self.report,
document[0],
action="server",
doc_format="qweb-pdf",
tray=False,
)
def test_print_document_not_printable(self):
""" It should print the report, regardless of the defined behaviour """
self.report.printing_printer_id = self.new_printer()
with mock.patch('odoo.addons.base_report_to_printer.models.'
'printing_printer.PrintingPrinter.'
'print_document') as print_document:
with mock.patch(
"odoo.addons.base_report_to_printer.models."
"printing_printer.PrintingPrinter."
"print_document"
) as print_document:
self.report.print_document(self.partners.ids)
print_document.assert_called_once()
def test_print_document_printable(self):
""" It should print the report, regardless of the defined behaviour """
self.report.property_printing_action_id.action_type = 'server'
self.report.property_printing_action_id.action_type = "server"
self.report.printing_printer_id = self.new_printer()
with mock.patch('odoo.addons.base_report_to_printer.models.'
'printing_printer.PrintingPrinter.'
'print_document') as print_document:
with mock.patch(
"odoo.addons.base_report_to_printer.models."
"printing_printer.PrintingPrinter."
"print_document"
) as print_document:
self.report.print_document(self.partners.ids)
print_document.assert_called_once()

View File

@@ -7,54 +7,49 @@ from odoo.tests import common
@common.at_install(False)
@common.post_install(True)
class TestResUsers(common.TransactionCase):
def setUp(self):
super(TestResUsers, self).setUp()
self.user_vals = {'name': 'Test',
'login': 'login',
}
self.user_vals = {"name": "Test", "login": "login"}
def new_record(self):
return self.env['res.users'].create(self.user_vals)
return self.env["res.users"].create(self.user_vals)
def test_available_action_types_excludes_user_default(self):
""" It should not contain `user_default` in avail actions """
self.user_vals['printing_action'] = 'user_default'
self.user_vals["printing_action"] = "user_default"
with self.assertRaises(ValueError):
self.new_record()
def test_available_action_types_includes_something_else(self):
""" It should still contain other valid keys """
self.user_vals['printing_action'] = 'server'
self.user_vals["printing_action"] = "server"
self.assertTrue(self.new_record())
def test_onchange_printer_tray_id_empty(self):
user = self.env['res.users'].new(
{'printer_tray_id': False})
user = self.env["res.users"].new({"printer_tray_id": False})
user.onchange_printing_printer_id()
self.assertFalse(user.printer_tray_id)
def test_onchange_printer_tray_id_not_empty(self):
server = self.env['printing.server'].create({})
printer = self.env['printing.printer'].create({
'name': 'Printer',
'server_id': server.id,
'system_name': 'Sys Name',
'default': True,
'status': 'unknown',
'status_message': 'Msg',
'model': 'res.users',
'location': 'Location',
'uri': 'URI',
})
tray = self.env['printing.tray'].create({
'name': 'Tray',
'system_name': 'TrayName',
'printer_id': printer.id,
})
server = self.env["printing.server"].create({})
printer = self.env["printing.printer"].create(
{
"name": "Printer",
"server_id": server.id,
"system_name": "Sys Name",
"default": True,
"status": "unknown",
"status_message": "Msg",
"model": "res.users",
"location": "Location",
"uri": "URI",
}
)
tray = self.env["printing.tray"].create(
{"name": "Tray", "system_name": "TrayName", "printer_id": printer.id}
)
user = self.env['res.users'].new(
{'printer_tray_id': tray.id})
user = self.env["res.users"].new({"printer_tray_id": tray.id})
self.assertEqual(user.printer_tray_id, tray)
user.onchange_printing_printer_id()
self.assertFalse(user.printer_tray_id)

View File

@@ -6,4 +6,3 @@
</xpath>
</template>
</odoo>

View File

@@ -1,2 +1 @@
from . import printing_printer_update_wizard

View File

@@ -8,22 +8,20 @@ import logging
from odoo import models
_logger = logging.getLogger(__name__)
class PrintingPrinterUpdateWizard(models.TransientModel):
_name = 'printing.printer.update.wizard'
_description = 'Printing Printer Update Wizard'
_name = "printing.printer.update.wizard"
_description = "Printing Printer Update Wizard"
def action_ok(self):
self.env['printing.server'].search([]) \
.update_printers(raise_on_error=True)
self.env["printing.server"].search([]).update_printers(raise_on_error=True)
return {
'name': 'Printers',
'view_mode': 'tree,form',
'res_model': 'printing.printer',
'type': 'ir.actions.act_window',
'target': 'current',
"name": "Printers",
"view_mode": "tree,form",
"res_model": "printing.printer",
"type": "ir.actions.act_window",
"target": "current",
}