Skip to content

Commit 59f90f5

Browse files
committed
test: add round-trip and serialization tests
Add 32 tests covering to_dict(), Decimal field serialize/deserialize, and round-trip (from_dict -> to_dict) for all models. Documents serde behavior where None-valued optional fields are omitted on serialization. Closes #24
1 parent 7077cc0 commit 59f90f5

1 file changed

Lines changed: 397 additions & 0 deletions

File tree

tests/test_serialization.py

Lines changed: 397 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,397 @@
1+
"""
2+
Tests for serialization (to_dict) and round-trip (from_dict -> to_dict)
3+
behavior across all models.
4+
"""
5+
6+
import json
7+
from decimal import Decimal as PyDecimal
8+
from pathlib import Path
9+
10+
from l9format import (
11+
Certificate,
12+
DatasetSummary,
13+
GeoLocation,
14+
GeoPoint,
15+
L9Event,
16+
L9HttpEvent,
17+
L9LeakEvent,
18+
L9ServiceEvent,
19+
L9SSLEvent,
20+
Network,
21+
ServiceCredentials,
22+
Software,
23+
SoftwareModule,
24+
)
25+
from l9format.l9format import Decimal as DecimalField
26+
27+
TESTS_DIR = Path(__file__).parent
28+
29+
30+
class TestDecimalField:
31+
"""Test the custom Decimal field serialize/deserialize paths."""
32+
33+
def test_serialize_without_resolution(self) -> None:
34+
field = DecimalField()
35+
assert field.serialize(PyDecimal("1.5")) == "1.5"
36+
37+
def test_serialize_with_resolution(self) -> None:
38+
field = DecimalField(resolution=6)
39+
assert field.serialize(PyDecimal("1.5")) == "1.500000"
40+
41+
def test_serialize_rounds_to_resolution(self) -> None:
42+
field = DecimalField(resolution=2)
43+
assert field.serialize(PyDecimal("1.555")) == "1.56"
44+
45+
def test_deserialize_without_resolution(self) -> None:
46+
field = DecimalField()
47+
result = field.deserialize("1.5")
48+
assert result == PyDecimal("1.5")
49+
50+
def test_deserialize_with_resolution(self) -> None:
51+
field = DecimalField(resolution=6)
52+
result = field.deserialize("1.5")
53+
assert result == PyDecimal("1.500000")
54+
55+
def test_deserialize_from_int(self) -> None:
56+
field = DecimalField()
57+
result = field.deserialize(42)
58+
assert result == PyDecimal("42")
59+
60+
def test_deserialize_from_float(self) -> None:
61+
field = DecimalField()
62+
result = field.deserialize(1.5)
63+
assert result == PyDecimal("1.5")
64+
65+
def test_round_trip_with_resolution(self) -> None:
66+
field = DecimalField(resolution=6)
67+
original = "1.500000"
68+
deserialized = field.deserialize(original)
69+
serialized = field.serialize(deserialized)
70+
assert serialized == original
71+
72+
def test_round_trip_without_resolution(self) -> None:
73+
field = DecimalField()
74+
original = "3.14159"
75+
deserialized = field.deserialize(original)
76+
serialized = field.serialize(deserialized)
77+
assert serialized == original
78+
79+
80+
class TestGeoPointRoundTrip:
81+
"""Test GeoPoint serialization and round-trip."""
82+
83+
def test_to_dict(self) -> None:
84+
gp = GeoPoint.from_dict({"lat": "1.5", "lon": "2.5"})
85+
d = gp.to_dict()
86+
assert d["lat"] == "1.5"
87+
assert d["lon"] == "2.5"
88+
89+
def test_round_trip(self) -> None:
90+
data = {"lat": "48.8566", "lon": "2.3522"}
91+
assert GeoPoint.from_dict(data).to_dict() == data
92+
93+
def test_round_trip_zero(self) -> None:
94+
data = {"lat": "0", "lon": "0"}
95+
assert GeoPoint.from_dict(data).to_dict() == data
96+
97+
def test_round_trip_negative(self) -> None:
98+
data = {"lat": "-33.8688", "lon": "-151.2093"}
99+
assert GeoPoint.from_dict(data).to_dict() == data
100+
101+
102+
class TestNetworkRoundTrip:
103+
"""Test Network serialization and round-trip."""
104+
105+
def test_to_dict(self) -> None:
106+
net = Network.from_dict(
107+
{
108+
"organization_name": "Test Org",
109+
"asn": 12345,
110+
"network": "1.0.0.0/8",
111+
}
112+
)
113+
d = net.to_dict()
114+
assert d["organization_name"] == "Test Org"
115+
assert d["asn"] == 12345
116+
assert d["network"] == "1.0.0.0/8"
117+
118+
def test_round_trip(self) -> None:
119+
data = {
120+
"organization_name": "Test Org",
121+
"asn": 12345,
122+
"network": "1.0.0.0/8",
123+
}
124+
assert Network.from_dict(data).to_dict() == data
125+
126+
127+
class TestCertificateRoundTrip:
128+
"""Test Certificate serialization and round-trip."""
129+
130+
def test_to_dict(self) -> None:
131+
data = {
132+
"cn": "example.com",
133+
"domain": ["a.example.com", "b.example.com"],
134+
"fingerprint": "abc123",
135+
"key_algo": "RSA",
136+
"key_size": 2048,
137+
"issuer_name": "Test CA",
138+
"not_before": "2024-01-01T00:00:00+00:00",
139+
"not_after": "2024-12-31T23:59:59+00:00",
140+
"valid": True,
141+
}
142+
cert = Certificate.from_dict(data)
143+
d = cert.to_dict()
144+
assert d["cn"] == "example.com"
145+
assert d["domain"] == ["a.example.com", "b.example.com"]
146+
assert d["key_size"] == 2048
147+
assert d["valid"] is True
148+
149+
def test_round_trip(self) -> None:
150+
data = {
151+
"cn": "example.com",
152+
"domain": ["a.example.com"],
153+
"fingerprint": "abc123",
154+
"key_algo": "RSA",
155+
"key_size": 2048,
156+
"issuer_name": "Test CA",
157+
"not_before": "2024-01-01T00:00:00+00:00",
158+
"not_after": "2024-12-31T23:59:59+00:00",
159+
"valid": True,
160+
}
161+
assert Certificate.from_dict(data).to_dict() == data
162+
163+
def test_round_trip_null_domain_omitted(self) -> None:
164+
"""serde omits None-valued optional fields from to_dict()."""
165+
data = {
166+
"cn": "example.com",
167+
"domain": None,
168+
"fingerprint": "abc123",
169+
"key_algo": "RSA",
170+
"key_size": 2048,
171+
"issuer_name": "Test CA",
172+
"not_before": "2024-01-01T00:00:00+00:00",
173+
"not_after": "2024-12-31T23:59:59+00:00",
174+
"valid": True,
175+
}
176+
result = Certificate.from_dict(data).to_dict()
177+
assert "domain" not in result
178+
179+
180+
class TestHttpEventRoundTrip:
181+
"""Test L9HttpEvent serialization and round-trip."""
182+
183+
def test_round_trip(self) -> None:
184+
data = {
185+
"root": "/",
186+
"url": "/index.html",
187+
"status": 200,
188+
"length": 1024,
189+
"header": {"Content-Type": "text/html"},
190+
"title": "Welcome",
191+
"favicon_hash": "abc123",
192+
}
193+
assert L9HttpEvent.from_dict(data).to_dict() == data
194+
195+
def test_round_trip_null_header_omitted(self) -> None:
196+
"""serde omits None-valued optional fields from to_dict()."""
197+
data = {
198+
"root": "/",
199+
"url": "/",
200+
"status": 0,
201+
"length": 0,
202+
"header": None,
203+
"title": "",
204+
"favicon_hash": "",
205+
}
206+
result = L9HttpEvent.from_dict(data).to_dict()
207+
assert "header" not in result
208+
209+
210+
class TestSoftwareRoundTrip:
211+
"""Test Software and SoftwareModule serialization."""
212+
213+
def test_software_module_round_trip(self) -> None:
214+
data = {"name": "PHP", "version": "8.2.0", "fingerprint": "php-8-2-0"}
215+
assert SoftwareModule.from_dict(data).to_dict() == data
216+
217+
def test_software_with_modules_round_trip(self) -> None:
218+
data = {
219+
"name": "Apache",
220+
"version": "2.4.52",
221+
"os": "Ubuntu",
222+
"modules": [
223+
{"name": "PHP", "version": "8.2.0", "fingerprint": "php-8-2-0"}
224+
],
225+
"fingerprint": "apache-2-4-52",
226+
}
227+
assert Software.from_dict(data).to_dict() == data
228+
229+
def test_software_null_modules_omitted(self) -> None:
230+
"""serde omits None-valued optional fields from to_dict()."""
231+
data = {
232+
"name": "nginx",
233+
"version": "1.24.0",
234+
"os": "",
235+
"modules": None,
236+
"fingerprint": "nginx-1-24-0",
237+
}
238+
result = Software.from_dict(data).to_dict()
239+
assert "modules" not in result
240+
assert result["name"] == "nginx"
241+
242+
243+
class TestLeakEventRoundTrip:
244+
"""Test L9LeakEvent and DatasetSummary serialization."""
245+
246+
def test_dataset_summary_round_trip(self) -> None:
247+
data = {
248+
"rows": 100,
249+
"files": 5,
250+
"size": 65536,
251+
"collections": 3,
252+
"infected": False,
253+
"ransom_notes": ["Pay up"],
254+
}
255+
assert DatasetSummary.from_dict(data).to_dict() == data
256+
257+
def test_dataset_summary_null_ransom_notes_omitted(self) -> None:
258+
"""serde omits None-valued optional fields from to_dict()."""
259+
data = {
260+
"rows": 100,
261+
"files": 5,
262+
"size": 65536,
263+
"collections": 3,
264+
"infected": False,
265+
"ransom_notes": None,
266+
}
267+
result = DatasetSummary.from_dict(data).to_dict()
268+
assert "ransom_notes" not in result
269+
270+
def test_leak_event_round_trip(self) -> None:
271+
data = {
272+
"stage": "open",
273+
"type": "database",
274+
"severity": "high",
275+
"dataset": {
276+
"rows": 1000,
277+
"files": 0,
278+
"size": 1048576,
279+
"collections": 10,
280+
"infected": True,
281+
"ransom_notes": ["Pay up"],
282+
},
283+
}
284+
assert L9LeakEvent.from_dict(data).to_dict() == data
285+
286+
287+
class TestGeoLocationRoundTrip:
288+
"""Test GeoLocation serialization with nested GeoPoint."""
289+
290+
def test_round_trip_with_location(self) -> None:
291+
data = {
292+
"continent_name": "Europe",
293+
"region_iso_code": "FR-IDF",
294+
"city_name": "Paris",
295+
"country_iso_code": "FR",
296+
"country_name": "France",
297+
"region_name": "Ile-de-France",
298+
"location": {"lat": "48.8566", "lon": "2.3522"},
299+
}
300+
assert GeoLocation.from_dict(data).to_dict() == data
301+
302+
def test_round_trip_all_null_omitted(self) -> None:
303+
"""serde omits None-valued optional fields from to_dict()."""
304+
data = {
305+
"continent_name": None,
306+
"region_iso_code": None,
307+
"city_name": None,
308+
"country_iso_code": None,
309+
"country_name": None,
310+
"region_name": None,
311+
"location": None,
312+
}
313+
result = GeoLocation.from_dict(data).to_dict()
314+
assert len(result) == 0
315+
316+
317+
class TestServiceEventRoundTrip:
318+
"""Test L9ServiceEvent serialization."""
319+
320+
def test_round_trip(self) -> None:
321+
data = {
322+
"credentials": {
323+
"noauth": True,
324+
"username": "",
325+
"password": "",
326+
"key": "",
327+
"raw": "dGVzdA==",
328+
},
329+
"software": {
330+
"name": "Apache",
331+
"version": "2.4.52",
332+
"os": "Ubuntu",
333+
"modules": [
334+
{
335+
"name": "PHP",
336+
"version": "8.2.0",
337+
"fingerprint": "php-8-2-0",
338+
}
339+
],
340+
"fingerprint": "apache-2-4-52",
341+
},
342+
}
343+
assert L9ServiceEvent.from_dict(data).to_dict() == data
344+
345+
def test_round_trip_null_optionals_omitted(self) -> None:
346+
"""serde omits None-valued optional fields from to_dict()."""
347+
data = {
348+
"credentials": {
349+
"noauth": False,
350+
"username": "",
351+
"password": "",
352+
"key": "",
353+
"raw": None,
354+
},
355+
"software": {
356+
"name": "nginx",
357+
"version": "1.24.0",
358+
"os": "",
359+
"modules": None,
360+
"fingerprint": "nginx-1-24-0",
361+
},
362+
}
363+
result = L9ServiceEvent.from_dict(data).to_dict()
364+
assert "raw" not in result["credentials"]
365+
assert "modules" not in result["software"]
366+
367+
368+
class TestL9EventRoundTrip:
369+
"""Test full L9Event round-trip from reference JSON.
370+
371+
The reference JSON uses raw types (numeric lat/lon, Z-suffix
372+
datetimes) that differ from the serialized form (string decimals,
373+
+00:00 datetimes). A true round-trip asserts idempotency:
374+
serialize(deserialize(x)) produces a stable output that survives
375+
another cycle unchanged.
376+
"""
377+
378+
def test_round_trip_from_reference(self) -> None:
379+
path = TESTS_DIR / "l9event.json"
380+
with open(path) as f:
381+
data = json.load(f)
382+
first = L9Event.from_dict(data).to_dict()
383+
second = L9Event.from_dict(first).to_dict()
384+
assert first == second
385+
386+
def test_round_trip_ip4scout(self) -> None:
387+
ip4scout_files = [
388+
f
389+
for f in Path.iterdir(TESTS_DIR)
390+
if Path.is_file(f) and "ip4scout" in f.name
391+
]
392+
for path in ip4scout_files:
393+
with open(path) as f:
394+
data = json.load(f)
395+
first = L9Event.from_dict(data).to_dict()
396+
second = L9Event.from_dict(first).to_dict()
397+
assert first == second

0 commit comments

Comments
 (0)