fix line number detection bug with empty lines

ansible_risk_insight/finder.py

@@ -262,21 +262,31 @@ def identify_lines_with_jsonpath(fpath: str = "", yaml_str: str = "", jsonpath:
                 blocks = find_child_yaml_block(current_lines, line_num_offset=current_line_num)
                 current_lines, line_num_tuple = blocks[p_num]
                 current_line_num = line_num_tuple[0]
-            except Exception:
-                print(p)
+            except Exception as e:
+                logger.debug(f"error occurred while detecting line number: {e}")
                 pass
     return current_lines, line_num_tuple
 
 
 def find_child_yaml_block(yaml_str: str, key: str = "", line_num_offset: int = -1) -> list:
-    skip_condition_funcs = [lambda x: x.strip().startswith("#"), lambda x: x.strip() == "---"]
+    skip_condition_funcs = [
+        # for YAML separator
+        lambda x: x.strip() == "---",
+        # for empty line
+        lambda x: x.strip() == "",
+        # for comment line
+        lambda x: x.strip().startswith("#"),
+    ]
 
     def match_condition_func(x):
         if key:
             return x.strip().startswith(f"{key}:")
         else:
             return x.strip().startswith("- ")
 
+    def is_yaml_end_separator(x):
+        return x.strip() == "..."
+
     def get_indent_level(x):
         return len(x) - len(x.lstrip())
 
@@ -331,46 +341,42 @@ def get_indent_level(x):
         for i, line in enumerate(yaml_str.splitlines()):
             line_num = i + 1
             current_indent = get_indent_level(line)
-            if current_indent == top_level_indent:
-                new_block = False
-                if match_condition_func(line):
-                    skip = False
-                    for skip_cond_func in skip_condition_funcs:
-                        if skip_cond_func(line):
-                            skip = True
-                            break
-                    if not skip:
-                        new_block = True
-                if new_block:
-                    if line_buffer:
-                        block_str = ""
-                        if isolated_line_buffer:
-                            block_str += "\n".join(isolated_line_buffer)
-                            buffer_begin = 1
-                        block_str += "\n".join(line_buffer)
-                        begin = buffer_begin
-                        end = line_num - 1
-                        if line_num_offset > 0:
-                            begin += line_num_offset - 1
-                            end += line_num_offset - 1
-                        line_num_tuple = (begin, end)
-                        blocks.append((block_str, line_num_tuple))
-                        line_buffer = []
-                        isolated_line_buffer = []
-                    buffer_begin = line_num
-                    line_buffer.append(line)
+            new_block = False
+            if current_indent == top_level_indent and match_condition_func(line):
+                skip = False
+                for skip_cond_func in skip_condition_funcs:
+                    if skip_cond_func(line):
+                        skip = True
+                        break
+                if not skip:
+                    new_block = True
+            if new_block:
+                if line_buffer:
+                    block_str = ""
+                    block_str += "\n".join(line_buffer)
+                    begin = buffer_begin
+                    end = line_num - 1
+                    if line_num_offset > 0:
+                        begin += line_num_offset - 1
+                        end += line_num_offset - 1
+                    line_num_tuple = (begin, end)
+                    blocks.append((block_str, line_num_tuple))
+                    line_buffer = []
+                    isolated_line_buffer = []
+                buffer_begin = line_num
+                line_buffer.append(line)
             else:
                 if buffer_begin < 0:
                     isolated_line_buffer.append(line)
                 else:
                     line_buffer.append(line)
         if line_buffer:
             block_str = ""
-            if isolated_line_buffer:
-                block_str += "\n".join(isolated_line_buffer)
             block_str += "\n".join(line_buffer)
             begin = buffer_begin
             end = line_num
+            if is_yaml_end_separator(line_buffer[-1]):
+                end = line_num - 1
             if line_num_offset > 0:
                 begin += line_num_offset - 1
                 end += line_num_offset - 1

test/test_scanner.py

@@ -59,14 +59,19 @@ def test_scanner_with_role(type, name):
     assert result.detail["executed_file"][0] == "/etc/install.sh"
 
 
-@pytest.mark.parametrize("type, name", [("playbook", "test/testdata/files/test_line_number.yml")])
-def test_scanner_line_number_detection(type, name):
+@pytest.mark.parametrize(
+    "type, name, expected_line_numbers",
+    [
+        ("playbook", "test/testdata/files/test_line_number.yml", [[6, 13], [14, 18], [20, 23], [29, 33]]),
+        ("playbook", "test/testdata/files/test_line_number2.yml", [[12, 15], [16, 17]]),
+    ],
+)
+def test_scanner_line_number_detection(type, name, expected_line_numbers):
     ari_result, _ = _scan(type=type, name=name, playbook_only=True)
     assert ari_result
     playbook_result = ari_result.playbook(path=name)
     assert playbook_result
     task_results = playbook_result.tasks()
-    expected_line_numbers = [[5, 12], [13, 17], [19, 22], [28, 32]]
     for i, task_result in enumerate(task_results.nodes):
         assert task_result.node.spec.line_num_in_file
         detected = task_result.node.spec.line_num_in_file

test/testdata/files/test_line_number.yml

@@ -1,5 +1,6 @@
 ---
 - name: Copy Apache configuration file over and restart httpd
+
   hosts: all
   tasks:
     - name: Copy Apache configuration file over

test/testdata/files/test_line_number2.yml

@@ -0,0 +1,17 @@
+---
+- hosts: localhost 
+  gather_facts: no
+
+  tags:
+    - test
+
+  collections:
+    - community.general
+
+  tasks:
+    - name: Task name here
+      import_role:
+        name: abc
+
+    - import_role:
+        name: def