<commit_before>#!/usr/bin/env python

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Copyright (c) 2017 Mozilla Corporation


import os
import json
import time

import pytest
from dateutil.parser import parse

from rest_test_suite import RestTestSuite


class TestTestRoute(RestTestSuite):
    routes = ['/test', '/test/']

    status_code = 200
    body = ''


class TestStatusRoute(RestTestSuite):
    routes = ['/status', '/status/']

    status_code = 200
    body = '{"status": "ok", "service": "restapi"}'


class TestKibanaDashboardsRoute(RestTestSuite):
    routes = ['/kibanadashboards', '/kibanadashboards/']

    status_code = 200

    def teardown(self):
        super(TestKibanaDashboardsRoute, self).teardown()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)

    def setup(self):
        super(TestKibanaDashboardsRoute, self).setup()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)
            self.es_client.create_index('.kibana')

        json_dashboard_location = os.path.join(os.path.dirname(__file__), "ssh_dashboard.json")
        self.es_client.save_dashboard(json_dashboard_location, "Example SSH Dashboard")
        self.es_client.save_dashboard(json_dashboard_location, "Example FTP Dashboard")
        self.flush('.kibana')

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)

            assert response.status_code == self.status_code

            assert type(json_resp) == list
            assert len(json_resp) == 2

            json_resp.sort()

            assert json_resp[1]['url'].endswith("/app/kibana#/dashboard/Example-SSH-Dashboard") is True
            assert json_resp[1]['name'] == 'Example SSH Dashboard'

            assert json_resp[0]['url'].endswith("/app/kibana#/dashboard/Example-FTP-Dashboard") is True
            assert json_resp[0]['name'] == 'Example FTP Dashboard'


class TestKibanaDashboardsRouteWithoutDashboards(RestTestSuite):
    routes = ['/kibanadashboards', '/kibanadashboards/']

    status_code = 200

    def setup(self):
        super(TestKibanaDashboardsRouteWithoutDashboards, self).setup()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)
            self.es_client.create_index('.kibana')
        time.sleep(0.2)

    def teardown(self):
        super(TestKibanaDashboardsRouteWithoutDashboards, self).teardown()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)

            assert response.status_code == self.status_code
            assert json_resp == []


class TestLdapLoginsRoute(RestTestSuite):

    routes = ['/logincounts', '/logincounts/']
    status_code = 200

    def setup(self):
        super(TestLdapLoginsRoute, self).setup()

        # ttesterson test events
        for count in range(10):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Failed login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(5):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        # ttester test events
        for count in range(9):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Failed Login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(7):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "ldap"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        # qwerty test events
        for count in range(8):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "ldap"
                ],
                "timestamp": timestamp,
                "summary": "Failed Login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(3):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "ldap"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL>",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        for count in range(3):
            timestamp = RestTestSuite.subtract_from_timestamp({'hours': 2})
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "ldap"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL>",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        self.flush('events')

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)
<commit_msg>correct tags<commit_after>#!/usr/bin/env python

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Copyright (c) 2017 Mozilla Corporation


import os
import json
import time

import pytest
from dateutil.parser import parse

from rest_test_suite import RestTestSuite


class TestTestRoute(RestTestSuite):
    routes = ['/test', '/test/']

    status_code = 200
    body = ''


class TestStatusRoute(RestTestSuite):
    routes = ['/status', '/status/']

    status_code = 200
    body = '{"status": "ok", "service": "restapi"}'


class TestKibanaDashboardsRoute(RestTestSuite):
    routes = ['/kibanadashboards', '/kibanadashboards/']

    status_code = 200

    def teardown(self):
        super(TestKibanaDashboardsRoute, self).teardown()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)

    def setup(self):
        super(TestKibanaDashboardsRoute, self).setup()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)
            self.es_client.create_index('.kibana')

        json_dashboard_location = os.path.join(os.path.dirname(__file__), "ssh_dashboard.json")
        self.es_client.save_dashboard(json_dashboard_location, "Example SSH Dashboard")
        self.es_client.save_dashboard(json_dashboard_location, "Example FTP Dashboard")
        self.flush('.kibana')

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)

            assert response.status_code == self.status_code

            assert type(json_resp) == list
            assert len(json_resp) == 2

            json_resp.sort()

            assert json_resp[1]['url'].endswith("/app/kibana#/dashboard/Example-SSH-Dashboard") is True
            assert json_resp[1]['name'] == 'Example SSH Dashboard'

            assert json_resp[0]['url'].endswith("/app/kibana#/dashboard/Example-FTP-Dashboard") is True
            assert json_resp[0]['name'] == 'Example FTP Dashboard'


class TestKibanaDashboardsRouteWithoutDashboards(RestTestSuite):
    routes = ['/kibanadashboards', '/kibanadashboards/']

    status_code = 200

    def setup(self):
        super(TestKibanaDashboardsRouteWithoutDashboards, self).setup()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)
            self.es_client.create_index('.kibana')
        time.sleep(0.2)

    def teardown(self):
        super(TestKibanaDashboardsRouteWithoutDashboards, self).teardown()
        if pytest.config.option.delete_indexes:
            self.es_client.delete_index('.kibana', True)

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)

            assert response.status_code == self.status_code
            assert json_resp == []


class TestLdapLoginsRoute(RestTestSuite):

    routes = ['/logincounts', '/logincounts/']
    status_code = 200

    def setup(self):
        super(TestLdapLoginsRoute, self).setup()

        # ttesterson test events
        for count in range(10):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Failed login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(5):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        # ttester test events
        for count in range(9):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Failed Login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(7):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        # qwerty test events
        for count in range(8):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Failed Login from <EMAIL> srcIP=1.1.1.1",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Failed Login",
                }
            }
            self.populate_test_event(event)
        for count in range(3):
            timestamp = self.current_timestamp()
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL>",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        for count in range(3):
            timestamp = RestTestSuite.subtract_from_timestamp({'hours': 2})
            event = {
                "receivedtimestamp": timestamp,
                "utctimestamp": timestamp,
                "tags": [
                    "auth0"
                ],
                "timestamp": timestamp,
                "summary": "Success Login for <EMAIL>",
                "details": {
                    "username": "<EMAIL>",
                    "type": "Success Login",
                }
            }
            self.populate_test_event(event)

        self.flush('events')

    def test_route_endpoints(self):
        for route in self.routes:
            response = self.response_per_route(route)
            json_resp = json.loads(response.body)

{'fraction_non_alphanumeric': 0.1108694219234065, 'fraction_numerical': 0.008437149825324633, 'mean_word_length': 1.608665749656121, 'pattern_counts': {'":': 116, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5912253', 'n_tokens_mistral': 4276, 'n_tokens_neox': 4023, 'n_words': 835}