Enable CM integration tests back (elastic#9625)

A few things broke the integration tests, change in response and also
now Kibana also blacklist some elements.

I have removed the blacklisted integration tests, theses are tested
using unit tests, because kibana will not return anything that is
blacklisted.

I've moved the integration tests away from using the file output because
it its blacklisted, instead we now use the Elasticsearch output to do
the assertions.

(cherry picked from commit e46c773)

x-pack/libbeat/management/config_test.go

@@ -0,0 +1,18 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package management
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func EnsureBlacklistItems(t *testing.T) {
+	// NOTE: We do not permit to configure the console or the file output with CM for security reason.
+	c := defaultConfig()
+	v, _ := c.Blacklist.Patterns["output"]
+	assert.Equal(t, "console|file", v)
+}

x-pack/libbeat/tests/system/test_management.py

@@ -2,21 +2,38 @@
 import os
 import json
 import requests
+import string
+import random
 import unittest
+from elasticsearch import Elasticsearch
+
 
 from base import BaseTest
 
 
 INTEGRATION_TESTS = os.environ.get('INTEGRATION_TESTS', False)
 KIBANA_PASSWORD = 'changeme'
+TIMEOUT = 5 * 60
 
 
 class TestManagement(BaseTest):
-    @unittest.skip('Skipped because of issue: https://github.com/elastic/beats/issues/9597')
+
+    def setUp(self):
+        super(TestManagement, self).setUp()
+        # NOTES: Theses options are linked to the specific of the docker compose environment for
+        # CM.
+        self.es_host = os.getenv('ES_HOST', 'localhost') + ":" + os.getenv('ES_POST', '9200')
+        self.es_user = "elastic"
+        self.es_pass = "changeme"
+        self.es = Elasticsearch([self.get_elasticsearch_url()], verify_certs=True)
+
+    @unittest.skipIf(not INTEGRATION_TESTS,
+                     "integration tests are disabled, run with INTEGRATION_TESTS=1 to enable them.")
     def test_enroll(self):
         """
         Enroll the beat in Kibana Central Management
         """
+
         # We don't care about this as it will be replaced by enrollment
         # process:
         config_path = os.path.join(self.working_dir, "mockbeat.yml")
@@ -27,6 +44,7 @@ def test_enroll(self):
         exit_code = self.enroll(KIBANA_PASSWORD)
 
         assert exit_code == 0
+
         assert self.log_contains("Enrolled and ready to retrieve settings")
 
         # Enroll creates a keystore (to store access token)
@@ -43,7 +61,8 @@ def test_enroll(self):
         backup_content = open(config_path + ".bak", 'r').read()
         assert config_content == backup_content
 
-    @unittest.skip('Skipped because of issue: https://github.com/elastic/beats/issues/9597')
+    @unittest.skipIf(not INTEGRATION_TESTS,
+                     "integration tests are disabled, run with INTEGRATION_TESTS=1 to enable them.")
     def test_enroll_bad_pw(self):
         """
         Try to enroll the beat in Kibana Central Management with a bad password
@@ -67,7 +86,8 @@ def test_enroll_bad_pw(self):
         new_content = open(config_path, 'r').read()
         assert config_content == new_content
 
-    @unittest.skip('Skipped because of issue: https://github.com/elastic/beats/issues/9597')
+    @unittest.skipIf(not INTEGRATION_TESTS,
+                     "integration tests are disabled, run with INTEGRATION_TESTS=1 to enable them.")
     def test_fetch_configs(self):
         """
         Config is retrieved from Central Management and updates are applied
@@ -78,6 +98,7 @@ def test_fetch_configs(self):
         exit_code = self.enroll(KIBANA_PASSWORD)
         assert exit_code == 0
 
+        index = self.random_index()
         # Configure an output
         self.create_and_assing_tag([
             {
@@ -86,9 +107,10 @@ def test_fetch_configs(self):
                     {
                         "output": "elasticsearch",
                         "elasticsearch": {
-                            "hosts": ["localhost:9200"],
-                            "username":"elastic",
-                            "password": KIBANA_PASSWORD,
+                            "hosts": [self.es_host],
+                            "username": self.es_user,
+                            "password": self.es_pass,
+                            "index": index,
                         }
                     }
                 ]
@@ -98,38 +120,41 @@ def test_fetch_configs(self):
         # Start beat
         proc = self.start_beat(extra_args=[
             "-E", "management.period=1s",
-            # do not blacklist file/elasticsearch outputs
-            "-E", "management.blacklist.output='foo'",
         ])
 
         # Wait for beat to apply new conf
         self.wait_log_contains("Applying settings for output")
 
+        self.wait_until(lambda: self.log_contains("PublishEvents: "))
+
+        self.wait_documents(index, 1)
+
+        index2 = self.random_index()
+
         # Update output configuration
         self.create_and_assing_tag([
             {
                 "type": "output",
                 "configs": [
                     {
-                        "output": "file",
-                        "file": {
-                            "path": os.path.join(self.working_dir, "output"),
-                            "filename": "mockbeat",
+                        "output": "elasticsearch",
+                        "elasticsearch": {
+                            "hosts": [self.es_host],
+                            "username": self.es_user,
+                            "password": self.es_pass,
+                            "index": index2,
                         }
                     }
                 ]
             }
         ])
-
-        # Wait for beat to apply new conf, now it logs to console
-        self.wait_until(
-            cond=lambda: self.log_contains_count("Applying settings for output") == 2)
-
-        self.wait_until(cond=lambda: self.output_has(1))
+        self.wait_until(lambda: self.log_contains("PublishEvents: "))
+        self.wait_documents(index2, 1)
 
         proc.check_kill_and_wait()
 
-    @unittest.skip('Skipped because of issue: https://github.com/elastic/beats/issues/9597')
+    @unittest.skipIf(not INTEGRATION_TESTS,
+                     "integration tests are disabled, run with INTEGRATION_TESTS=1 to enable them.")
     def test_configs_cache(self):
         """
         Config cache is used if Kibana is not available
@@ -140,83 +165,49 @@ def test_configs_cache(self):
         exit_code = self.enroll(KIBANA_PASSWORD)
         assert exit_code == 0
 
+        index = self.random_index()
+
         # Update output configuration
         self.create_and_assing_tag([
             {
                 "type": "output",
                 "configs": [
                     {
-                        "output": "file",
-                        "file": {
-                            "path": os.path.join(self.working_dir, "output"),
-                            "filename": "mockbeat_managed",
+
+                        "output": "elasticsearch",
+                        "elasticsearch": {
+                            "hosts": [self.es_host],
+                            "username": self.es_user,
+                            "password": self.es_pass,
+                            "index": index,
                         }
                     }
                 ]
             }
         ])
 
-        output_file = os.path.join("output", "mockbeat_managed")
-
         # Start beat
         proc = self.start_beat(extra_args=[
-            # do not blacklist file output
-            "-E", "management.blacklist.output='elasticsearch'",
+            "-E", "management.period=1s",
         ])
-        self.wait_until(cond=lambda: self.output_has(
-            1, output_file=output_file))
+
+        self.wait_until(lambda: self.log_contains("PublishEvents: "), )
+        self.wait_documents(index, 1)
         proc.check_kill_and_wait()
 
-        # Remove output file
-        os.remove(os.path.join(self.working_dir, output_file))
+        # Remove the index
+        self.es.indices.delete(index)
 
         # Cache should exists already, start with wrong kibana settings:
         proc = self.start_beat(extra_args=[
+            "-E", "management.period=1s",
             "-E", "management.kibana.host=wronghost",
             "-E", "management.kibana.timeout=0.5s",
-            # do not blacklist file output
-            "-E", "management.blacklist.output='elasticsearch'",
         ])
-        self.wait_until(cond=lambda: self.output_has(
-            1, output_file=output_file))
-        proc.check_kill_and_wait()
-
-    @unittest.skip('Skipped because of issue: https://github.com/elastic/beats/issues/9597')
-    def test_blacklist(self):
-        """
-        Blacklist blocks bad configs
-        """
-        # Enroll the beat
-        config_path = os.path.join(self.working_dir, "mockbeat.yml")
-        self.render_config_template("mockbeat", config_path)
-        exit_code = self.enroll(KIBANA_PASSWORD)
-        assert exit_code == 0
 
-        # Update output configuration
-        self.create_and_assing_tag([
-            {
-                "type": "output",
-                "configs": [
-                    {
-                        "output": "file",
-                        "file": {
-                            "path": os.path.join(self.working_dir, "output"),
-                            "filename": "mockbeat_managed",
-                        }
-                    }
-                ]
-            }
-        ])
-
-        output_file = os.path.join("output", "mockbeat_managed")
-
-        # Start beat
-        proc = self.start_beat()
-
-        self.wait_until(
-            cond=lambda: self.log_contains("Config for 'output' is blacklisted"))
+        self.wait_until(lambda: self.log_contains("PublishEvents: "))
+        self.wait_documents(index, 1)
         proc.check_kill_and_wait()
-        assert not os.path.isfile(os.path.join(self.working_dir, output_file))
 
     def enroll(self, password):
         return self.run_beat(
@@ -227,6 +218,18 @@ def enroll(self, password):
                 'PASS': password,
             })
 
+    def check_kibana_status(self):
+        headers = {
+            "kbn-xsrf": "1"
+        }
+
+        # Create tag
+        url = self.get_kibana_url() + "/api/status"
+
+        r = requests.get(url, headers=headers,
+                         auth=('elastic', KIBANA_PASSWORD))
+        print(r.text)
+
     def create_and_assing_tag(self, blocks):
         tag_name = "test"
         headers = {
@@ -255,5 +258,20 @@ def create_and_assing_tag(self, blocks):
                           auth=('elastic', KIBANA_PASSWORD))
         assert r.status_code == 200
 
+    def get_elasticsearch_url(self):
+        return 'http://' + self.es_user + ":" + self.es_pass + '@' + os.getenv('ES_HOST', 'localhost') + ':' + os.getenv('ES_PORT', '5601')
+
     def get_kibana_url(self):
         return 'http://' + os.getenv('KIBANA_HOST', 'kibana') + ':' + os.getenv('KIBANA_PORT', '5601')
+
+    def random_index(self):
+        return ''.join(random.choice(string.ascii_lowercase) for i in range(10))
+
+    def check_document_count(self, index, count):
+        try:
+            return self.es.search(index=index, body={"query": {"match_all": {}}})['hits']['total'] >= count
+        except:
+            return False
+
+    def wait_documents(self, index, count):
+        self.wait_until(lambda: self.check_document_count(index, count), max_timeout=TIMEOUT)