|
1 | 1 | # pylint: disable=missing-docstring |
2 | 2 | # pylint: disable=protected-access |
3 | | -import pytest |
4 | | - |
5 | 3 | import json |
6 | 4 | from pathlib import Path |
| 5 | +from string import Template |
7 | 6 | from unittest import mock |
| 7 | + |
| 8 | +import pytest |
8 | 9 | import responses |
9 | 10 |
|
10 | 11 | from logprep.factory import Factory |
@@ -398,3 +399,60 @@ def test_match_list_field(self, testcase, system, result): |
398 | 399 | processor.setup() |
399 | 400 | processor.process(document) |
400 | 401 | assert document == expected, testcase |
| 402 | + |
| 403 | + @pytest.mark.parametrize( |
| 404 | + "http_list_content, expected_result", |
| 405 | + [ |
| 406 | + ("", set()), |
| 407 | + ("\n", {""}), |
| 408 | + ], |
| 409 | + ) |
| 410 | + @responses.activate |
| 411 | + def test_list_comparison_empty_http_list_or_empty_line_updates_compare_sets( |
| 412 | + self, |
| 413 | + http_list_content, |
| 414 | + expected_result, |
| 415 | + ): |
| 416 | + document = {"user": "Foo"} |
| 417 | + expected_document = {"user": "Foo", "user_results": {"not_in_list": ["bad_users.list"]}} |
| 418 | + |
| 419 | + url_template = "http://localhost/tests/testdata/${LOGPREP_LIST}?ref=bla" |
| 420 | + list_name = "bad_users.list" |
| 421 | + url = Template(url_template).substitute({"LOGPREP_LIST": list_name}) |
| 422 | + |
| 423 | + responses.add( |
| 424 | + responses.GET, |
| 425 | + url=url, |
| 426 | + body=http_list_content, |
| 427 | + status=200, |
| 428 | + ) |
| 429 | + |
| 430 | + rule_dict = { |
| 431 | + "filter": "user", |
| 432 | + "list_comparison": { |
| 433 | + "source_fields": ["user"], |
| 434 | + "target_field": "user_results", |
| 435 | + "list_file_paths": [list_name], |
| 436 | + }, |
| 437 | + "description": "", |
| 438 | + } |
| 439 | + |
| 440 | + config = { |
| 441 | + "type": "list_comparison", |
| 442 | + "rules": [], |
| 443 | + "list_search_base_path": url_template, |
| 444 | + } |
| 445 | + |
| 446 | + HttpGetter._shared.clear() |
| 447 | + |
| 448 | + processor = Factory.create({"custom_lister": config}) |
| 449 | + rule = processor.rule_class.create_from_dict(rule_dict) |
| 450 | + processor._rule_tree.add_rule(rule) |
| 451 | + |
| 452 | + processor.setup() |
| 453 | + processor.process(document) |
| 454 | + |
| 455 | + assert document == expected_document |
| 456 | + assert len(responses.calls) == 1 |
| 457 | + assert responses.calls[0].request.url == url |
| 458 | + assert rule.compare_sets[list_name] == expected_result |
0 commit comments