Merge pull request #1946 from phw/refactor_cdrom

Refactor picard.util.cdrom

picard/util/cdrom.py

@@ -5,7 +5,7 @@
 # Copyright (C) 2004 Robert Kaye
 # Copyright (C) 2007 Lukáš Lalinský
 # Copyright (C) 2008 Will
-# Copyright (C) 2008, 2018-2020 Philipp Wolfer
+# Copyright (C) 2008, 2018-2021 Philipp Wolfer
 # Copyright (C) 2009 david
 # Copyright (C) 2013 Johannes Dewender
 # Copyright (C) 2013 Sebastian Ramacher
@@ -28,22 +28,14 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 
+import os.path
 
-from PyQt5.QtCore import (
-    QFile,
-    QIODevice,
-)
-
+from picard import log
 from picard.config import get_config
 from picard.const.sys import (
     IS_LINUX,
     IS_WIN,
 )
-from picard.util import uniqify
-
-
-if IS_WIN:
-    from ctypes import windll
 
 
 try:
@@ -63,67 +55,76 @@ if discid is not None:
 
 LINUX_CDROM_INFO = '/proc/sys/dev/cdrom/info'
 
-# if get_cdrom_drives() lists ALL drives available on the machine
+
+def _generic_iter_drives():
+    config = get_config()
+    yield from (
+        device.strip() for device
+        in config.setting["cd_lookup_device"].split(",")
+        if device and not device.isspace()
+    )
+
+
+def _parse_linux_cdrom_info(f):
+    drive_names = []
+    drive_audio_caps = []
+    while True:
+        line = f.readline()
+        if not line:
+            break
+        if ":" in line:
+            key, values = line.split(':')
+            if key == 'drive name':
+                drive_names = values.split()
+            elif key == 'Can play audio':
+                drive_audio_caps = [v == '1' for v in values.split()]
+                break  # no need to continue past this line
+    yield from zip(drive_names, drive_audio_caps)
+
+
 if IS_WIN:
+    from ctypes import windll
+
     AUTO_DETECT_DRIVES = True
-elif IS_LINUX and QFile.exists(LINUX_CDROM_INFO):
+    DRIVE_TYPE_CDROM = 5
+
+    def _iter_drives():
+        GetLogicalDrives = windll.kernel32.GetLogicalDrives
+        GetDriveType = windll.kernel32.GetDriveTypeW
+        mask = GetLogicalDrives()
+        for i in range(26):
+            if mask >> i & 1:
+                drive = chr(i + ord("A")) + ":"
+                if GetDriveType(drive) == DRIVE_TYPE_CDROM:
+                    yield drive
+
+elif IS_LINUX and os.path.isfile(LINUX_CDROM_INFO):
     AUTO_DETECT_DRIVES = True
+
+    def _iter_drives():
+        # Read info from /proc/sys/dev/cdrom/info
+        with open(LINUX_CDROM_INFO, 'r') as f:
+            # Show only drives that are capable of playing audio
+            yield from (
+                os.path.realpath('/dev/%s' % drive)
+                for drive, can_play_audio in _parse_linux_cdrom_info(f)
+                if can_play_audio
+            )
+
 else:
     # There might be more drives we couldn't detect
     # setting uses a text field instead of a drop-down
     AUTO_DETECT_DRIVES = False
+    _iter_drives = _generic_iter_drives
 
 
 def get_cdrom_drives():
     """List available disc drives on the machine
     """
     # add default drive from libdiscid to the list
-    drives = list(DEFAULT_DRIVES)
-
-    if IS_WIN:
-        GetLogicalDrives = windll.kernel32.GetLogicalDrives
-        GetDriveType = windll.kernel32.GetDriveTypeW
-        DRIVE_CDROM = 5
-        mask = GetLogicalDrives()
-        for i in range(26):
-            if mask >> i & 1:
-                drive = chr(i + ord("A")) + ":"
-                if GetDriveType(drive) == DRIVE_CDROM:
-                    drives.append(drive)
-
-    elif IS_LINUX and AUTO_DETECT_DRIVES:
-        # Read info from /proc/sys/dev/cdrom/info
-        cdinfo = QFile(LINUX_CDROM_INFO)
-        if cdinfo.open(QIODevice.ReadOnly | QIODevice.Text):
-            drive_names = []
-            drive_audio_caps = []
-            while True:
-                line = bytes(cdinfo.readLine()).decode()
-                if not line:
-                    break
-                if ":" in line:
-                    key, values = line.split(':')
-                    if key == 'drive name':
-                        drive_names = values.split()
-                    elif key == 'Can play audio':
-                        drive_audio_caps = [v == '1' for v in values.split()]
-                        break  # no need to continue past this line
-            # Show only drives that are capable of playing audio
-            for index, drive in enumerate(drive_names):
-                if drive_audio_caps[index]:
-                    device = '/dev/%s' % drive
-                    symlink_target = QFile.symLinkTarget(device)
-                    if symlink_target != '':
-                        device = symlink_target
-                    drives.append(device)
-
-    else:
-        config = get_config()
-        for device in config.setting["cd_lookup_device"].split(","):
-            # Need to filter out empty strings,
-            # particularly if the device list is empty
-            if device.strip() != '':
-                drives.append(device.strip())
-
-    # make sure no drive is listed twice (given by multiple sources)
-    return sorted(uniqify(drives))
+    drives = set(DEFAULT_DRIVES)
+    try:
+        drives |= set(_iter_drives())
+    except OSError as error:
+        log.error(error)
+    return sorted(drives)

test/test_util_cdrom.py

@@ -0,0 +1,130 @@
+# -*- coding: utf-8 -*-
+#
+# Picard, the next-generation MusicBrainz tagger
+#
+# Copyright (C) 2021 Philipp Wolfer
+#
+# 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 io
+from typing import Iterable
+import unittest
+
+from test.picardtestcase import PicardTestCase
+
+from picard.const.sys import IS_WIN
+from picard.util import cdrom
+
+
+MOCK_CDROM_INFO = """CD-ROM information, Id: cdrom.c 3.20 2003/12/17
+
+drive name:		sr1	sr0
+drive speed:		24	24
+drive # of slots:	1	1
+Can close tray:		0	1
+Can open tray:		1	1
+Can lock tray:		1	1
+Can change speed:	1	1
+Can select disk:	0	0
+Can read multisession:	1	1
+Can read MCN:		1	1
+Reports media changed:	1	1
+Can play audio:		1	0
+Can write CD-R:		1	1
+Can write CD-RW:	1	1
+Can read DVD:		1	1
+Can write DVD-R:	1	1
+Can write DVD-RAM:	1	1
+Can read MRW:		1	1
+Can write MRW:		1	1
+Can write RAM:		1	1
+"""
+
+MOCK_CDROM_INFO_EMPTY = """CD-ROM information, Id: cdrom.c 3.20 2003/12/17
+
+drive name:
+drive speed:
+drive # of slots:
+Can close tray:
+Can open tray:
+Can lock tray:
+Can change speed:
+Can select disk:
+Can read multisession:
+Can read MCN:
+Reports media changed:
+Can play audio:
+Can write CD-R:
+Can write CD-RW:
+Can read DVD:
+Can write DVD-R:
+Can write DVD-RAM:
+Can read MRW:
+Can write MRW:
+Can write RAM:
+"""
+
+
+class LinuxParseCdromInfoTest(PicardTestCase):
+
+    def test_drives(self):
+        with io.StringIO(MOCK_CDROM_INFO) as f:
+            drives = list(cdrom._parse_linux_cdrom_info(f))
+            self.assertEqual([('sr1', True), ('sr0', False)], drives)
+
+    def test_empty(self):
+        with io.StringIO(MOCK_CDROM_INFO_EMPTY) as f:
+            drives = list(cdrom._parse_linux_cdrom_info(f))
+            self.assertEqual([], drives)
+
+    def test_empty_string(self):
+        with io.StringIO("") as f:
+            drives = list(cdrom._parse_linux_cdrom_info(f))
+            self.assertEqual([], drives)
+
+
+class GetCdromDrivesTest(PicardTestCase):
+
+    def test_get_cdrom_drives(self):
+        self.set_config_values({"cd_lookup_device": "/dev/cdrom"})
+        # Independent of the implementation get_cdrom_drives must not rais
+        # and return an Iterable.
+        drives = cdrom.get_cdrom_drives()
+        self.assertIsInstance(drives, Iterable)
+        self.assertTrue(set(cdrom.DEFAULT_DRIVES).issubset(drives))
+
+    def test_generic_iter_drives(self):
+        self.set_config_values({"cd_lookup_device": "/dev/cdrom"})
+        self.assertEqual(["/dev/cdrom"], list(cdrom._generic_iter_drives()))
+        self.set_config_values({"cd_lookup_device": "/dev/cdrom, /dev/sr0"})
+        self.assertEqual(["/dev/cdrom", "/dev/sr0"], list(cdrom._generic_iter_drives()))
+        self.set_config_values({"cd_lookup_device": ""})
+        self.assertEqual([], list(cdrom._generic_iter_drives()))
+        self.set_config_values({"cd_lookup_device": " ,, ,\t, "})
+        self.assertEqual([], list(cdrom._generic_iter_drives()))
+
+
+@unittest.skipUnless(IS_WIN, "windows test")
+class WindowsGetCdromDrivesTest(PicardTestCase):
+
+    def test_autodetect(self):
+        self.assertTrue(cdrom.AUTO_DETECT_DRIVES)
+
+    def test_iter_drives(self):
+        drives = cdrom._iter_drives()
+        self.assertIsInstance(drives, Iterable)
+        # This should not raise
+        list(drives)