Skip to content

AppTestGenerator

Test Generator for an App. Generates test cases of Fields and CIM.

AppTestGenerator

Bases: object

Test Generator for an App. Generates test cases of Fields and CIM. The test generator is to include all the specific test generators.

AppTestGenerator should not have any direct generation methods, it should call a specific test generator methods only. Make sure there is no heavy initialization in init, all the configurations and operations should only take place in generate_tests method.

Parameters:

Name Type Description Default
pytest_config

To get the options given to pytest

required
Source code in pytest_splunk_addon/app_test_generator.py
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class AppTestGenerator(object):
    """
    Test Generator for an App.
    Generates test cases of Fields and CIM.
    The test generator is to include all the specific test generators.

    AppTestGenerator should not have any direct generation methods, it should call a specific
    test generator methods only. Make sure there is no heavy initialization in __init__, all the
    configurations and operations should only take place in generate_tests method.

    Args:
        pytest_config: To get the options given to pytest
    """

    def __init__(self, pytest_config):
        self.pytest_config = pytest_config
        self.seen_tests = set()

        store_events = self.pytest_config.getoption("store_events")
        config_path = self.pytest_config.getoption("splunk_data_generator")
        sample_generator = SampleXdistGenerator(
            self.pytest_config.getoption("splunk_app"), config_path
        )
        store_sample = sample_generator.get_samples(store_events)
        self.tokenized_events = store_sample.get("tokenized_events")
        LOGGER.debug("Initializing FieldTestGenerator to generate the test cases")
        self.fieldtest_generator = FieldTestGenerator(
            self.pytest_config.getoption("splunk_app"),
            self.tokenized_events,
            field_bank=self.pytest_config.getoption("field_bank", False),
        )

        data_model_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "data_models"
        )
        LOGGER.debug("Initializing CIMTestGenerator to generate the test cases")
        self.cim_test_generator = CIMTestGenerator(
            self.pytest_config.getoption("splunk_app"),
            self.pytest_config.getoption("splunk_dm_path") or data_model_path,
            self.tokenized_events,
        )
        self.indextime_test_generator = IndexTimeTestGenerator()

    def generate_tests(self, fixture):
        """
        Generate the test cases based on the fixture provided
        supported fixtures:

        * splunk_app_searchtime_*
        * splunk_app_cim_*
        * splunk_indextime

        Args:
            fixture(str): fixture name
        """
        if fixture in (
            "splunk_searchtime_fields_requirements",
            "splunk_searchtime_fields_datamodels",
        ):
            yield from self.fieldtest_generator.generate_tests(fixture)
        elif fixture == "splunk_searchtime_cim_fields_recommended":
            yield from self.cim_test_generator.generate_tests(fixture)
        elif fixture.startswith("splunk_searchtime_fields"):
            yield from self.dedup_tests(
                self.fieldtest_generator.generate_tests(fixture),
                fixture,
            )
        elif fixture.startswith("splunk_searchtime_cim"):
            yield from self.dedup_tests(
                self.cim_test_generator.generate_tests(fixture),
                fixture,
            )

        elif fixture.startswith("splunk_indextime"):
            # TODO: What should be the id of the test case?
            # Sourcetype + Host + Key field + _count

            pytest_params = None

            store_events = self.pytest_config.getoption("store_events")
            app_path = self.pytest_config.getoption("splunk_app")
            config_path = self.pytest_config.getoption("splunk_data_generator")

            if "key_fields" in fixture:
                pytest_params = list(
                    self.indextime_test_generator.generate_tests(
                        store_events,
                        app_path=app_path,
                        config_path=config_path,
                        test_type="key_fields",
                    )
                )

            elif "_time" in fixture:
                pytest_params = list(
                    self.indextime_test_generator.generate_tests(
                        store_events,
                        app_path=app_path,
                        config_path=config_path,
                        test_type="_time",
                    )
                )

            elif "line_breaker" in fixture:
                pytest_params = list(
                    self.indextime_test_generator.generate_tests(
                        store_events,
                        app_path=app_path,
                        config_path=config_path,
                        test_type="line_breaker",
                    )
                )

            yield from sorted(pytest_params, key=lambda param: param.id)

    def dedup_tests(self, test_list, fixture):
        """
        Deduplicate the test case parameters based on param.id

        Args:
            test_list (Generator): Generator of pytest.param
            fixture (str): fixture name

        Yields:
            Generator: De-duplicated pytest.param
        """
        param_list = []
        for each_param in test_list:
            if (fixture, each_param.id) not in self.seen_tests:
                param_list.append(each_param)
                self.seen_tests.add((fixture, each_param.id))

        # Sort the test generated.
        # ACD-4138: As pytest-xdist expects the tests to be ordered
        return sorted(param_list, key=lambda param: param.id)

dedup_tests(test_list, fixture)

Deduplicate the test case parameters based on param.id

Parameters:

Name Type Description Default
test_list Generator

Generator of pytest.param

required
fixture str

fixture name

required

Yields:

Name Type Description
Generator

De-duplicated pytest.param

Source code in pytest_splunk_addon/app_test_generator.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def dedup_tests(self, test_list, fixture):
    """
    Deduplicate the test case parameters based on param.id

    Args:
        test_list (Generator): Generator of pytest.param
        fixture (str): fixture name

    Yields:
        Generator: De-duplicated pytest.param
    """
    param_list = []
    for each_param in test_list:
        if (fixture, each_param.id) not in self.seen_tests:
            param_list.append(each_param)
            self.seen_tests.add((fixture, each_param.id))

    # Sort the test generated.
    # ACD-4138: As pytest-xdist expects the tests to be ordered
    return sorted(param_list, key=lambda param: param.id)

generate_tests(fixture)

Generate the test cases based on the fixture provided supported fixtures:

  • splunk_app_searchtime_*
  • splunk_app_cim_*
  • splunk_indextime

Parameters:

Name Type Description Default
fixture(str)

fixture name

required
Source code in pytest_splunk_addon/app_test_generator.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def generate_tests(self, fixture):
    """
    Generate the test cases based on the fixture provided
    supported fixtures:

    * splunk_app_searchtime_*
    * splunk_app_cim_*
    * splunk_indextime

    Args:
        fixture(str): fixture name
    """
    if fixture in (
        "splunk_searchtime_fields_requirements",
        "splunk_searchtime_fields_datamodels",
    ):
        yield from self.fieldtest_generator.generate_tests(fixture)
    elif fixture == "splunk_searchtime_cim_fields_recommended":
        yield from self.cim_test_generator.generate_tests(fixture)
    elif fixture.startswith("splunk_searchtime_fields"):
        yield from self.dedup_tests(
            self.fieldtest_generator.generate_tests(fixture),
            fixture,
        )
    elif fixture.startswith("splunk_searchtime_cim"):
        yield from self.dedup_tests(
            self.cim_test_generator.generate_tests(fixture),
            fixture,
        )

    elif fixture.startswith("splunk_indextime"):
        # TODO: What should be the id of the test case?
        # Sourcetype + Host + Key field + _count

        pytest_params = None

        store_events = self.pytest_config.getoption("store_events")
        app_path = self.pytest_config.getoption("splunk_app")
        config_path = self.pytest_config.getoption("splunk_data_generator")

        if "key_fields" in fixture:
            pytest_params = list(
                self.indextime_test_generator.generate_tests(
                    store_events,
                    app_path=app_path,
                    config_path=config_path,
                    test_type="key_fields",
                )
            )

        elif "_time" in fixture:
            pytest_params = list(
                self.indextime_test_generator.generate_tests(
                    store_events,
                    app_path=app_path,
                    config_path=config_path,
                    test_type="_time",
                )
            )

        elif "line_breaker" in fixture:
            pytest_params = list(
                self.indextime_test_generator.generate_tests(
                    store_events,
                    app_path=app_path,
                    config_path=config_path,
                    test_type="line_breaker",
                )
            )

        yield from sorted(pytest_params, key=lambda param: param.id)