Merge pull request #49 from bakhtos/maven-xml-parsing

Parse `pom.xml` using XML parsers

Author: SimonAdrian

requirements.txt

@@ -3,3 +3,4 @@ plantuml==0.3.0
 PyYAML==6.0
 ruamel.base==1.0.0
 PyDriller==2.6.0
+lxml==5.3.0

technology_specific_extractors/maven/mvn_entry.py

@@ -9,55 +9,53 @@
 import technology_specific_extractors.docker.dcr_entry as dcr
 import tmp.tmp as tmp
 import output_generators.traceability as traceability
-from core.service import CService
+
+try:
+    from lxml import etree
+    XML_BACKEND = "LXML"
+except ImportError:
+    import xml.etree.ElementTree as etree
+    XML_BACKEND = "PYTHON"
+NAMESPACE = {"mvn": "http://maven.apache.org/POM/4.0.0"}
 
 
 def set_microservices(dfd) -> dict:
     """Extracts the list of services from pom.xml files and sets the variable in the tmp-file.
     """
 
-    if not used_in_application():
-        return False
-
     if tmp.tmp_config.has_option("DFD", "microservices"):
         microservices = ast.literal_eval(tmp.tmp_config["DFD"]["microservices"])
     else:
         microservices = dict()
     microservices_set = set()
 
     pom_files = fi.get_file_as_lines("pom.xml")
-    module_tuples = list()
+    module_dict = dict()
 
     for pf in pom_files.keys():
         pom_file = pom_files[pf]
         image = "image_placeholder"
         modules = extract_modules(pom_file)
         if modules:
-            module_tuples.append((pom_file["name"], modules))
+            module_dict[(pom_file["name"])] = modules
         else:
             microservice, properties = parse_configurations(pom_file)
-            properties = extract_dependencies(properties, pom_file["content"])
+            properties = extract_dependencies(properties, pom_file)
             if microservice[0]:
                 port = dcr.detect_port(pom_file["path"])
                 # create microservice in dict
-                try:
-                    id = max(microservices.keys()) + 1
-                except:
-                    id = 0
-                microservices[id] = dict()
-
-                microservices[id]["name"] = microservice[0]
-                microservices[id]["image"] = image
-                microservices[id]["type"] = "internal"
-                microservices[id]["pom_path"] = pom_file["path"]
-                microservices[id]["properties"] = properties
-                microservices[id]["stereotype_instances"] = list()
+                id_ = max(microservices.keys(), default=-1) + 1
+                microservices[id_] = dict()
+                microservices[id_]["name"] = microservice[0]
+                microservices[id_]["image"] = image
+                microservices[id_]["type"] = "internal"
+                microservices[id_]["pom_path"] = pom_file["path"]
+                microservices[id_]["properties"] = properties
+                microservices[id_]["stereotype_instances"] = list()
                 if port:
-                    microservices[id]["tagged_values"] = [("Port", port)]
+                    microservices[id_]["tagged_values"] = [("Port", port)]
                 else:
-                    microservices[id]["tagged_values"] = list()
-
-                new_service = CService(microservice[0], )
+                    microservices[id_]["tagged_values"] = list()
                 try:
                     trace = dict()
                     name = microservice[0]
@@ -70,54 +68,58 @@ def set_microservices(dfd) -> dict:
                 except:
                     pass
 
-    nested_microservices = check_nested_modules(module_tuples)
-    for m in nested_microservices:
-        microservices_set.add(m)
+    nested_microservices = check_nested_modules(module_dict)
+    microservices_set.update(nested_microservices)
 
     tmp.tmp_config.set("DFD", "microservices", str(microservices).replace("%", "%%"))   # Need to escape single percentage signs for ConfigParser
 
     return microservices
 
 
-def extract_dependencies(properties: set, pom_file_lines) -> set:
+def extract_dependencies(properties: set, pom_file) -> set:
     """Parses pom_file to check for dependencies.
     """
 
-    for line in pom_file_lines:
-        if "spring-cloud-starter-netflix-hystrix" in line:
-            properties.add(("circuit_breaker", "Hystrix", ("file", "line", "span")))
+    file_name = pom_file["path"]
+    pom_path = os.path.join(tmp.tmp_config.get("Repository", "local_path"), file_name)
+    tree = etree.parse(pom_path)
+    root = tree.getroot()
 
-    return properties
-
-
-def used_in_application() -> bool:
-    """Checks if application has pom.xml file.
-    """
+    dependencies = root.find('mvn:dependencies', NAMESPACE)
+    if dependencies is not None:
+        for dependency in dependencies.findall('mvn:dependency', NAMESPACE):
+            artifactId = dependency.find('mvn:artifactId', NAMESPACE)
+            if artifactId is not None and artifactId.text.strip() == "spring-cloud-starter-netflix-hystrix":
+                properties.add(("circuit_breaker", "Hystrix", ("file", "line", "span")))
 
-    return fi.file_exists("pom.xml")
+    return properties
 
 
-def extract_modules(file: list) -> list:
+def extract_modules(pom_file: dict) -> list:
     """Extracts modules of a Maven project based on the <module> </module>-tag.
     """
 
-    modules = list()
-    for line in file["content"]:
-        if "<module>" in line:
-            modules.append(line.split("<module>")[1].split("</module>")[0].strip())
+    file_name = pom_file["path"]
+    pom_path = os.path.join(tmp.tmp_config.get("Repository", "local_path"), file_name)
+    tree = etree.parse(pom_path)
+    root = tree.getroot()
+
+    modules_list = set()
+    modules = root.find('mvn:modules', NAMESPACE)
+    if modules is not None:
+        modules_list = {module.text.strip() for module in modules.findall('mvn:module', NAMESPACE)}
 
-    return modules
+    return modules_list
 
 
-def check_nested_modules(module_tuples: list) -> list:
-    """Takes list of tuples of the form [(component, [modules])] and checks for links between them. If yes, returns list of components = services that need to be added to the list.
+def check_nested_modules(module_tuples: dict) -> set:
+    """Takes list of tuples of the form [(component, [modules])] and checks for links between them.
+    If yes, returns list of components = services that need to be added to the list.
     """
 
-    microservices = list()
-    for tuple1 in module_tuples:
-        for tuple2 in module_tuples:
-            if tuple1[0] in tuple2[1]:
-                microservices.append(tuple1[0])
+    modules = set(*module_tuples.values())
+    components = set(module_tuples.keys())
+    microservices = components & modules
 
     return microservices
 
@@ -126,17 +128,13 @@ def parse_configurations(pom_file) -> str:
     """Extracts servicename and properties for a given file. Tries properties file first, then pom file.
     """
 
-    properties = set()
     microservice, properties = parse_properties_file(pom_file["path"])
     if not microservice[0]:
-        microservice = extract_servicename_pom_file(pom_file["content"], pom_file["path"])
-
-        if microservice[0]:
-            microservice[0] = "pom_" + microservice[0]
-    if microservice[0]:
-        return microservice, properties
+        microservice = extract_servicename_pom_file(pom_file)
+    if not microservice[0]:
+        return (False, False), set()
 
-    return (False, False), properties
+    return microservice, properties
 
 
 def parse_properties_file(pom_path: str):
@@ -181,119 +179,48 @@ def parse_properties_file(pom_path: str):
     return microservice, properties
 
 
-def extract_servicename_pom_file(pom_file: list, file_name: str) -> str:
+def extract_servicename_pom_file(pom_file) -> str:
     """Extracts the name of a Maven-module based on the <finalName> tag if existing, else the <artifactIf>.
     """
 
     microservice = [False, False]
-    found_finalName = False
-    for line_nr in range(len(pom_file)):
-        line = pom_file[line_nr]
-        if "<finalName>" in line:
-            microservice[0] = line.split("<finalName>")[1].split("</finalName>")[0].strip()
-
-            # traceability
-            line_number = line_nr + 1
-            length_tuple = re.search(re.escape(microservice[0]), line).span()
-            span = "[" + str(length_tuple[0]) +  ":" + str(length_tuple[1]) + "]"
-            trace = (file_name, line_number, span)
-            microservice[1] = trace
-
-            found_finalName = True
-    if not found_finalName:
-        for line_nr in range(len(pom_file)):
-            line = pom_file[line_nr]
-            if "<artifactId>" in line:
-                if not in_dependency(pom_file, line_nr) and not in_parent(pom_file, line_nr) and not in_plugin(pom_file, line_nr):
-                    microservice[0] = line.split("<artifactId>")[1].split("</artifactId>")[0].strip()
-
-                    # traceability
-                    line_number = line_nr + 1
-                    length_tuple = re.search(microservice[0], line).span()
-                    span = "[" + str(length_tuple[0]) +  ":" + str(length_tuple[1]) + "]"
-                    trace = (file_name, line_number, span)
-                    microservice[1] = trace
-
-    return microservice
-
-
-def in_dependency(file: list, line_nr: str) -> bool:
-    """Checks if provided line is inside a <dependency> </dependency> block in the pom.xml .
-    """
-
-    count = line_nr
-    while count >= 0:
-        if "<dependency>" in file[count] and "</dependency>" in file[count]:
-            return False
-        if "<dependency>" in file[count]:
-            return True
-        if "</dependency>" in file[count]:
-            return False
-        count -= 1
-    return False
-
-
-def in_parent(file: list, line_nr: str) -> bool:
-    """Checks if provided line is inside a <plugin> </parent> block in the pom.xml .
-    """
-
-    count = line_nr
-    while count >= 0:
-        if "<parent>" in file[count] and "</parent>" in file[count]:
-            return False
-        if "<parent>" in file[count]:
-            return True
-        if "</parent>" in file[count]:
-            return False
-        count -= 1
-    return False
-
-
-def in_plugin(file: list, line_nr: str) -> bool:
-    """Checks if provided line is inside a <plugin> </plugin> block in the pom.xml .
-    """
-
-    count = line_nr
-    while count >= 0:
-        if "<plugin>" in file[count] and "</plugin>" in file[count]:
-            return False
-        if "<plugin>" in file[count]:
-            return True
-        if "</plugin>" in file[count]:
-            return False
-        count -= 1
-    return False
-
-
-def in_comment(file:list, line_nr: str) -> bool:
-    """Checks if provided line is inside a comment block in the pom.xml .
-    """
+    file_name = pom_file["path"]
+    pom_path = os.path.join(tmp.tmp_config.get("Repository", "local_path"), file_name)
+    tree = etree.parse(pom_path)
+    root = tree.getroot()
+
+    artifactId = root.find('mvn:build/mvn:finalName', NAMESPACE)
+    if artifactId is None:
+        artifactId = root.find('mvn:artifactId', NAMESPACE)
+    if artifactId is None:
+        return microservice
+
+    microservice[0] = artifactId.text.strip()
+
+    # tracing
+    if XML_BACKEND == "LXML":
+        line_nr = artifactId.sourceline - 1
+        line = pom_file["content"][line_nr]
+        length_tuple = re.search(microservice[0], line).span()
+        span = "[" + str(length_tuple[0]) + ":" + str(length_tuple[1]) + "]"
+    else:
+        line_nr = None
+        span = "[?:?]"
+    trace = (file_name, line_nr, span)
 
-    count = line_nr
-    while count >= 0:
-        if "<--" in file[count] and "-->" in file[count]:
-            return False
-        if "<--" in file[count]:
-            return True
-        if "-->" in file[count]:
-            return False
-        count -= 1
-    return False
+    microservice[0] = "pom_" + microservice[0]
+    microservice[1] = trace
 
+    return microservice
 
-count = 0
 
 def detect_microservice(file_path, dfd):
     """Detects which microservice a file belongs to by looking for next pom.xml.
     """
 
-    if not used_in_application():
-        return False
-
     microservice = [False, False]
     microservices = tech_sw.get_microservices(dfd)
 
-
     path = file_path
     found_pom = False
 
@@ -303,7 +230,7 @@ def detect_microservice(file_path, dfd):
     path = os.path.dirname(path)
     while not found_pom and path != "":
         dirs.append(os.scandir(os.path.join(local_repo_path, path)))
-        while dirs:
+        while dirs and not found_pom:
             dir = dirs.pop()
             for entry in dir:
                 if entry.is_file():