fix(integrations): Add support for CODEOWNERS sections

src/sentry/ownership/grammar.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import re
-from collections import namedtuple
+from collections import OrderedDict, namedtuple
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from typing import Any, NamedTuple
 
@@ -13,6 +13,7 @@
 from sentry.eventstore.models import EventSubjectTemplateData
 from sentry.integrations.models.repository_project_path_config import RepositoryProjectPathConfig
 from sentry.models.organizationmember import OrganizationMember
+from sentry.ownership.section_line import SectionLine
 from sentry.types.actor import Actor, ActorType
 from sentry.users.services.user.service import user_service
 from sentry.utils.codeowners import codeowners_match
@@ -408,29 +409,49 @@ def convert_codeowners_syntax(
     """
 
     result = ""
+    section_owners = []
+    section_lines = OrderedDict()
 
     for rule in codeowners.splitlines():
-        if rule.startswith("#") or not len(rule):
-            # We want to preserve comments from CODEOWNERS
-            result += f"{rule}\n"
+        # End of current section
+        if not len(rule):
+            section_result = get_codeowners_section_result(
+                section_lines, associations, code_mapping
+            )
+            result += f"{section_result}\n"
+            section_lines.clear()
             continue
 
-        # Skip lines that are only empty space characters
-        if re.match(r"^\s*$", rule):
+        path, code_owners = get_codeowners_path_and_owners(rule)
+
+        # Start of new section
+        if re.search(r"(^\[([^]^\s]*)\])", path):
+            section_lines.clear()
+            section_owners = code_owners
             continue
 
-        path, code_owners = get_codeowners_path_and_owners(rule)
-        # Escape invalid paths https://docs.github.com/en/github/creating-cloning-and-archiving-repositories/creating-a-repository-on-github/about-code-owners#syntax-exceptions
-        # Check if path has whitespace
-        # Check if path has '#' not as first character
-        # Check if path contains '!'
-        # Check if path has a '[' followed by a ']'
-        if re.search(r"(\[([^]^\s]*)\])|[\s!#]", path):
+        section_line = SectionLine(rule, path, list(code_owners), section_owners)
+        if section_line.should_skip():
             continue
 
-        sentry_assignees = []
+        section_lines[section_line.get_dict_key()] = section_line
 
-        for owner in code_owners:
+    return result + get_codeowners_section_result(section_lines, associations, code_mapping)
+
+
+def get_codeowners_section_result(
+    section_lines: OrderedDict[str, SectionLine],
+    associations: Mapping[str, Any],
+    code_mapping: RepositoryProjectPathConfig,
+) -> str:
+    result = ""
+    for section_line in section_lines.values():
+        if section_line.is_preserved_comment:
+            result += f"{section_line.original_line}\n"
+            continue
+
+        sentry_assignees = []
+        for owner in section_line.get_owners():
             try:
                 sentry_assignees.append(associations[owner])
             except KeyError:
@@ -449,15 +470,15 @@ def convert_codeowners_syntax(
             # foo/dir -> anchored
             # foo/dir/ -> anchored
             # foo/ -> not anchored
-            if re.search(r"[\/].{1}", path):
-                path_with_stack_root = path.replace(
+            if re.search(r"[\/].{1}", section_line.path):
+                path_with_stack_root = section_line.path.replace(
                     code_mapping.source_root, code_mapping.stack_root, 1
                 )
                 # flatten multiple '/' if not protocol
                 formatted_path = re.sub(r"(?<!:)\/{2,}", "/", path_with_stack_root)
                 result += f'codeowners:{formatted_path} {" ".join(sentry_assignees)}\n'
             else:
-                result += f'codeowners:{path} {" ".join(sentry_assignees)}\n'
+                result += f'codeowners:{section_line.path} {" ".join(sentry_assignees)}\n'
 
     return result
 

src/sentry/ownership/section_line.py

@@ -0,0 +1,42 @@
+import re
+
+
+class SectionLine:
+    original_line: str
+    path: str = ""
+    is_preserved_comment: bool
+    _path_owners: list[str] = []
+    _section_owners: list[str] = []
+    _has_valid_path: bool
+
+    def __init__(
+        self,
+        original_line: str,
+        path: str,
+        path_owners: list[str],
+        section_owners: list[str],
+    ):
+        self.original_line = original_line
+        self.path = path
+        self.is_preserved_comment = original_line.startswith("#") or not len(original_line)
+        self._path_owners = path_owners
+        self._section_owners = section_owners
+        self._has_valid_path = re.search(r"(\[([^]^\s]*)\])|[\s!#]", path) is None
+
+    def get_dict_key(self) -> str:
+        return self.path if self._has_valid_path else self.original_line
+
+    def get_owners(self) -> list[str]:
+        return self._path_owners if len(self._path_owners) > 0 else self._section_owners
+
+    def should_skip(self) -> bool:
+        if self.is_preserved_comment:
+            return False
+
+        if re.match(r"^\s*$", self.original_line):
+            return True
+
+        if not self._has_valid_path:
+            return True
+
+        return False

tests/sentry/ownership/test_grammar.py

@@ -43,6 +43,15 @@
 tests/file\ with\ spaces/ @NisanthanNanthakumar
 """
 
+associations = {
+    "@getsentry/frontend": "front-sentry",
+    "@getsentry/docs": "docs-sentry",
+    "@getsentry/ecosystem": "ecosystem",
+    "@NisanthanNanthakumar": "nisanthan.nanthakumar@sentry.io",
+    "@AnotherUser": "anotheruser@sentry.io",
+    "nisanthan.nanthakumar@sentry.io": "nisanthan.nanthakumar@sentry.io",
+}
+
 
 def test_parse_rules():
     assert parse_rules(fixture_data) == [
@@ -920,20 +929,39 @@ def test_convert_codeowners_syntax():
     assert (
         convert_codeowners_syntax(
             codeowners_fixture_data,
-            {
-                "@getsentry/frontend": "front-sentry",
-                "@getsentry/docs": "docs-sentry",
-                "@getsentry/ecosystem": "ecosystem",
-                "@NisanthanNanthakumar": "nisanthan.nanthakumar@sentry.io",
-                "@AnotherUser": "anotheruser@sentry.io",
-                "nisanthan.nanthakumar@sentry.io": "nisanthan.nanthakumar@sentry.io",
-            },
+            associations,
             code_mapping,
         )
         == "\n# cool stuff comment\ncodeowners:*.js front-sentry nisanthan.nanthakumar@sentry.io\n# good comment\n\n\ncodeowners:webpack://docs/* docs-sentry ecosystem\ncodeowners:src/sentry/* anotheruser@sentry.io\ncodeowners:api/* nisanthan.nanthakumar@sentry.io\n"
     )
 
 
+def test_convert_codeowners_multiple_sections_with_overrides():
+    code_mapping = type("", (), {})()
+    code_mapping.stack_root = ""
+    code_mapping.source_root = ""
+
+    result = convert_codeowners_syntax(
+        r"""
+/fileA.txt @getsentry/frontend
+
+[Docs] @getsentry/docs
+/fileC.txt
+
+[Some_Section]
+/fileD.txt @getsentry/docs
+/fileD.txt @getsentry/frontend
+""",
+        associations,
+        code_mapping,
+    )
+
+    assert (
+        result
+        == "\ncodeowners:/fileA.txt front-sentry\n\ncodeowners:/fileC.txt docs-sentry\n\ncodeowners:/fileD.txt front-sentry\n"
+    )
+
+
 def test_convert_codeowners_syntax_excludes_invalid():
     code_mapping = type("", (), {})()
     code_mapping.stack_root = "webpack://static/"

tests/sentry/ownership/test_section_line.py

@@ -0,0 +1,56 @@
+from sentry.ownership.section_line import SectionLine
+
+
+def test_get_owners_only_path_owners_returns_path_owners():
+    line = SectionLine("", "", ["a", "b"], [])
+    assert line.get_owners() == ["a", "b"]
+
+
+def test_get_owners_only_section_owners_returns_section_owners():
+    line = SectionLine("", "", [], ["a", "b"])
+    assert line.get_owners() == ["a", "b"]
+
+
+def test_get_owners_both_owners_returns_path_owners():
+    line = SectionLine("", "", ["a", "b"], ["c", "d"])
+    assert line.get_owners() == ["a", "b"]
+
+
+def test_is_preserved_comment_empty_line_returns_true():
+    line = SectionLine("", "", [], [])
+    assert line.is_preserved_comment is True
+
+
+def test_is_preserved_valid_comment_returns_true():
+    line = SectionLine("# some comment", "", [], [])
+    assert line.is_preserved_comment is True
+
+
+def test_should_skip_valid_comment_returns_false():
+    line = SectionLine("# some comment", "", [], [])
+    assert line.should_skip() is False
+
+
+def test_should_skip_line_with_spaces_returns_true():
+    line = SectionLine(" ", " ", [], [])
+    assert line.should_skip() is True
+
+
+def test_should_skip_line_valid_path_returns_false():
+    line = SectionLine("/fileA.txt", "/fileA.txt", [], [])
+    assert line.should_skip() is False
+
+
+def test_should_skip_line_invalid_path_returns_true():
+    line = SectionLine(" cde", " cde", [], [])
+    assert line.should_skip() is True
+
+
+def test_get_dict_key_invalid_path_returns_original_line():
+    line = SectionLine("[Section] owner", "[Section]", [], [])
+    assert line.get_dict_key() == "[Section] owner"
+
+
+def test_get_dict_key_returns_path_only():
+    line = SectionLine("/fileA.txt githubuser@sentry.io", "/fileA.txt", [], [])
+    assert line.get_dict_key() == "/fileA.txt"