Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 56 additions & 15 deletions newprinter.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
from gi.repository import Gdk
from gi.repository import Gtk
import functools
import urllib.parse

import cups

Expand Down Expand Up @@ -171,6 +172,58 @@ def download_gpg_fingerprint(url):

return None

def _normalize_uri(uri):
"""
Normalize a device URI for comparison by lowercasing the scheme
and injecting default network ports if missing (e.g. socket -> 9100).
"""
if not uri:
return ""

try:
parsed = urllib.parse.urlsplit(uri)
scheme = parsed.scheme.lower()
netloc = parsed.netloc
port = parsed.port
except ValueError:
return uri

if port is None:
default_ports = {
"http": 80,
"https": 443,
"ipp": 631,
"ipps": 631,
"socket": 9100,
"lpd": 515
}
if scheme in default_ports:
netloc = f"{netloc}:{default_ports[scheme]}"

return urllib.parse.urlunsplit((scheme, netloc, parsed.path, parsed.query, parsed.fragment))

def _filter_configured_devices(discovered_devices, configured_printers):
"""
Filter out already configured printers from the discovered devices list.
Matching is based on URI comparison using identical normalization rules.
"""
configured_uris = set()
if configured_printers:
for p in configured_printers.values():
uri = getattr(p, 'device_uri', "")
if uri:
configured_uris.add(_normalize_uri(uri))

filtered_devices = []
for d in discovered_devices:
if _normalize_uri(d.uri) not in configured_uris:
filtered_devices.append(d)
else:
debugprint("Suppressing already configured device: %s" % d.uri)

return filtered_devices


class NewPrinterGUI(GtkGUI):

__gsignals__ = {
Expand Down Expand Up @@ -2479,20 +2532,8 @@ def add_devices (self, devices, current_uri, no_more=False):
current_uri = None
current = None

for device in devices:
if device.type == "socket":
# Remove default port to more easily find duplicate URIs
device.uri = device.uri.replace (":9100", "")

# Map generic URIs to something canonical
def replace_generic (device):
if device.uri == "hp:/no_device_found":
device.uri = "hp"
elif device.uri == "hpfax:/no_device_found":
device.uri = "hpfax"
return device

devices = list(map (replace_generic, devices))
# Filter out already configured printers
devices = _filter_configured_devices(devices, getattr(self, 'printers', None))

# Mark duplicate URIs for deletion
for i in range (len (devices) - 1):
Expand All @@ -2501,7 +2542,7 @@ def replace_generic (device):
device2 = devices[j]
if device1.uri == "delete" or device2.uri == "delete":
continue
if device1.uri == device2.uri:
if _normalize_uri(device1.uri) == _normalize_uri(device2.uri):
# Keep the one with the longer (better) device ID
if (not device1.id):
device1.uri = "delete"
Expand Down
93 changes: 93 additions & 0 deletions test_newprinter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
#!/usr/bin/python3

## Copyright (C) 2015 Red Hat, Inc.
## Authors:
## Alexander Pevzner <pzz@apevzner.com>
## Ayush Singh <ayushsinghcee@gmail.com>

## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.

## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.

## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.

import pytest
import newprinter

class MockPrinter:
def __init__(self, uri):
self.device_uri = uri

class MockDevice:
def __init__(self, uri):
self.uri = uri

def test_configured_uri_exact_match():
dev = MockDevice("ipp://printer/ipp/print")
printers = {"Printer1": MockPrinter("ipp://printer/ipp/print")}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 0

def test_configured_uri_no_match():
dev = MockDevice("ipp://other-printer/ipp/print")
printers = {"Printer1": MockPrinter("ipp://printer/ipp/print")}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 1
assert filtered[0].uri == "ipp://other-printer/ipp/print"

def test_configured_uri_missing():
dev = MockDevice("ipp://printer/ipp/print")
printers = {
"Printer1": MockPrinter(None),
"Printer2": MockPrinter("")
}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 1

def test_multiple_configured_printers():
dev1 = MockDevice("usb://dev1")
dev2 = MockDevice("usb://dev2")
dev3 = MockDevice("usb://dev3") # not configured
printers = {
"P1": MockPrinter("usb://dev1"),
"P2": MockPrinter("usb://dev2")
}
filtered = newprinter._filter_configured_devices([dev1, dev2, dev3], printers)
assert len(filtered) == 1
assert filtered[0].uri == "usb://dev3"

def test_default_ports_normalization():
tests = [
("socket://192.168.1.5", "socket://192.168.1.5:9100"),
("http://192.168.1.5", "http://192.168.1.5:80"),
("https://192.168.1.5", "https://192.168.1.5:443"),
("ipp://192.168.1.5", "ipp://192.168.1.5:631"),
("ipps://192.168.1.5", "ipps://192.168.1.5:631"),
("lpd://192.168.1.5", "lpd://192.168.1.5:515"),
("IPP://192.168.1.5:631", "ipp://192.168.1.5"),
]
for disc_uri, conf_uri in tests:
dev = MockDevice(disc_uri)
printers = {"Printer1": MockPrinter(conf_uri)}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 0, f"Expected match between {disc_uri} and {conf_uri}"

def test_different_port_no_match():
dev = MockDevice("socket://192.168.1.5:1234")
printers = {"Printer1": MockPrinter("socket://192.168.1.5:9100")}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 1, "Different ports should not match"

def test_hp_no_device_found_does_not_match_real_printer():
dev = MockDevice("hp:/no_device_found")
printers = {"Printer1": MockPrinter("hp:/net/printer?ip=1.2.3.4")}
filtered = newprinter._filter_configured_devices([dev], printers)
assert len(filtered) == 1, "hp:/no_device_found should not match a real HP printer"