from datetime import datetime, timedelta
from django.urls import reverse
from django.test import TestCase
from django.utils.crypto import get_random_string
from prometheus_client.parser import text_string_to_metric_families
from zentral.contrib.munki.models import ManagedInstall, MunkiState
from zentral.conf import settings


class MunkiMetricsViewsTestCase(TestCase):
    # utility methods

    def _force_managed_install(self, failed=False, count=1, reinstall=False, age_days=22):
        mi = ManagedInstall.objects.create(
            machine_serial_number=get_random_string(),
            name=get_random_string(),
            installed_version=get_random_string(),
            installed_at=datetime.now(),
            failed_version=get_random_string() if failed else None,
            failed_at=datetime.now() if failed else None,
            reinstall=reinstall,
        )
        last_seen = datetime.utcnow() - timedelta(days=age_days)
        ms = MunkiState.objects.create(machine_serial_number=mi.machine_serial_number)
        MunkiState.objects.filter(pk=ms.pk).update(last_seen=last_seen)
        return mi

    def _make_authenticated_request(self):
        return self.client.get(reverse("munki_metrics:all"),
                               HTTP_AUTHORIZATION=f'Bearer {settings["api"]["metrics_bearer_token"]}')

    # metrics

    def test_metrics_permission_denied(self):
        response = self.client.get(reverse("munki_metrics:all"))
        self.assertEqual(response.status_code, 403)

    def test_metrics_permission_ok(self):
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)

    def test_active_machines(self):
        for age in (2, 22, 31):
            ms = MunkiState.objects.create(machine_serial_number=get_random_string())
            MunkiState.objects.filter(pk=ms.pk).update(last_seen=datetime.utcnow() - timedelta(days=age))
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_active_machines_bucket":
                continue
            else:
                self.assertEqual(len(family.samples), 7)
                for sample in family.samples:
                    le = sample.labels["le"]
                    if le == "1":
                        self.assertEqual(sample.value, 0)
                    elif le in ("7", "14"):
                        self.assertEqual(sample.value, 1)
                    elif le == "30":
                        self.assertEqual(sample.value, 2)
                    else:
                        self.assertEqual(sample.value, 3)
                break
        else:
            raise AssertionError("could not find expected metric family")

    def test_installed_pkginfos(self):
        mi = self._force_managed_install(age_days=22)
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_installed_pkginfos_bucket":
                continue
            else:
                self.assertEqual(len(family.samples), 7)
                for sample in family.samples:
                    self.assertEqual(sample.labels["name"], mi.name)
                    self.assertEqual(sample.labels["version"], mi.installed_version)
                    if sample.labels["le"] in ("1", "7", "14"):
                        # last seen 22 days ago
                        self.assertEqual(sample.value, 0)
                    else:
                        self.assertEqual(sample.value, 1)
                break
        else:
            raise AssertionError("could not find expected metric family")

    def test_failed_pkginfos(self):
        mi = self._force_managed_install(failed=True)
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_failed_pkginfos":
                continue
            else:
                self.assertEqual(len(family.samples), 1)
                sample = family.samples[0]
                self.assertEqual(sample.value, 1)
                self.assertEqual(sample.labels["name"], mi.name)
                self.assertEqual(sample.labels["version"], mi.failed_version)
                break
        else:
            raise AssertionError("could not find expected metric family")

from datetime import datetime, timedelta
from django.urls import reverse
from django.test import TestCase
from django.utils.crypto import get_random_string
from prometheus_client.parser import text_string_to_metric_families
from zentral.contrib.munki.models import ManagedInstall, MunkiState
from zentral.conf import settings


class MunkiMetricsViewsTestCase(TestCase):
    # utility methods

    def _force_managed_install(self, failed=False, count=1, reinstall=False, age_days=22):
        mi = ManagedInstall.objects.create(
            machine_serial_number=get_random_string(),
            name=get_random_string(),
            installed_version=get_random_string(),
            installed_at=datetime.now(),
            failed_version=get_random_string() if failed else None,
            failed_at=datetime.now() if failed else None,
            reinstall=reinstall,
        )
        last_seen = datetime.utcnow() - timedelta(days=age_days)
        ms = MunkiState.objects.create(machine_serial_number=mi.machine_serial_number)
        MunkiState.objects.filter(pk=ms.pk).update(last_seen=last_seen)
        return mi

    def _make_authenticated_request(self):
        return self.client.get(reverse("munki_metrics:all"),
                               HTTP_AUTHORIZATION=f'Bearer {settings["api"]["metrics_bearer_token"]}')

    # metrics

    def test_metrics_permission_denied(self):
        response = self.client.get(reverse("munki_metrics:all"))
        self.assertEqual(response.status_code, 403)

    def test_metrics_permission_ok(self):
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)

    def test_active_machines(self):
        for age in (2, 22, 31):
            ms = MunkiState.objects.create(machine_serial_number=get_random_string())
            MunkiState.objects.filter(pk=ms.pk).update(last_seen=datetime.utcnow() - timedelta(days=age))
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_active_machines_bucket":
                continue
            else:
                self.assertEqual(len(family.samples), 7)
                for sample in family.samples:
                    le = sample.labels["le"]
                    if le == "1":
                        self.assertEqual(sample.value, 0)
                    elif le in ("7", "14"):
                        self.assertEqual(sample.value, 1)
                    elif le == "30":
                        self.assertEqual(sample.value, 2)
                    else:
                        self.assertEqual(sample.value, 3)
                break
        else:
            raise AssertionError("could not find expected metric family")

    def test_installed_pkginfos(self):
        mi = self._force_managed_install(age_days=22)
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_installed_pkginfos_bucket":
                continue
            else:
                self.assertEqual(len(family.samples), 7)
                for sample in family.samples:
                    self.assertEqual(sample.labels["name"], mi.name)
                    self.assertEqual(sample.labels["version"], mi.installed_version)
                    if sample.labels["le"] in ("1", "7", "14"):
                        # last seen 22 days ago
                        self.assertEqual(sample.value, 0)
                    else:
                        self.assertEqual(sample.value, 1)
                break
        else:
            raise AssertionError("could not find expected metric family")

    def test_failed_pkginfos(self):
        mi = self._force_managed_install(failed=True)
        response = self._make_authenticated_request()
        self.assertEqual(response.status_code, 200)
        for family in text_string_to_metric_families(response.content.decode("utf-8")):
            if family.name != "zentral_munki_failed_pkginfos":
                continue
            else:
                self.assertEqual(len(family.samples), 1)
                sample = family.samples[0]
                self.assertEqual(sample.value, 1)
                self.assertEqual(sample.labels["name"], mi.name)
                self.assertEqual(sample.labels["version"], mi.failed_version)
                break
        else:
            raise AssertionError("could not find expected metric family")

# utility methods # metrics # last seen 22 days ago