109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
from app.services.processing import ProcessingOptions, filter_records, sort_records
|
|
|
|
|
|
def test_filter_records_supports_case_insensitive_filters():
|
|
records = [
|
|
{"policy": "ProdPolicy", "severity_level": "HIGH"},
|
|
{"policy": "OtherPolicy", "severity_level": "low"},
|
|
]
|
|
options = ProcessingOptions(
|
|
policy_cs="",
|
|
policy_ci="prod",
|
|
severity_cs="",
|
|
severity_ci="high",
|
|
sort_by="datetime",
|
|
order="asc",
|
|
mode="vendor",
|
|
)
|
|
|
|
filtered = list(filter_records(records, options))
|
|
|
|
assert filtered == [{"policy": "ProdPolicy", "severity_level": "HIGH"}]
|
|
|
|
|
|
def test_sort_records_by_severity_desc_uses_defined_ranking():
|
|
records = [
|
|
{"severity_level": "medium"},
|
|
{"severity_level": "critical"},
|
|
{"severity_level": "info"},
|
|
]
|
|
options = ProcessingOptions(
|
|
policy_cs="",
|
|
policy_ci="",
|
|
severity_cs="",
|
|
severity_ci="",
|
|
sort_by="severity",
|
|
order="desc",
|
|
mode="vendor",
|
|
)
|
|
|
|
sorted_records = sort_records(records, options)
|
|
|
|
assert [record["severity_level"] for record in sorted_records] == [
|
|
"critical",
|
|
"medium",
|
|
"info",
|
|
]
|
|
|
|
|
|
def test_sort_records_by_datetime_asc_places_invalid_records_last():
|
|
records = [
|
|
{"v015xxxxdate": "2024-05-03", "time": "08:00:00", "msg": "latest-valid"},
|
|
{"v015xxxxdate": "", "time": "09:00:00", "msg": "missing-date"},
|
|
{"v015xxxxdate": "2024-05-01", "time": "10:00:00", "msg": "earliest-valid"},
|
|
{"v015xxxxdate": "2024-05-02", "time": "", "msg": "missing-time"},
|
|
{"v015xxxxdate": "bad-date", "time": "99:99:99", "msg": "invalid-datetime"},
|
|
{"v015xxxxdate": "2024-05-02", "time": "09:30:00", "msg": "middle-valid"},
|
|
]
|
|
options = ProcessingOptions(
|
|
policy_cs="",
|
|
policy_ci="",
|
|
severity_cs="",
|
|
severity_ci="",
|
|
sort_by="datetime",
|
|
order="asc",
|
|
mode="vendor",
|
|
)
|
|
|
|
sorted_records = sort_records(records, options)
|
|
|
|
assert [record["msg"] for record in sorted_records] == [
|
|
"earliest-valid",
|
|
"middle-valid",
|
|
"latest-valid",
|
|
"missing-date",
|
|
"missing-time",
|
|
"invalid-datetime",
|
|
]
|
|
|
|
|
|
def test_sort_records_by_datetime_desc_places_invalid_records_last():
|
|
records = [
|
|
{"v015xxxxdate": "2024-05-03", "time": "08:00:00", "msg": "latest-valid"},
|
|
{"v015xxxxdate": "", "time": "09:00:00", "msg": "missing-date"},
|
|
{"v015xxxxdate": "2024-05-01", "time": "10:00:00", "msg": "earliest-valid"},
|
|
{"v015xxxxdate": "2024-05-02", "time": "", "msg": "missing-time"},
|
|
{"v015xxxxdate": "bad-date", "time": "99:99:99", "msg": "invalid-datetime"},
|
|
{"v015xxxxdate": "2024-05-02", "time": "09:30:00", "msg": "middle-valid"},
|
|
]
|
|
options = ProcessingOptions(
|
|
policy_cs="",
|
|
policy_ci="",
|
|
severity_cs="",
|
|
severity_ci="",
|
|
sort_by="datetime",
|
|
order="desc",
|
|
mode="vendor",
|
|
)
|
|
|
|
sorted_records = sort_records(records, options)
|
|
|
|
assert [record["msg"] for record in sorted_records] == [
|
|
"latest-valid",
|
|
"middle-valid",
|
|
"earliest-valid",
|
|
"missing-date",
|
|
"missing-time",
|
|
"invalid-datetime",
|
|
]
|