import metricbeat
import os
import pytest
import sys
import unittest


class Test(metricbeat.BaseTest):

    COMPOSE_SERVICES = ['postgresql']

    def common_checks(self, output):
        # Ensure no errors or warnings exist in the log.
        self.assert_no_logged_warnings()

        for evt in output:
            top_level_fields = metricbeat.COMMON_FIELDS + ["postgresql"]
            self.assertCountEqual(self.de_dot(top_level_fields), evt.keys())

            self.assert_fields_are_documented(evt)

    def get_hosts(self):
        username = "postgres"
        host = self.compose_host()
        dsn = "postgres://{}?sslmode=disable".format(host)
        return (
            [dsn],
            username,
            os.getenv("POSTGRESQL_PASSWORD"),
        )

    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")
    @pytest.mark.tag('integration')
    def test_activity(self):
        """
        PostgreSQL module outputs an event.
        """
        hosts, username, password = self.get_hosts()
        self.render_config_template(modules=[{
            "name": "postgresql",
            "metricsets": ["activity"],
            "hosts": hosts,
            "username": username,
            "password": password,
            "period": "5s"
        }])
        proc = self.start_beat()
        self.wait_until(lambda: self.output_lines() > 0)
        proc.check_kill_and_wait()

        output = self.read_output_json()
        self.common_checks(output)

        for evt in output:
            assert "name" in evt["postgresql"]["activity"]["database"]
            assert "oid" in evt["postgresql"]["activity"]["database"]
            assert "state" in evt["postgresql"]["activity"]

    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")
    @pytest.mark.tag('integration')
    def test_database(self):
        """
        PostgreSQL module outputs an event.
        """
        hosts, username, password = self.get_hosts()
        self.render_config_template(modules=[{
            "name": "postgresql",
            "metricsets": ["database"],
            "hosts": hosts,
            "username": username,
            "password": password,
            "period": "5s"
        }])
        proc = self.start_beat()
        self.wait_until(lambda: self.output_lines() > 0)
        proc.check_kill_and_wait()

        output = self.read_output_json()
        self.common_checks(output)

        for evt in output:
            assert "name" in evt["postgresql"]["database"]
            assert "oid" in evt["postgresql"]["database"]
            assert "blocks" in evt["postgresql"]["database"]
            assert "rows" in evt["postgresql"]["database"]
            assert "conflicts" in evt["postgresql"]["database"]
            assert "deadlocks" in evt["postgresql"]["database"]

    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")
    @pytest.mark.tag('integration')
    def test_bgwriter(self):
        """
        PostgreSQL module outputs an event.
        """
        hosts, username, password = self.get_hosts()
        self.render_config_template(modules=[{
            "name": "postgresql",
            "metricsets": ["bgwriter"],
            "hosts": hosts,
            "username": username,
            "password": password,
            "period": "5s"
        }])
        proc = self.start_beat()
        self.wait_until(lambda: self.output_lines() > 0)
        proc.check_kill_and_wait()

        output = self.read_output_json()
        self.common_checks(output)

        for evt in output:
            assert "checkpoints" in evt["postgresql"]["bgwriter"]
            assert "buffers" in evt["postgresql"]["bgwriter"]
            assert "stats_reset" in evt["postgresql"]["bgwriter"]

[((32, 5, 32, 74), 'unittest.skipUnless', 'unittest.skipUnless', ({(32, 25, 32, 53): 'metricbeat.INTEGRATION_TESTS', (32, 55, 32, 73): '"""integration test"""'}, {}), "(metricbeat.INTEGRATION_TESTS, 'integration test')", False, 'import unittest\n'), ((33, 5, 33, 35), 'pytest.mark.tag', 'pytest.mark.tag', ({(33, 21, 33, 34): '"""integration"""'}, {}), "('integration')", False, 'import pytest\n'), ((59, 5, 59, 74), 'unittest.skipUnless', 'unittest.skipUnless', ({(59, 25, 59, 53): 'metricbeat.INTEGRATION_TESTS', (59, 55, 59, 73): '"""integration test"""'}, {}), "(metricbeat.INTEGRATION_TESTS, 'integration test')", False, 'import unittest\n'), ((60, 5, 60, 35), 'pytest.mark.tag', 'pytest.mark.tag', ({(60, 21, 60, 34): '"""integration"""'}, {}), "('integration')", False, 'import pytest\n'), ((89, 5, 89, 74), 'unittest.skipUnless', 'unittest.skipUnless', ({(89, 25, 89, 53): 'metricbeat.INTEGRATION_TESTS', (89, 55, 89, 73): '"""integration test"""'}, {}), "(metricbeat.INTEGRATION_TESTS, 'integration test')", False, 'import unittest\n'), ((90, 5, 90, 35), 'pytest.mark.tag', 'pytest.mark.tag', ({(90, 21, 90, 34): '"""integration"""'}, {}), "('integration')", False, 'import pytest\n'), ((29, 12, 29, 44), 'os.getenv', 'os.getenv', ({(29, 22, 29, 43): '"""POSTGRESQL_PASSWORD"""'}, {}), "('POSTGRESQL_PASSWORD')", False, 'import os\n')]