Cover every line and branch of the package

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 3.0.1 - 2026-09-05
+
+### Changed
+
+- Tests now cover every line and branch of the package: the type coercion
+  helpers, the per-log post-processors' edge cases, the index mapping for
+  every field shape, and the CLI and reader error paths.
+- The ECS index mapping skips children of `flattened` fields, which
+  Elasticsearch cannot map; the published field table has none, so the
+  generated mapping is unchanged.
+- Removed the unused `ZeekLog.field_types` property; use `ZeekLog.types`.
+
 ## 3.0.0 - 2026-09-05
 
 Rewrite for current Python and Zeek. The 2.x `ParseZeekLogs` iterator still

parsezeeklogs/__init__.py

@@ -23,7 +23,7 @@
     write_json_lines,
 )
 
-__version__ = "3.0.0"
+__version__ = "3.0.1"
 
 __all__ = [
     "ParseZeekLogs",

parsezeeklogs/ecs.py

@@ -743,11 +743,8 @@ def unflatten(flat: Flat) -> dict[str, Any]:
                 node[part] = child_obj
                 child = child_obj
             node = child
-        leaf = parts[-1]
-        if isinstance(node.get(leaf), dict):
-            node[leaf]["value"] = flat[key]
-        else:
-            node[leaf] = flat[key]
+        # Keys are visited shortest first, so a leaf can never already be an object here.
+        node[parts[-1]] = flat[key]
     return out
 
 
@@ -806,6 +803,17 @@ def to_ecs(
 _CONTAINER_TYPES = {"object", "nested", "flattened"}
 
 
+def _mapping_parent(root: dict[str, Any], parts: list[str]) -> dict[str, Any] | None:
+    """The ``properties`` dict that holds ``parts``; None when an ancestor is flattened."""
+    node = root
+    for part in parts:
+        node = node.setdefault(part, {"type": "object", "properties": {}})
+        if node.get("type") == "flattened":
+            return None
+        node = node.setdefault("properties", {})
+    return node
+
+
 def ecs_index_body() -> dict[str, Any]:
     """Elasticsearch index body whose mapping mirrors :data:`FIELD_TYPES`."""
     types = dict(FIELD_TYPES)
@@ -816,12 +824,9 @@ def ecs_index_body() -> dict[str, Any]:
     for key in sorted(types, key=lambda k: (k.count("."), k)):
         ftype = types[key]
         parts = key.split(".")
-        node = root
-        for part in parts[:-1]:
-            node = node.setdefault(part, {"type": "object", "properties": {}})
-            if ftype != "flattened" and "properties" not in node:
-                node["properties"] = {}
-            node = node["properties"]
+        node = _mapping_parent(root, parts[:-1])
+        if node is None:
+            continue  # children of a flattened field are not mapped
         leaf = parts[-1]
         if key in prefixes and ftype not in _CONTAINER_TYPES:
             # A scalar that is also a parent: Elasticsearch cannot map both, keep the object.

parsezeeklogs/reader.py

@@ -100,10 +100,6 @@ def apply(self, line: str) -> None:
             self.types = rest.split(self.separator) if rest else []
         self.directives[key] = rest
 
-    @property
-    def field_types(self) -> dict[str, str]:
-        return dict(zip(self.fields, self.types, strict=False))
-
 
 def _format_time(value: float, time_format: TimeFormat) -> Any:
     if time_format == "epoch":

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "parsezeeklogs"
-version = "3.0.0"
+version = "3.0.1"
 description = "Read Zeek (Bro) network security monitor logs, TSV or JSON, into typed Python records; convert to JSON or CSV; load into Elasticsearch."
 readme = "README.md"
 license = "Apache-2.0"
@@ -73,6 +73,9 @@ source = ["parsezeeklogs"]
 branch = true
 omit = ["parsezeeklogs/__main__.py"]
 
+[tool.coverage.report]
+exclude_also = ["if __name__ == .__main__.:"]
+
 [tool.ruff]
 line-length = 100
 target-version = "py310"

sonar-project.properties

@@ -12,3 +12,5 @@ sonar.exclusions=parsezeeklogs/_ecs_tables.py,**/__pycache__/**,**/.venv/**,**/.
 sonar.sourceEncoding=UTF-8
 sonar.python.version=3.10,3.11,3.12,3.13,3.14
 sonar.python.coverage.reportPaths=coverage.xml
+# The -m entry point is exercised by a subprocess test, which coverage cannot see.
+sonar.coverage.exclusions=parsezeeklogs/__main__.py

tests/test_cli.py

@@ -43,6 +43,13 @@ def test_json_to_file_multiple_inputs(tmp_path, data_dir, caplog):
     assert f"{len(lines)} records written" in caplog.text
 
 
+def test_csv_to_file_logs_count(tmp_path, data_dir, caplog):
+    out = tmp_path / "out.csv"
+    assert cli.main(["csv", str(data_dir / RDP), "-o", str(out)]) == 0
+    assert len(out.read_text().splitlines()) == _data_lines(data_dir / RDP) + 1
+    assert f"{_data_lines(data_dir / RDP)} records written" in caplog.text
+
+
 def test_csv_default_fields_and_safe_headers(capsys, data_dir):
     assert cli.main(["csv", str(data_dir / RDP), "--safe-headers", "-m", '{"tag": "x"}']) == 0
     out = capsys.readouterr().out.splitlines()
@@ -199,6 +206,15 @@ def test_elk_transport_error(data_dir, caplog):
     assert "Elasticsearch error" in caplog.text
 
 
+def test_elk_unreadable_input(data_dir, caplog):
+    with (
+        mock.patch("parsezeeklogs.elastic.make_client", return_value=mock.Mock()),
+        mock.patch("parsezeeklogs.elastic.ZeekToElk.load_many", side_effect=OSError("gone")),
+    ):
+        assert cli.main(["elk", str(data_dir / RDP), "localhost"]) == 1
+    assert "cannot read input" in caplog.text
+
+
 def test_elk_without_extra(data_dir, caplog):
     with mock.patch("parsezeeklogs.elastic._require_elasticsearch", side_effect=ImportError("x")):
         assert cli.main(["elk", str(data_dir / RDP), "localhost"]) == 1

tests/test_ecs_branches.py

@@ -0,0 +1,256 @@
+"""Edge cases of the ECS helpers and per-log post-processors that the fixtures never hit."""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from parsezeeklogs import ecs
+from parsezeeklogs.ecs import ecs_index_body, unflatten
+
+NS = "zeek.x"
+
+
+# -- coercion helpers ----------------------------------------------------------------------
+
+
+def test_append_variants():
+    flat = {}
+    ecs._append(flat, "k", None)
+    assert flat == {}
+    ecs._append(flat, "k", "a")
+    ecs._append(flat, "k", "a")  # scalar, same value: unchanged
+    assert flat["k"] == ["a"]
+    ecs._append(flat, "k", "b")
+    ecs._append(flat, "k", "b")  # list, duplicate: unchanged
+    assert flat["k"] == ["a", "b"]
+    flat["s"] = "x"
+    ecs._append(flat, "s", "x")
+    assert flat["s"] == "x"
+    ecs._append(flat, "s", "y")
+    assert flat["s"] == ["x", "y"]
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (True, 1),
+        (7, 7),
+        (2.6, 3),
+        ("0x1F", 31),
+        (" 12 ", 12),
+        ("3.7", 4),
+        ("abc", None),
+        (None, None),
+        ([1], None),
+    ],
+)
+def test_as_int(value, expected):
+    assert ecs._as_int(value) == expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [(True, 1.0), (2, 2.0), ("1.5", 1.5), ("x", None), (None, None)],
+)
+def test_as_float(value, expected):
+    assert ecs._as_float(value) == expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        (False, False),
+        (0, False),
+        (2.0, True),
+        ("T", True),
+        ("yes", True),
+        ("F", False),
+        ("no", False),
+        ("maybe", None),
+        (None, None),
+    ],
+)
+def test_as_bool(value, expected):
+    assert ecs._as_bool(value) is expected
+
+
+def test_as_str_and_ip():
+    assert ecs._as_str(True) == "true"
+    assert ecs._as_str(False) == "false"
+    stamp = datetime(2020, 9, 13, 12, 26, 40, tzinfo=timezone.utc)
+    assert ecs._as_str(stamp) == "2020-09-13T12:26:40+00:00"
+    assert ecs._as_str(5) == "5"
+    assert ecs._as_ip(" 10.0.0.1 ") == "10.0.0.1"
+    assert ecs._as_ip("not-an-ip") is None
+
+
+def test_parse_dn_skips_unknown_and_malformed_parts():
+    assert ecs._parse_dn("CN=a\\, b,X=1,junk,O=Org") == {
+        "common_name": "a b",
+        "organization": "Org",
+    }
+
+
+# -- post-processors -----------------------------------------------------------------------
+
+
+def test_dns_edge_cases():
+    flat = {f"{NS}.qclass": 0, f"{NS}.answers": ["a", "b"], f"{NS}.TTLs": [1.0]}
+    ecs._post_dns(flat, NS)
+    assert "dns.question.class" not in flat
+    assert "dns.answers" not in flat  # answers and TTLs disagree in length
+    flat = {f"{NS}.answers": ["cname.example"], f"{NS}.TTLs": [5.0]}
+    ecs._post_dns(flat, NS)
+    assert flat["dns.answers"] == [{"data": "cname.example", "ttl": 5}]
+    assert "dns.resolved_ip" not in flat
+    assert flat["dns.type"] == "query"
+
+
+def test_http_without_action_or_status():
+    flat = {}
+    ecs._post_http(flat, NS)
+    assert flat == {}
+    flat = {"event.action": "GET", "http.response.status_code": "n/a"}
+    ecs._post_http(flat, NS)
+    assert flat["event.action"] == "get"
+    assert "event.outcome" not in flat
+
+
+@pytest.mark.parametrize(
+    ("version", "expected"),
+    [
+        ("SSLv3", ("3.0", "ssl")),
+        ("TLSv12", ("1.2", "tls")),
+        ("TLSv", None),
+        ("v12", None),
+        ("weird", None),
+        (12, None),
+    ],
+)
+def test_ssl_version_parsing(version, expected):
+    flat = {f"{NS}.version": version}
+    ecs._post_ssl(flat, NS)
+    if expected is None:
+        assert "tls.version" not in flat
+    else:
+        assert (flat["tls.version"], flat["tls.version_protocol"]) == expected
+
+
+def test_kerberos_without_validity_or_result():
+    flat = {f"{NS}.valid.from": 1.0}
+    ecs._post_kerberos(flat, NS)
+    assert f"{NS}.valid.days" not in flat
+    assert "event.outcome" not in flat
+
+
+def test_files_hosts_and_session_edge_cases():
+    flat = {"event.id": "C1", f"{NS}.session_ids": [], f"{NS}.tx_hosts": "10.0.0.6"}
+    ecs._post_files(flat, NS)
+    assert "server.ip" not in flat
+    flat = {"event.id": "C1", f"{NS}.session_ids": ["C9"], f"{NS}.rx_hosts": ["nope"]}
+    ecs._post_files(flat, NS)
+    assert flat["event.id"] == "C1"
+    assert flat[f"{NS}.rx_host"] == "nope"
+    assert "client.ip" not in flat
+    assert "related.ip" not in flat
+
+
+@pytest.mark.parametrize(
+    ("action", "expected"),
+    [
+        ("SMB::FILE_DELETE", "deletion"),
+        ("SMB::FILE_RENAME", "change"),
+        ("SMB::FILE_SET_ATTRIBUTE", "change"),
+        ("SMB::FILE_OPEN", "info"),
+        (None, None),
+    ],
+)
+def test_smb_files_actions(action, expected):
+    flat = {f"{NS}.path": "\\\\srv\\share"}  # path without a name: no file.path
+    if action:
+        flat[f"{NS}.action"] = action
+    ecs._post_smb_files(flat, NS)
+    assert "file.path" not in flat
+    assert "file.accessed" not in flat
+    assert flat.get("event.type", [None])[-1] == expected
+
+
+def test_missing_or_odd_status_fields_are_ignored():
+    for post in (ecs._post_smb_cmd, ecs._post_socks, ecs._post_sip, ecs._post_mysql, ecs._post_ssh):
+        flat = {}
+        post(flat, NS)
+        assert flat == {}
+    flat = {f"{NS}.status": 5, f"{NS}.status.code": "x", f"{NS}.cmd": 1}
+    for post in (ecs._post_smb_cmd, ecs._post_sip, ecs._post_mysql):
+        post(flat, NS)
+    assert "event.outcome" not in flat
+    assert "event.type" not in flat
+
+
+def test_sip_success_and_mysql_connect_out():
+    flat = {f"{NS}.status.code": 200}
+    ecs._post_sip(flat, NS)
+    assert flat["event.outcome"] == "success"
+    assert "event.type" not in flat
+    flat = {f"{NS}.cmd": "connect_out"}
+    ecs._post_mysql(flat, NS)
+    assert flat["event.type"] == ["access", "end"]
+
+
+def test_x509_without_algorithm_or_validity():
+    flat = {f"{NS}.certificate.signature_algorithm": None, f"{NS}.certificate.valid.from": None}
+    ecs._post_x509(flat, NS)
+    assert "file.x509.signature_algorithm" not in flat
+    assert "file.x509.not_before" not in flat
+
+
+def test_common_post_skips_community_id_it_cannot_compute():
+    flat = {"source.ip": "10.0.0.1", "destination.ip": "10.0.0.2", "network.transport": "bogus"}
+    ecs._common_post(flat)
+    assert "network.community_id" not in flat
+    assert flat["event.category"] == ["network"]
+
+
+def test_add_constants_applies_network_defaults():
+    flat = {"network.protocol": "dns"}
+    spec = {"event": {"type": ["info"]}, "network": {"protocol": "http", "transport": "udp"}}
+    ecs._add_constants(flat, None, "dns", spec)
+    assert flat["network.protocol"] == "dns"  # existing value wins
+    assert flat["network.transport"] == "udp"
+    assert "@timestamp" not in flat
+    assert flat["event.kind"] == "event"
+
+
+# -- coercion of arrays, unflatten, mapping -------------------------------------------------
+
+
+def test_empty_array_field_is_dropped():
+    assert ecs._coerce_value("dns.answers", []) is None
+    assert ecs._coerce_value("dns.answers", "a") == ["a"]
+
+
+def test_unflatten_dict_values_and_conflicts():
+    doc = unflatten({"a": {"x": 1}, "a.b": 2, "c.d": 3, "c": 4})
+    assert doc == {"a": {"x": 1, "b": 2}, "c": {"value": 4, "d": 3}}
+
+
+def test_index_body_handles_every_field_shape(monkeypatch):
+    monkeypatch.setattr(
+        ecs,
+        "FIELD_TYPES",
+        {
+            "a": "keyword",  # a scalar that is also a parent: the object wins
+            "a.b": "long",
+            "f": "flattened",
+            "f.x": "keyword",  # children of a flattened field are not mapped
+            "n": "nested",
+            "n.y": "ip",
+            "o.p": "keyword",
+        },
+    )
+    props = ecs_index_body()["mappings"]["properties"]
+    assert props["a"] == {"type": "object", "properties": {"b": {"type": "long"}}}
+    assert props["f"] == {"type": "flattened"}
+    assert props["n"] == {"type": "nested", "properties": {"y": {"type": "ip"}}}
+    assert props["o"] == {"type": "object", "properties": {"p": {"type": "keyword"}}}
+    assert props["@timestamp"] == {"type": "date"}