cleanup: remove v6 garbage, restore original files, unify views.py workers
- Delete 15 garbage scripts (wnl_to_user_csv, eval_clustering, debug_db2/3, debug_persistence, test_orch/orch2, test_pca, test_profile, test_tool_*, test_e2e_full, test_32b, test_chinese_path, verify_runtime) - Delete data/wnl_converted.csv - Delete analysis/value_normalizer.py + tests/test_value_normalizer.py - Remove normalize_lf references from data_loader.py - Remove _norm column lookups from entity_aggregator.py - Fix settings.py: FILE_UPLOAD_MAX_MEMORY_SIZE 10MB -> 100MB - Fix upload.html: add per-file X delete button + removeFile() - Refactor views.py: unify _run_analysis_worker/_run_llm_analysis into _run_pipeline_worker with callback pattern - Wrap sync ORM call in tool_registry.py with sync_to_async - Update run_pipeline.py to use _run_pipeline_worker - Rebuild gen_test_data.py as simple version (replace 3-profile v6 version) - Add runtime/ db.sqlite3 logs/ .omo/ to .gitignore - Delete old .omo/plans/ plan files (tianxiu-v4/v5/v6, globe-v6) - 120/120 tests passing
This commit is contained in:
PM-pinou
@@ -1,537 +0,0 @@
|
||||
"""Tests for analysis.value_normalizer — TLS value normalisation.
|
||||
|
||||
Covers: TLS version (hex & display), cipher suite (hex & display),
|
||||
named curve (hex & display), CNRS/ISRS, and the top-level
|
||||
:func:`normalize_lf` integration.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import polars as pl
|
||||
|
||||
from analysis.value_normalizer import (
|
||||
TLS_VERSION_MAP,
|
||||
CIPHER_SUITE_MAP,
|
||||
CURVE_MAP,
|
||||
normalize_lf,
|
||||
)
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# TLS Version
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormTlsVersion:
|
||||
"""_norm_tls_version — hex & display → canonical enum name."""
|
||||
|
||||
def test_hex_03_04_tls_1_3(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 04']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_3'
|
||||
|
||||
def test_hex_03_03_tls_1_2(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 03']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_2'
|
||||
|
||||
def test_hex_03_02_tls_1_1(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 02']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_1'
|
||||
|
||||
def test_hex_03_01_tls_1_0(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 01']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_0'
|
||||
|
||||
def test_hex_02_00_ssl_2_0(self):
|
||||
lf = pl.LazyFrame({'0ver': ['02 00']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_2_0'
|
||||
|
||||
def test_hex_03_00_ssl_3_0(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 00']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_3_0'
|
||||
|
||||
def test_display_tlsv1_3(self):
|
||||
lf = pl.LazyFrame({'0ver': ['TLSv1.3']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_3'
|
||||
|
||||
def test_display_tlsv1_2(self):
|
||||
lf = pl.LazyFrame({'0ver': ['tlsv1.2']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_2'
|
||||
|
||||
def test_display_tls_1_1(self):
|
||||
lf = pl.LazyFrame({'0ver': ['TLS 1.1']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_1'
|
||||
|
||||
def test_display_bare_1_3(self):
|
||||
lf = pl.LazyFrame({'0ver': ['1.3']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_3'
|
||||
|
||||
def test_display_bare_1_2(self):
|
||||
lf = pl.LazyFrame({'0ver': ['1.2']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_2'
|
||||
|
||||
def test_display_tlsv1(self):
|
||||
lf = pl.LazyFrame({'0ver': ['TLSv1']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'tls_1_0'
|
||||
|
||||
def test_display_sslv2(self):
|
||||
lf = pl.LazyFrame({'0ver': ['SSLv2']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_2_0'
|
||||
|
||||
def test_display_sslv3(self):
|
||||
lf = pl.LazyFrame({'0ver': ['sslv3']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_3_0'
|
||||
|
||||
def test_display_bare_2(self):
|
||||
lf = pl.LazyFrame({'0ver': ['2']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_2_0'
|
||||
|
||||
def test_display_bare_3(self):
|
||||
lf = pl.LazyFrame({'0ver': ['3']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'ssl_3_0'
|
||||
|
||||
def test_case_insensitive(self):
|
||||
lf = pl.LazyFrame({'0ver': ['tlsv1.3', 'TLSv1.3', 'TlsV1.3']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'].to_list() == ['tls_1_3', 'tls_1_3', 'tls_1_3']
|
||||
|
||||
def test_unknown_version(self):
|
||||
lf = pl.LazyFrame({'0ver': ['ff ff']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'unknown_version'
|
||||
|
||||
def test_unknown_display(self):
|
||||
lf = pl.LazyFrame({'0ver': ['TLSv9.9']})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] == 'unknown_version'
|
||||
|
||||
def test_null_remains_null(self):
|
||||
lf = pl.LazyFrame({'0ver': [None]})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'][0] is None
|
||||
|
||||
def test_mixed_values(self):
|
||||
lf = pl.LazyFrame({'0ver': ['03 04', 'tlsv1.2', '1.3', 'ssl 2', None]})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert result['0ver_norm'].to_list() == [
|
||||
'tls_1_3', 'tls_1_2', 'tls_1_3', 'ssl_2_0', None,
|
||||
]
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# Cipher Suite — hex
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormCipher:
|
||||
"""_norm_cipher — hex cipher suite → canonical name."""
|
||||
|
||||
def test_hex_13_01(self):
|
||||
lf = pl.LazyFrame({'0cph': ['13 01']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_aes_128_gcm_sha256'
|
||||
|
||||
def test_hex_13_02(self):
|
||||
lf = pl.LazyFrame({'0cph': ['13 02']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_aes_256_gcm_sha384'
|
||||
|
||||
def test_hex_13_03(self):
|
||||
lf = pl.LazyFrame({'0cph': ['13 03']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_chacha20_poly1305_sha256'
|
||||
|
||||
def test_hex_c0_2b(self):
|
||||
lf = pl.LazyFrame({'0cph': ['c0 2b']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_ecdhe_ecdsa_aes_128_gcm_sha256'
|
||||
|
||||
def test_hex_00_35(self):
|
||||
lf = pl.LazyFrame({'0cph': ['00 35']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_rsa_aes_128_cbc_sha'
|
||||
|
||||
def test_hex_00_2f(self):
|
||||
"""00 2f also maps to tls_rsa_aes_128_cbc_sha (same canonical)."""
|
||||
lf = pl.LazyFrame({'0cph': ['00 2f']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_rsa_aes_128_cbc_sha'
|
||||
|
||||
def test_unknown_hex(self):
|
||||
lf = pl.LazyFrame({'0cph': ['ab cd']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'unknown_cipher_0xab cd'
|
||||
|
||||
def test_null_remains_null(self):
|
||||
lf = pl.LazyFrame({'0cph': [None]})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] is None
|
||||
|
||||
def test_case_insensitive_hex(self):
|
||||
lf = pl.LazyFrame({'0cph': ['C0 2B']})
|
||||
result = normalize_lf(lf, {'0cph': None}).collect()
|
||||
assert result['0cph_norm'][0] == 'tls_ecdhe_ecdsa_aes_128_gcm_sha256'
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# Cipher Suite — display name
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormCipherDisplay:
|
||||
"""_norm_cipher_display — IANA/display name → canonical name."""
|
||||
|
||||
def test_iana_full_name(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['TLS_AES_128_GCM_SHA256']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_aes_128_gcm_sha256'
|
||||
|
||||
def test_iana_full_name_lowercase(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['tls_aes_256_gcm_sha384']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_aes_256_gcm_sha384'
|
||||
|
||||
def test_display_aes128gcm(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['AES 128 GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_aes_128_gcm_sha256'
|
||||
|
||||
def test_display_aes256gcm(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['AES 256 GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_aes_256_gcm_sha384'
|
||||
|
||||
def test_display_chacha20(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['CHACHA20-POLY1305']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_chacha20_poly1305_sha256'
|
||||
|
||||
def test_display_ecdhe_ecdsa_aes128gcm(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['ECDHE-ECDSA-AES128-GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_ecdhe_ecdsa_aes_128_gcm_sha256'
|
||||
|
||||
def test_display_ecdhe_rsa_aes256gcm(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['ECDHE-RSA-AES256-GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_ecdhe_rsa_aes_256_gcm_sha384'
|
||||
|
||||
def test_display_rsa_aes128gcm(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['RSA AES 128 GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_rsa_aes_128_gcm_sha256'
|
||||
|
||||
def test_display_hyphen_variants(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['ECDHE-RSA-AES128-GCM']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'tls_ecdhe_rsa_aes_128_gcm_sha256'
|
||||
|
||||
def test_unknown_display(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': ['CUSTOM_CIPHER_999']})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] == 'customcipher999_unknown'
|
||||
|
||||
def test_null_remains_null(self):
|
||||
lf = pl.LazyFrame({'cipher-suite': [None]})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
assert result['cipher_suite_norm'][0] is None
|
||||
|
||||
def test_mixed_display_values(self):
|
||||
lf = pl.LazyFrame({
|
||||
'cipher-suite': [
|
||||
'TLS_AES_128_GCM_SHA256',
|
||||
'AES 256 GCM',
|
||||
'ECDHE-RSA-AES128-GCM',
|
||||
'bogus',
|
||||
None,
|
||||
],
|
||||
})
|
||||
result = normalize_lf(lf, {'cipher-suite': None}).collect()
|
||||
expected = [
|
||||
'tls_aes_128_gcm_sha256',
|
||||
'tls_aes_256_gcm_sha384',
|
||||
'tls_ecdhe_rsa_aes_128_gcm_sha256',
|
||||
'bogus_unknown',
|
||||
None,
|
||||
]
|
||||
assert result['cipher_suite_norm'].to_list() == expected
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# Named Curve — hex
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormCurve:
|
||||
"""_norm_curve — hex curve → canonical name."""
|
||||
|
||||
def test_hex_00_1d_x25519(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 1d']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'x25519'
|
||||
|
||||
def test_hex_00_17_secp256r1(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 17']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_hex_00_18_secp384r1(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 18']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'secp384r1'
|
||||
|
||||
def test_hex_00_19_secp521r1(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 19']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'secp521r1'
|
||||
|
||||
def test_hex_00_1e_x448(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 1e']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'x448'
|
||||
|
||||
def test_unknown_hex_curve(self):
|
||||
lf = pl.LazyFrame({'0crv': ['ff ff']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'unknown_curve_0xff ff'
|
||||
|
||||
def test_null_remains_null(self):
|
||||
lf = pl.LazyFrame({'0crv': [None]})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] is None
|
||||
|
||||
def test_case_insensitive_hex_curve(self):
|
||||
lf = pl.LazyFrame({'0crv': ['00 1D']})
|
||||
result = normalize_lf(lf, {'0crv': None}).collect()
|
||||
assert result['0crv_norm'][0] == 'x25519'
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# Named Curve — display name
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormCurveDisplay:
|
||||
"""_norm_curve_display — display name → canonical name."""
|
||||
|
||||
def test_x25519(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['X25519']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'x25519'
|
||||
|
||||
def test_curve25519(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['Curve25519']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'x25519'
|
||||
|
||||
def test_secp256r1(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['secp256r1']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_prime256v1(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['prime256v1']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_p256(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P256']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_p_256(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P-256']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_p_256_underscore(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P_256']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp256r1'
|
||||
|
||||
def test_secp384r1(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['secp384r1']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp384r1'
|
||||
|
||||
def test_p384(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P384']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp384r1'
|
||||
|
||||
def test_p_384(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['p-384']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp384r1'
|
||||
|
||||
def test_secp521r1(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['Secp521r1']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp521r1'
|
||||
|
||||
def test_p521(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P521']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'secp521r1'
|
||||
|
||||
def test_x448(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['x448']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'x448'
|
||||
|
||||
def test_curve448(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['Curve448']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'x448'
|
||||
|
||||
def test_unknown_curve_display(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['brainpoolP256r1']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'unknown_curve_brainpoolp256r1'
|
||||
|
||||
def test_null_remains_null(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': [None]})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'][0] is None
|
||||
|
||||
def test_case_insensitive_p256(self):
|
||||
lf = pl.LazyFrame({'ecdhe-named-curve': ['P-256', 'p_256', 'P256']})
|
||||
result = normalize_lf(lf, {'ecdhe-named-curve': None}).collect()
|
||||
assert result['ecdhe_named_curve_norm'].to_list() == [
|
||||
'secp256r1', 'secp256r1', 'secp256r1',
|
||||
]
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# CNRS / ISRS
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormCnrs:
|
||||
"""_norm_cnrs — '+' → 'yes', blank/null → 'no'."""
|
||||
|
||||
def test_plus_yes(self):
|
||||
lf = pl.LazyFrame({'cnrs': ['+']})
|
||||
result = normalize_lf(lf, {'cnrs': None}).collect()
|
||||
assert result['cnrs_norm'][0] == 'yes'
|
||||
|
||||
def test_blank_no(self):
|
||||
lf = pl.LazyFrame({'cnrs': ['']})
|
||||
result = normalize_lf(lf, {'cnrs': None}).collect()
|
||||
assert result['cnrs_norm'][0] == 'no'
|
||||
|
||||
def test_null_no(self):
|
||||
lf = pl.LazyFrame({'cnrs': [None]})
|
||||
result = normalize_lf(lf, {'cnrs': None}).collect()
|
||||
assert result['cnrs_norm'][0] == 'no'
|
||||
|
||||
def test_whitespace_plus(self):
|
||||
lf = pl.LazyFrame({'cnrs': [' + ']})
|
||||
result = normalize_lf(lf, {'cnrs': None}).collect()
|
||||
assert result['cnrs_norm'][0] == 'yes'
|
||||
|
||||
def test_anything_else_no(self):
|
||||
lf = pl.LazyFrame({'cnrs': ['-', 'true', 'maybe', '0', '1']})
|
||||
result = normalize_lf(lf, {'cnrs': None}).collect()
|
||||
assert result['cnrs_norm'].to_list() == ['no', 'no', 'no', 'no', 'no']
|
||||
|
||||
def test_isrs_plus_yes(self):
|
||||
lf = pl.LazyFrame({'isrs': ['+']})
|
||||
result = normalize_lf(lf, {'isrs': None}).collect()
|
||||
assert result['isrs_norm'][0] == 'yes'
|
||||
|
||||
def test_isrs_blank_no(self):
|
||||
lf = pl.LazyFrame({'isrs': ['']})
|
||||
result = normalize_lf(lf, {'isrs': None}).collect()
|
||||
assert result['isrs_norm'][0] == 'no'
|
||||
|
||||
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
# Integration — normalize_lf
|
||||
# ═════════════════════════════════════════════════════════════════════════
|
||||
|
||||
class TestNormalizeLf:
|
||||
"""Top-level normalize_lf — column routing & no-op cases."""
|
||||
|
||||
def test_all_columns_at_once(self):
|
||||
lf = pl.LazyFrame({
|
||||
'0ver': ['03 04'],
|
||||
'0cph': ['13 01'],
|
||||
'cipher-suite': ['TLS_AES_128_GCM_SHA256'],
|
||||
'0crv': ['00 1d'],
|
||||
'ecdhe-named-curve': ['X25519'],
|
||||
'cnrs': ['+'],
|
||||
'isrs': [''],
|
||||
})
|
||||
type_map = {
|
||||
'0ver': None,
|
||||
'0cph': None,
|
||||
'cipher-suite': None,
|
||||
'0crv': None,
|
||||
'ecdhe-named-curve': None,
|
||||
'cnrs': None,
|
||||
'isrs': None,
|
||||
}
|
||||
result = normalize_lf(lf, type_map).collect()
|
||||
|
||||
assert result['0ver_norm'][0] == 'tls_1_3'
|
||||
assert result['0cph_norm'][0] == 'tls_aes_128_gcm_sha256'
|
||||
assert result['cipher_suite_norm'][0] == 'tls_aes_128_gcm_sha256'
|
||||
assert result['0crv_norm'][0] == 'x25519'
|
||||
assert result['ecdhe_named_curve_norm'][0] == 'x25519'
|
||||
assert result['cnrs_norm'][0] == 'yes'
|
||||
assert result['isrs_norm'][0] == 'no'
|
||||
|
||||
def test_no_matching_columns(self):
|
||||
"""Empty type_map → no _norm columns added."""
|
||||
lf = pl.LazyFrame({'0ver': ['03 04']})
|
||||
result = normalize_lf(lf, {}).collect()
|
||||
assert '0ver_norm' not in result.columns
|
||||
assert result['0ver'][0] == '03 04'
|
||||
|
||||
def test_subset_of_columns(self):
|
||||
"""Only columns present in type_map get normalised."""
|
||||
lf = pl.LazyFrame({
|
||||
'0ver': ['03 04'],
|
||||
'0cph': ['13 01'],
|
||||
'cnrs': ['+'],
|
||||
})
|
||||
result = normalize_lf(lf, {'0ver': None, 'cnrs': None}).collect()
|
||||
assert '0ver_norm' in result.columns
|
||||
assert 'cnrs_norm' in result.columns
|
||||
assert '0cph_norm' not in result.columns
|
||||
|
||||
def test_preserves_original_columns(self):
|
||||
"""Original columns are kept unchanged."""
|
||||
lf = pl.LazyFrame({
|
||||
'0ver': ['03 04'],
|
||||
'src_ip': ['10.0.0.1'],
|
||||
})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
assert '0ver' in result.columns
|
||||
assert 'src_ip' in result.columns
|
||||
assert result['src_ip'][0] == '10.0.0.1'
|
||||
|
||||
def test_duplicate_normalized_column_exists(self):
|
||||
"""If a _norm column already exists, Polars appends suffixed."""
|
||||
lf = pl.LazyFrame({
|
||||
'0ver': ['03 04'],
|
||||
'0ver_norm': ['old'],
|
||||
})
|
||||
result = normalize_lf(lf, {'0ver': None}).collect()
|
||||
# Polars will create 0ver_norm with new value
|
||||
assert result['0ver_norm'][0] == 'tls_1_3'
|
||||
|
||||
def test_lazyness_preserved(self):
|
||||
"""normalize_lf returns a LazyFrame, not materialised."""
|
||||
lf = pl.LazyFrame({'0ver': ['03 04']})
|
||||
result = normalize_lf(lf, {'0ver': None})
|
||||
assert isinstance(result, pl.LazyFrame)
|
||||
Reference in New Issue
Block a user