[Beats][pytest] Asserting if filebeat logs include errors

filebeat/tests/system/test_modules.py

@@ -131,22 +131,33 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
             cmd.append("{module}.{fileset}.var.format=json".format(module=module, fileset=fileset))
 
         output_path = os.path.join(self.working_dir)
-        output = open(os.path.join(output_path, "output.log"), "ab")
-        output.write(bytes(" ".join(cmd) + "\n", "utf-8"))
-
-        # Use a fixed timezone so results don't vary depending on the environment
-        # Don't use UTC to avoid hiding that non-UTC timezones are not being converted as needed,
-        # this can happen because UTC uses to be the default timezone in date parsers when no other
-        # timezone is specified.
-        local_env = os.environ.copy()
-        local_env["TZ"] = 'Etc/GMT+2'
-
-        subprocess.Popen(cmd,
-                         env=local_env,
-                         stdin=None,
-                         stdout=output,
-                         stderr=subprocess.STDOUT,
-                         bufsize=0).wait()
+        # Runs inside a with loop to ensure file is closed afterwards
+        with open(os.path.join(output_path, "output.log"), "ab") as output:
+            output.write(bytes(" ".join(cmd) + "\n", "utf-8"))
+
+            # Use a fixed timezone so results don't vary depending on the environment
+            # Don't use UTC to avoid hiding that non-UTC timezones are not being converted as needed,
+            # this can happen because UTC uses to be the default timezone in date parsers when no other
+            # timezone is specified.
+            local_env = os.environ.copy()
+            local_env["TZ"] = 'Etc/GMT+2'
+
+            subprocess.Popen(cmd,
+                             env=local_env,
+                             stdin=None,
+                             stdout=output,
+                             stderr=subprocess.STDOUT,
+                             bufsize=0).wait()
+            output.close()
+
+        # List of errors to check in filebeat output logs
+        errors = ["Error loading pipeline for fileset"]
+        # Checks if the output of filebeat includes errors
+        contains_error, error_line = file_contains(os.path.join(output_path, "output.log"), errors)
+        assert contains_error is False, "Error found in log:{}".format(error_line)
+
+        # Ensure file is actually closed to ensure large amount of file handlers is not spawning
+        assert output.closed is True
 
         # Make sure index exists
         self.wait_until(lambda: self.es.indices.exists(self.index_name))
@@ -296,5 +307,14 @@ def delete_key(obj, key):
         del obj[key]
 
 
+def file_contains(filepath, strings):
+    with open(filepath, 'r') as file:
+        for line in file:
+            for string in strings:
+                if string in line:
+                    return True, line
+    return False, None
+
+
 def pretty_json(obj):
     return json.dumps(obj, indent=2, separators=(',', ': '))