Fix up listdir and read for pipes

smbclient/_io.py

@@ -436,7 +436,13 @@ def readinto(self, b):
         if self._offset >= self.fd.end_of_file and self.FILE_TYPE != 'pipe':
             return 0
 
-        file_bytes = self.fd.read(self._offset, len(b))
+        try:
+            file_bytes = self.fd.read(self._offset, len(b))
+        except SMBResponseException as exc:
+            if exc.status == NtStatus.STATUS_PIPE_BROKEN:
+                file_bytes = b""
+            else:
+                raise
 
         b[:len(file_bytes)] = file_bytes
 

smbclient/_os.py

@@ -143,10 +143,15 @@ def listdir(path, search_pattern="*", **kwargs):
     :param kwargs: Common SMB Session arguments for smbclient.
     :return: A list containing the names of the entries in the directory.
     """
-    with SMBDirectoryIO(path, mode='r', **kwargs) as dir_fd:
-        return list(e['file_name'].get_value().decode('utf-16-le') for e in
-                    dir_fd.query_directory(search_pattern, FileInformationClass.FILE_NAMES_INFORMATION) if
-                    e['file_name'].get_value().decode('utf-16-le') not in ['.', '..'])
+    with SMBDirectoryIO(path, mode='r', share_access='r', **kwargs) as dir_fd:
+        try:
+            raw_filenames = dir_fd.query_directory(search_pattern, FileInformationClass.FILE_NAMES_INFORMATION)
+            return list(e['file_name'].get_value().decode('utf-16-le') for e in raw_filenames if
+                        e['file_name'].get_value().decode('utf-16-le') not in ['.', '..'])
+        except SMBResponseException as exc:
+            if exc.status == NtStatus.STATUS_NO_SUCH_FILE:
+                return []
+            raise
 
 
 def lstat(path, **kwargs):

tests/test_smbclient_os.py

@@ -221,6 +221,11 @@ def test_listdir_with_pattern(smb_share):
     assert smbclient.listdir(smb_share, search_pattern="file-test?.txt") == ["file-test1.txt"]
 
 
+def test_listdir_with_pattern_no_match(smb_share):
+    actual = smbclient.listdir(smb_share, search_pattern="no matching file")
+    assert actual == []
+
+
 def test_lstat_on_file(smb_share):
     filename = ntpath.join(smb_share, 'file.txt')
     with smbclient.open_file(filename, mode='w') as fd: