test_input.py

#!/usr/bin/env python
#
# Copyright © 2011-2023 Splunk, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"): you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import logging
import pytest
from splunklib.binding import HTTPError

from tests import testlib
from splunklib import client


def highest_port(service, base_port, *kinds):
    """Find the first port >= base_port not in use by any input in kinds."""
    highest_port = base_port
    for input in service.inputs.list(*kinds):
        port = int(input.name.split(':')[-1])
        highest_port = max(port, highest_port)
    return highest_port


class TestTcpInputNameHandling(testlib.SDKTestCase):
    def setUp(self):
        super().setUp()
        self.base_port = highest_port(self.service, 10000, 'tcp', 'splunktcp', 'udp') + 1

    def tearDown(self):
        for input in self.service.inputs.list('tcp', 'splunktcp'):
            port = int(input.name.split(':')[-1])
            if port >= self.base_port:
                input.delete()
        super().tearDown()

    def create_tcp_input(self, base_port, kind, **options):
        port = base_port
        while True:  # Find the next unbound port
            try:
                input = self.service.inputs.create(str(port), kind, **options)
                return input
            except client.HTTPError as he:
                if he.status == 400:
                    port += 1

    def test_create_tcp_port(self):
        for kind in ['tcp', 'splunktcp']:
            input = self.service.inputs.create(str(self.base_port), kind)
            self.check_entity(input)
            input.delete()

    def test_cannot_create_with_restrictToHost_in_name(self):
        self.assertRaises(
            client.HTTPError,
            lambda: self.service.inputs.create('boris:10000', 'tcp')
        )

    def test_create_tcp_ports_with_restrictToHost(self):
        for kind in ['tcp', 'splunktcp']:  # Multiplexed UDP ports are not supported
            # Make sure we can create two restricted inputs on the same port
            boris = self.service.inputs.create(str(self.base_port), kind, restrictToHost='boris')
            natasha = self.service.inputs.create(str(self.base_port), kind, restrictToHost='natasha')
            # And that they both function
            boris.refresh()
            natasha.refresh()
            self.check_entity(boris)
            self.check_entity(natasha)
            boris.delete()
            natasha.delete()

    # This test does not succeed on all OSes, disabling for now (it's really
    # testing Splunk, not the SDK)
    # def test_restricted_to_unrestricted_collision(self):
    #     for kind in ['tcp', 'splunktcp', 'udp']:
    #         restricted = self.service.inputs.create(str(self.base_port), kind, restrictToHost='boris')
    #         self.assertTrue('boris:' + str(self.base_port) in self.service.inputs)
    #         self.assertRaises(
    #             client.HTTPError,
    #             lambda: self.service.inputs.create(str(self.base_port), kind)
    #         )
    #         restricted.delete()

    def test_unrestricted_to_restricted_collision(self):
        for kind in ['tcp', 'splunktcp', 'udp']:
            unrestricted = self.service.inputs.create(str(self.base_port), kind)
            self.assertTrue(str(self.base_port) in self.service.inputs)
            self.assertRaises(
                client.HTTPError,
                lambda: self.service.inputs.create(str(self.base_port), kind, restrictToHost='boris')
            )
            unrestricted.delete()

    def test_update_restrictToHost_fails(self):
        for kind in ['tcp', 'splunktcp']:  # No UDP, since it's broken in Splunk
            boris = self.create_tcp_input(self.base_port, kind, restrictToHost='boris')

            self.assertRaises(
                client.IllegalOperationException,
                lambda: boris.update(restrictToHost='hilda')
            )


class TestRead(testlib.SDKTestCase):
    def test_read(self):
        inputs = self.service.inputs
        # count doesn't work on inputs; known problem tested for in
        # test_collection.py. This test will speed up dramatically
        # when that's fixed.
        for item in inputs.list(count=5):
            self.check_entity(item)
            item.refresh()
            self.check_entity(item)

    def test_read_kind(self):
        inputs = self.service.inputs
        logging.debug("Input kinds: %s", inputs.kinds)
        for kind in inputs.kinds:
            for item in inputs.list(kind, count=3):
                self.assertEqual(item.kind, kind)

    def test_inputs_list_on_one_kind(self):
        self.service.inputs.list('monitor')

    def test_read_invalid_input(self):
        name = testlib.tmpname()
        try:
            self.service.inputs.get(name)
            self.fail("Expected a 404 HTTPError")
        except HTTPError as he:
            self.assertTrue("HTTP 404 Not Found" in str(he))

    def test_inputs_list_on_one_kind_with_count(self):
        expected = [x.name for x in self.service.inputs.list('monitor')[:10]]
        found = [x.name for x in self.service.inputs.list('monitor', count=10)]
        self.assertEqual(expected, found)

    def test_inputs_list_on_one_kind_with_offset(self):
        N = 2
        expected = [x.name for x in self.service.inputs.list('monitor')[N:]]
        found = [x.name for x in self.service.inputs.list('monitor', offset=N)]
        self.assertEqual(expected, found)

    def test_inputs_list_on_one_kind_with_search(self):
        search = "SPLUNK"
        expected = [x.name for x in self.service.inputs.list('monitor') if search in x.name]
        found = [x.name for x in self.service.inputs.list('monitor', search=search)]
        self.assertEqual(expected, found)

    @pytest.mark.app
    def test_oneshot(self):
        self.install_app_from_collection('file_to_upload')

        index_name = testlib.tmpname()
        index = self.service.indexes.create(index_name)
        self.assertEventuallyTrue(lambda: index.refresh() and index['disabled'] == '0')

        eventCount = int(index['totalEventCount'])

        path = self.pathInApp("file_to_upload", ["log.txt"])
        self.service.inputs.oneshot(path, index=index_name)

        def f():
            index.refresh()
            return int(index['totalEventCount']) == eventCount + 4

        self.assertEventuallyTrue(f, timeout=60)

    def test_oneshot_on_nonexistant_file(self):
        name = testlib.tmpname()
        self.assertRaises(HTTPError,
                          self.service.inputs.oneshot, name)


class TestInput(testlib.SDKTestCase):
    def setUp(self):
        super().setUp()
        inputs = self.service.inputs
        unrestricted_port = str(highest_port(self.service, 10000, 'tcp', 'splunktcp', 'udp') + 1)
        restricted_port = str(highest_port(self.service, int(unrestricted_port) + 1, 'tcp', 'splunktcp') + 1)
        test_inputs = [{'kind': 'tcp', 'name': unrestricted_port, 'host': 'sdk-test'},
                       {'kind': 'udp', 'name': unrestricted_port, 'host': 'sdk-test'},
                       {'kind': 'tcp', 'name': 'boris:' + restricted_port, 'host': 'sdk-test'}]
        self._test_entities = {}

        self._test_entities['tcp'] = \
            inputs.create(unrestricted_port, 'tcp', host='sdk-test')
        self._test_entities['udp'] = \
            inputs.create(unrestricted_port, 'udp', host='sdk-test')
        self._test_entities['restrictedTcp'] = \
            inputs.create(restricted_port, 'tcp', restrictToHost='boris')

    def tearDown(self):
        super().tearDown()
        for entity in list(self._test_entities.values()):
            try:
                self.service.inputs.delete(
                    kind=entity.kind,
                    name=entity.name)
            except KeyError:
                pass

    def test_list(self):
        inputs = self.service.inputs
        input_list = inputs.list()
        self.assertTrue(len(input_list) > 0)
        for input in input_list:
            self.assertTrue(input.name is not None)

    @pytest.mark.app
    def test_lists_modular_inputs(self):
        # Install modular inputs to list, and restart
        # so they'll show up.
        self.install_app_from_collection("modular_inputs")
        self.uncheckedRestartSplunk()

        inputs = self.service.inputs
        if ('abcd', 'test2') not in inputs:
            inputs.create('abcd', 'test2', field1='boris')

        input = inputs['abcd', 'test2']
        self.assertEqual(input.field1, 'boris')

    def test_create(self):
        inputs = self.service.inputs
        for entity in list(self._test_entities.values()):
            self.check_entity(entity)
            self.assertTrue(isinstance(entity, client.Input))

    def test_get_kind_list(self):
        inputs = self.service.inputs
        kinds = inputs._get_kind_list()
        self.assertTrue('tcp/raw' in kinds)

    def test_read(self):
        inputs = self.service.inputs
        for this_entity in list(self._test_entities.values()):
            kind, name = this_entity.kind, this_entity.name
            read_entity = inputs[name, kind]
            self.assertEqual(this_entity.kind, read_entity.kind)
            self.assertEqual(this_entity.name, read_entity.name)
            self.assertEqual(this_entity.host, read_entity.host)

    def test_read_indiviually(self):
        tcp_input = self.service.input(self._test_entities['tcp'].path,
                                       self._test_entities['tcp'].kind)
        self.assertIsNotNone(tcp_input)
        self.assertTrue('tcp', tcp_input.kind)
        self.assertTrue(self._test_entities['tcp'].name, tcp_input.name)

    def test_update(self):
        inputs = self.service.inputs
        for entity in list(self._test_entities.values()):
            kind, name = entity.kind, entity.name
            kwargs = {'host': 'foo'}
            entity.update(**kwargs)
            entity.refresh()
            self.assertEqual(entity.host, kwargs['host'])

    @pytest.mark.skip('flaky')
    def test_delete(self):
        inputs = self.service.inputs
        remaining = len(self._test_entities) - 1
        for input_entity in list(self._test_entities.values()):
            name = input_entity.name
            kind = input_entity.kind
            self.assertTrue(name in inputs)
            self.assertTrue((name, kind) in inputs)
            if remaining == 0:
                inputs.delete(name)
                self.assertFalse(name in inputs)
            else:
                if not name.startswith('boris'):
                    self.assertRaises(client.AmbiguousReferenceException,
                                      inputs.delete, name)
                self.service.inputs.delete(name, kind)
                self.assertFalse((name, kind) in inputs)
            self.assertRaises(client.HTTPError,
                              input_entity.refresh)
            remaining -= 1


if __name__ == "__main__":
    import unittest

    unittest.main()