From fe4fefd880e23625f4743195dfbe8b2f1d387e3e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pawe=C5=82=20Redzy=C5=84ski?= <pawelredzynski@gmail.com>
Date: Wed, 5 Jan 2022 13:40:45 +0100
Subject: [PATCH] ls: handle broken symlink

Fixes: #7120
---
 dvc/repo/ls.py        | 14 +++++++++++---
 tests/func/test_ls.py | 21 +++++++++++++++++++++
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/dvc/repo/ls.py b/dvc/repo/ls.py
index d63acc2b2..35a729987 100644
--- a/dvc/repo/ls.py
+++ b/dvc/repo/ls.py
@@ -33,7 +33,7 @@ def ls(url, path=None, rev=None, recursive=None, dvc_only=False):
         if path:
             fs_path = os.path.abspath(repo.fs.path.join(fs_path, path))
 
-        ret = _ls(repo.repo_fs, fs_path, recursive, dvc_only)
+        ret = _ls(repo, fs_path, recursive, dvc_only)
 
         if path and not ret:
             raise PathMissingError(path, repo, dvc_only=dvc_only)
@@ -46,10 +46,13 @@ def ls(url, path=None, rev=None, recursive=None, dvc_only=False):
         return ret_list
 
 
-def _ls(fs, fs_path, recursive=None, dvc_only=False):
+def _ls(repo, fs_path, recursive=None, dvc_only=False):
+    from dvc.fs._metadata import Metadata
+
     def onerror(exc):
         raise exc
 
+    fs = repo.repo_fs
     infos = []
     try:
         for root, dirs, files in fs.walk(
@@ -66,7 +69,12 @@ def _ls(fs, fs_path, recursive=None, dvc_only=False):
 
     ret = {}
     for info in infos:
-        metadata = fs.metadata(info)
+        try:
+            metadata = fs.metadata(info)
+        except FileNotFoundError:
+            # broken symlink
+            metadata = Metadata(info, repo)
+
         if metadata.output_exists or not dvc_only:
             path = (
                 fs.path.name(fs_path)
diff --git a/tests/func/test_ls.py b/tests/func/test_ls.py
index 90707dad9..68e735ef5 100644
--- a/tests/func/test_ls.py
+++ b/tests/func/test_ls.py
@@ -557,3 +557,24 @@ def test_subrepo(dvc_top_level, erepo):
     assert _list_files(subrepo, ".") == common_outputs
     assert _list_files(subrepo, "scm_dir") == {"ipsum"}
     assert _list_files(subrepo, "dvc_dir") == {"lorem"}
+
+
+def test_broken_symlink(tmp_dir, dvc):
+    from dvc.system import System
+
+    tmp_dir.gen("file", "content")
+    System.symlink("file", "link")
+
+    os.remove("file")
+
+    entries = Repo.ls(os.fspath(tmp_dir))
+
+    assert entries == [
+        {
+            "isout": False,
+            "isdir": False,
+            "isexec": False,
+            "path": ".dvcignore",
+        },
+        {"isout": False, "isdir": False, "isexec": False, "path": "link"},
+    ]