SERVER-95738 Fix Headers (#28504)

GitOrigin-RevId: e9fdcd09c9d513c07cc69ddc316eef2f63a746e7

buildscripts/util/codeowners_utils.py

@@ -1,22 +1,35 @@
+import fnmatch
 import os
 import re
-from typing import Dict, List, Optional
+from functools import lru_cache
+from typing import Dict, List, Optional, Tuple
 
 import yaml
 
 
-def process_owners(owners_file_path: str) -> Dict[re.Pattern, List[str]]:
-    assert os.path.exists(owners_file_path)
+@lru_cache
+def process_owners(cur_dir: str) -> Tuple[Dict[re.Pattern, List[str]], bool]:
+    if not cur_dir:
+        return ({}, False)
+
+    owners_file_path = cur_dir + "/OWNERS.yml"
+    if not os.path.exists(owners_file_path):
+        return process_owners(os.path.dirname(cur_dir))
 
     with open(owners_file_path, "r") as f:
-        contents = yaml.safe_load(file)
+        contents = yaml.safe_load(f)
 
         assert "version" in contents, f"Version not found in {owners_file_path}"
         assert contents["version"] == "1.0.0", f"Invalid version in {owners_file_path}"
         assert "filters" in contents
 
+        no_parent_owners = False
+        if "options" in contents:
+            options = contents["options"]
+            no_parent_owners = "no_parent_owners" in options and options["no_parent_owners"]
+
         filters = {}
-        for file_filter in contents["filter"]:
+        for file_filter in contents["filters"]:
             assert "approvers" in file_filter
             approvers = file_filter["approvers"]
             del file_filter["approvers"]
@@ -24,35 +37,37 @@ def process_owners(owners_file_path: str) -> Dict[re.Pattern, List[str]]:
 
             assert len(file_filter) == 1
             pattern = next(iter(file_filter))
-            regex_pattern = re.compile(pattern)
+            pattern = f"{cur_dir}/{pattern}"
+
+            regex_pattern = re.compile(fnmatch.translate(pattern))
             filters[regex_pattern] = []
 
             for approver in approvers:
                 filters[regex_pattern].append(approver)
 
-    return filters
+    return (filters, no_parent_owners)
 
 
 class Owners:
     def __init__(self):
-        self.owners = {}
-        self.co_jira_map = yaml.safe_load("buildscripts/util/co_jira_map.yml")
+        self.co_jira_map = yaml.safe_load(open("buildscripts/util/co_jira_map.yml", "r"))
 
-    def get_filters_for_dir(self, d: str) -> Dict[re.Pattern, List[str]]:
-        if d in self.owners:
-            return self.owners[d]
-        owners_file_path = d + "/OWNERS.yml"
-        filters = process_owners(owners_file_path)
-        self.owners[d] = filters
-        return filters
-
-    def get_codeowners(self, file_name: str) -> List[str]:
-        filters = self.get_filters_for_dir(os.path.basename(file_name))
-        for regex_filter, codeowners in filters.items():
-            if regex_filter.match(file_name):
-                return codeowners
-
-        return []
+    def get_codeowners(self, file_path: str) -> List[str]:
+        cur_dir = os.path.dirname(file_path)
+        codeowners = []
+        # search up tree until matching filter found
+        while True:
+            filters, no_parent = process_owners(cur_dir)
+            if not cur_dir:
+                break
+            # latest applicable filter takes precedence
+            for regex_filter, cur_codeowners in filters.items():
+                if regex_filter.fullmatch(file_path):
+                    codeowners = cur_codeowners
+            if codeowners or no_parent:
+                break
+            cur_dir = os.path.dirname(cur_dir)
+        return codeowners
 
     def get_jira_team_from_codeowner(self, codeowner: str) -> List[str]:
         return self.co_jira_map[codeowner]
@@ -60,6 +75,6 @@ class Owners:
     def get_jira_team_owner(self, file_path: str) -> List[str]:
         return [
             jira_team
-            for jira_team in self.get_jira_team_from_codeowner(codeowner)
             for codeowner in self.get_codeowners(file_path)
+            for jira_team in self.get_jira_team_from_codeowner(codeowner)
         ]