From 2ba634757d66540c18c4724af75caf752f1feb68 Mon Sep 17 00:00:00 2001
From: Pipeline Agent <pipeline@opencomputing.cn>
Date: Sat, 15 Aug 2026 12:21:07 +0800
Subject: [PATCH] =?UTF-8?q?develop:=20=E9=87=8D=E5=81=9A=20fault-log-analy?=
 =?UTF-8?q?zer=20=E5=88=86=E6=9E=90=E6=A8=A1=E5=9D=97=EF=BC=88=E9=81=B5?=
 =?UTF-8?q?=E5=AE=88=E6=A8=A1=E5=9D=97=E5=BC=80=E5=8F=91=E8=A7=84=E8=8C=83?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/fault_log_analyzer/models.py |  18 ++++-
 src/fault_log_analyzer/parser.py |  21 +++---
 tests/test_api.py                | 125 ++++++++++++++++++++++++++++++-
 tests/test_classifier.py         |  67 ++++++++++++++++-
 tests/test_cluster.py            |  93 ++++++++++++++++++++++-
 tests/test_fingerprint.py        |  85 ++++++++++++++++++++-
 tests/test_integrations.py       |  60 ++++++++++++++-
 tests/test_pipeline.py           |  69 ++++++++++++++++-
 tests/test_root_cause.py         |  92 ++++++++++++++++++++++-
 tests/test_storage.py            |  94 ++++++++++++++++++++++-
 tests/test_workers.py            |  89 +++++++++++++++++++++-
 11 files changed, 792 insertions(+), 21 deletions(-)

diff --git a/src/fault_log_analyzer/models.py b/src/fault_log_analyzer/models.py
index 974077f..e134663 100644
--- a/src/fault_log_analyzer/models.py
+++ b/src/fault_log_analyzer/models.py
@@ -17,7 +17,11 @@ def _now() -> datetime:
 
 
 def _to_dt(value: Any) -> datetime:
-    """将字符串/时间戳转换为带时区的 datetime。"""
+    """将字符串/时间戳转换为带时区的 datetime。
+
+    支持 ISO8601（带 Z 或空格分隔）、Unix 时间戳以及 syslog 风格
+    ``Jan  1 10:18:00``（无年份，缺省使用当前年份）。
+    """
     if value is None:
         return _now()
     if isinstance(value, datetime):
@@ -26,13 +30,23 @@ def _to_dt(value: Any) -> datetime:
         return value
     if isinstance(value, (int, float)):
         return datetime.fromtimestamp(value, tz=timezone.utc)
+
     text = str(value).strip()
     if text.endswith("Z"):
         text = text[:-1] + "+00:00"
+
+    dt: Optional[datetime] = None
     try:
         dt = datetime.fromisoformat(text)
     except ValueError:
-        dt = datetime.fromisoformat(text.replace(" ", "T"))
+        try:
+            dt = datetime.fromisoformat(text.replace(" ", "T"))
+        except ValueError:
+            # syslog 风格：Jan  1 10:18:00（无年份，缺省当前年份）
+            normalized = " ".join(text.split())
+            dt = datetime.strptime(normalized, "%b %d %H:%M:%S")
+            dt = dt.replace(year=datetime.now().year)
+
     if dt.tzinfo is None:
         dt = dt.replace(tzinfo=timezone.utc)
     return dt
diff --git a/src/fault_log_analyzer/parser.py b/src/fault_log_analyzer/parser.py
index 59f6d7a..a4d7003 100644
--- a/src/fault_log_analyzer/parser.py
+++ b/src/fault_log_analyzer/parser.py
@@ -21,13 +21,14 @@ from .models import ParsedLog, _to_dt
 _LEVELS = ("FATAL", "ERROR", "WARN", "WARNING", "INFO", "DEBUG", "TRACE")
 
 # log4j / 通用时间前缀：2025-01-01 10:18:00,123 或 2025-01-01T10:18:00Z
-_TS_RE = re.compile(
-    r"^(?P<ts>\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d{3})?(?:Z|[+-]\d{2}:?\d{2})?)"
-)
+_TS_PAT = r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d{3})?(?:Z|[+-]\d{2}:?\d{2})?"
 
 _LOG4J_RE = re.compile(
-    r"^\s*(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d{3})?(?:Z|[+-]\d{2}:?\d{2})?)\s+"
-    r"([A-Za-z]+)\s+\[?([^\]]*?)\]?\s*(?:-)?\s*(.*)$"
+    r"^\s*(?P<ts>" + _TS_PAT + r")\s+"
+    r"(?P<level>[A-Za-z]+)\s+"
+    r"(?:\[(?P<thread>[^\]]*)\]\s*)?"
+    r"(?:-\s*)?"
+    r"(?P<message>.*)$"
 )
 
 _SYSLOG_RE = re.compile(
@@ -78,15 +79,15 @@ class LogParser:
         # 2) log4j 风格
         m = _LOG4J_RE.match(stripped)
         if m:
-            ts_text, level, thread, message = m.groups()
-            level = level.upper()
-            thread = (thread or "").strip()
+            level = m.group("level").upper()
+            thread = (m.group("thread") or "").strip()
+            message = (m.group("message") or "").strip()
             return ParsedLog(
-                timestamp=_to_dt(ts_text),
+                timestamp=_to_dt(m.group("ts")),
                 host_id=self._extract_host(message) or thread,
                 service=thread,
                 level=level,
-                message=message.strip(),
+                message=message,
                 trace_id=self._extract_trace(message),
                 raw=line,
             )
diff --git a/tests/test_api.py b/tests/test_api.py
index 2e7c81a..42b59ae 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -1 +1,124 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""REST API 测试（标准库 http.server）。"""
+import _bootstrap  # noqa: F401
+import json
+import threading
+import unittest
+from http.server import ThreadingHTTPServer
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+from fault_log_analyzer.api import create_handler
+from fault_log_analyzer.models import FaultLog, RootCause
+from fault_log_analyzer.storage import MemoryStorage
+
+
+class ApiTestBase(unittest.TestCase):
+    def setUp(self):
+        self.storage = MemoryStorage()
+        handler = create_handler(self.storage)
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
+        self.base = f"http://127.0.0.1:{self.server.server_address[1]}"
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=2)
+
+    def request(self, method, path, body=None):
+        url = self.base + path
+        data = None
+        headers = {}
+        if body is not None:
+            data = json.dumps(body).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        req = Request(url, data=data, method=method, headers=headers)
+        try:
+            resp = urlopen(req, timeout=5)
+            return resp.status, json.loads(resp.read().decode("utf-8"))
+        except HTTPError as e:
+            return e.code, json.loads(e.read().decode("utf-8"))
+
+
+class TestHealth(ApiTestBase):
+    def test_healthz(self):
+        status, body = self.request("GET", "/healthz")
+        self.assertEqual(status, 200)
+        self.assertEqual(body["status"], "ok")
+
+
+class TestFaultLogs(ApiTestBase):
+    def test_list_empty(self):
+        status, body = self.request("GET", "/api/v1/fault-logs")
+        self.assertEqual(status, 200)
+        self.assertEqual(body["data"]["total"], 0)
+
+    def test_list_and_detail(self):
+        self.storage.save_fault_log(
+            FaultLog(fault_log_id="fl-1", host_id="h-1", message="no space", level="ERROR", fault_type="disk_full")
+        )
+        status, body = self.request("GET", "/api/v1/fault-logs")
+        self.assertEqual(body["data"]["total"], 1)
+        status, body = self.request("GET", "/api/v1/fault-logs/fl-1")
+        self.assertEqual(body["data"]["fault_log_id"], "fl-1")
+
+    def test_detail_not_found(self):
+        status, body = self.request("GET", "/api/v1/fault-logs/nope")
+        self.assertEqual(status, 404)
+        self.assertEqual(body["code"], 40401)
+
+    def test_invalid_page_param(self):
+        status, body = self.request("GET", "/api/v1/fault-logs?page=abc")
+        self.assertEqual(status, 400)
+        self.assertEqual(body["code"], 40001)
+
+    def test_root_cause(self):
+        self.storage.save_fault_log(FaultLog(fault_log_id="fl-1", host_id="h-1", message="x"))
+        self.storage.save_root_cause(RootCause(fault_log_id="fl-1", cause_type="disk_full"))
+        status, body = self.request("GET", "/api/v1/fault-logs/fl-1/root-cause")
+        self.assertEqual(body["data"]["cause_type"], "disk_full")
+
+    def test_root_cause_not_found(self):
+        status, body = self.request("GET", "/api/v1/fault-logs/nope/root-cause")
+        self.assertEqual(status, 404)
+
+
+class TestFaultTypes(ApiTestBase):
+    def test_list_and_create(self):
+        status, body = self.request("GET", "/api/v1/fault-types")
+        self.assertEqual(body["data"]["items"], [])
+        status, body = self.request("POST", "/api/v1/fault-types", {"fault_type": "x", "name": "X"})
+        self.assertEqual(status, 201)
+        self.assertEqual(body["data"]["fault_type"], "x")
+        status, body = self.request("GET", "/api/v1/fault-types")
+        self.assertEqual(len(body["data"]["items"]), 1)
+
+    def test_create_missing_fields(self):
+        status, body = self.request("POST", "/api/v1/fault-types", {"name": "X"})
+        self.assertEqual(status, 400)
+        self.assertEqual(body["code"], 40002)
+
+
+class TestFaultFilters(ApiTestBase):
+    def test_list_and_create(self):
+        status, body = self.request("GET", "/api/v1/fault-filters")
+        self.assertEqual(body["data"]["items"], [])
+        status, body = self.request("POST", "/api/v1/fault-filters", {"name": "r1", "level": "ERROR"})
+        self.assertEqual(status, 201)
+        self.assertEqual(body["data"]["name"], "r1")
+
+    def test_create_missing_name(self):
+        status, body = self.request("POST", "/api/v1/fault-filters", {"level": "ERROR"})
+        self.assertEqual(status, 400)
+        self.assertEqual(body["code"], 40002)
+
+
+class TestNotFound(ApiTestBase):
+    def test_unknown_route(self):
+        status, body = self.request("GET", "/api/v1/unknown")
+        self.assertEqual(status, 404)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_classifier.py b/tests/test_classifier.py
index 2e7c81a..f94da7c 100644
--- a/tests/test_classifier.py
+++ b/tests/test_classifier.py
@@ -1 +1,66 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""簇 -> fault_type 归类测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+
+from fault_log_analyzer.classifier import DEFAULT_FAULT_TYPES, Classifier
+from fault_log_analyzer.models import FaultType
+
+
+class TestDefaultClassifier(unittest.TestCase):
+    def test_pattern_disk(self):
+        self.assertEqual(Classifier().classify("No space left on device"), "disk_full")
+
+    def test_pattern_memory(self):
+        self.assertEqual(
+            Classifier().classify("out of memory: process killed"), "memory_exhausted"
+        )
+
+    def test_pattern_network(self):
+        self.assertEqual(Classifier().classify("connection refused"), "network_unreachable")
+
+    def test_hint_timeout(self):
+        self.assertEqual(Classifier().classify("request timed out"), "timeout")
+
+    def test_hint_permission(self):
+        self.assertEqual(Classifier().classify("permission denied"), "permission_denied")
+
+    def test_hint_unknown(self):
+        self.assertIsNone(Classifier().classify("some ordinary message"))
+
+
+class TestClusterMapping(unittest.TestCase):
+    def test_registered_cluster_takes_priority(self):
+        c = Classifier()
+        c.register_cluster(1, "disk_full")
+        self.assertEqual(c.classify("nothing relevant", cluster_id=1), "disk_full")
+
+    def test_classify_cluster_sets_map(self):
+        c = Classifier()
+        self.assertEqual(c.classify_cluster(3, ["no space left on device"]), "disk_full")
+        self.assertEqual(c.classify("whatever", cluster_id=3), "disk_full")
+
+
+class TestCustomFaultTypes(unittest.TestCase):
+    def test_disabled_type_skipped(self):
+        c = Classifier([FaultType(fault_type="custom", name="C", pattern="custom", enabled=False)])
+        self.assertIsNone(c.classify("a custom message that is not a known hint"))
+
+    def test_custom_pattern_matches(self):
+        c = Classifier([FaultType(fault_type="custom", name="C", pattern="custom error")])
+        self.assertEqual(c.classify("a custom error happened"), "custom")
+
+    def test_invalid_pattern_regex_ignored(self):
+        c = Classifier([FaultType(fault_type="bad", name="B", pattern="(")])
+        self.assertIsNone(c.classify("anything"))
+
+
+class TestDefaultFaultTypes(unittest.TestCase):
+    def test_has_expected_types(self):
+        names = {ft.fault_type for ft in DEFAULT_FAULT_TYPES}
+        self.assertIn("disk_full", names)
+        self.assertIn("memory_exhausted", names)
+        self.assertIn("network_unreachable", names)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 2e7c81a..225257d 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -1 +1,92 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""TF-IDF 特征 + DBSCAN 聚类测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+
+from fault_log_analyzer.cluster import (
+    Clusterer,
+    cosine_distance,
+    cosine_similarity,
+    dbscan,
+    tfidf_vectors,
+)
+
+
+class TestTfidf(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual(tfidf_vectors([]), [])
+
+    def test_vectors_shape(self):
+        vecs = tfidf_vectors(["no space left on device", "out of memory error"])
+        self.assertEqual(len(vecs), 2)
+
+    def test_identical_docs_similar(self):
+        vecs = tfidf_vectors(["disk full", "disk full"])
+        self.assertAlmostEqual(cosine_similarity(vecs[0], vecs[1]), 1.0)
+
+    def test_distinct_docs_less_similar(self):
+        vecs = tfidf_vectors(["disk full error", "memory exhausted"])
+        self.assertLess(cosine_similarity(vecs[0], vecs[1]), 1.0)
+
+
+class TestCosine(unittest.TestCase):
+    def test_similarity_zero_for_empty(self):
+        self.assertEqual(cosine_similarity({}, {}), 0.0)
+
+    def test_distance_identical_is_zero(self):
+        self.assertAlmostEqual(cosine_distance({"a": 1.0}, {"a": 1.0}), 0.0)
+
+    def test_distance_is_one_for_orthogonal(self):
+        self.assertAlmostEqual(cosine_distance({"a": 1.0}, {"b": 1.0}), 1.0)
+
+
+class TestDbscan(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual(dbscan([], 0.75, 5), [])
+
+    def test_single_sample_is_noise(self):
+        self.assertEqual(dbscan([{"a": 1.0}], 0.75, 5), [-1])
+
+    def test_two_identical_with_min2(self):
+        labels = dbscan([{"a": 1.0}, {"a": 1.0}], 0.75, 2)
+        self.assertEqual(labels, [0, 0])
+
+    def test_two_distinct_with_min2_are_noise(self):
+        labels = dbscan([{"a": 1.0}, {"b": 1.0}], 0.75, 2)
+        self.assertEqual(labels, [-1, -1])
+
+
+class TestClusterer(unittest.TestCase):
+    def test_fit_predict_small_returns_noise(self):
+        c = Clusterer(eps=0.75, min_samples=5)
+        self.assertEqual(c.fit_predict(["only one message"]), [-1])
+
+    def test_assign_empty_centroids(self):
+        self.assertEqual(Clusterer().assign("anything"), -1)
+
+    def test_fit_predict_group(self):
+        messages = [
+            "no space left on device",
+            "no space left on device /var",
+            "no space left on device /tmp",
+            "out of memory error",
+            "out of memory killed process",
+        ]
+        c = Clusterer(eps=0.75, min_samples=2)
+        labels = c.fit_predict(messages)
+        self.assertEqual(len(labels), 5)
+        self.assertGreaterEqual(c.n_clusters, 1)
+
+    def test_assign_to_nearest_cluster(self):
+        messages = [
+            "no space left on device",
+            "no space left on device /var",
+            "no space left on device /tmp",
+        ]
+        c = Clusterer(eps=0.75, min_samples=2)
+        c.fit_predict(messages)
+        self.assertGreaterEqual(c.n_clusters, 1)
+        self.assertGreaterEqual(c.assign("no space left on device /opt"), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py
index 2e7c81a..435fc29 100644
--- a/tests/test_fingerprint.py
+++ b/tests/test_fingerprint.py
@@ -1 +1,84 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""MinHash 指纹与去重测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+
+from fault_log_analyzer.fingerprint import Deduplicator, MinHash, shingles
+
+
+class TestShingles(unittest.TestCase):
+    def test_basic(self):
+        self.assertEqual(shingles("abcde"), {"abc", "bcd", "cde"})
+
+    def test_short_text(self):
+        self.assertEqual(shingles("ab"), {"ab"})
+
+    def test_empty(self):
+        self.assertEqual(shingles(""), set())
+
+    def test_whitespace_collapsed(self):
+        self.assertEqual(shingles("a b"), {"a b"})
+
+
+class TestMinHash(unittest.TestCase):
+    def test_signature_empty(self):
+        mh = MinHash(num_perm=8)
+        self.assertEqual(mh.signature([]), [0] * 8)
+
+    def test_fingerprint_deterministic(self):
+        mh = MinHash()
+        self.assertEqual(
+            mh.fingerprint("no space left on device"),
+            mh.fingerprint("no space left on device"),
+        )
+
+    def test_fingerprint_length(self):
+        self.assertEqual(len(MinHash().fingerprint("hello world")), 16)
+
+    def test_fingerprint_empty(self):
+        self.assertEqual(MinHash().fingerprint(""), "0" * 16)
+
+    def test_fingerprint_is_hex(self):
+        self.assertRegex(MinHash().fingerprint("hello world"), r"^[0-9a-f]{16}$")
+
+    def test_similar_text_share_signature(self):
+        mh = MinHash(num_perm=16)
+        sig1 = mh.signature(shingles("disk full on /dev/sda1"))
+        sig2 = mh.signature(shingles("disk full on /dev/sda1"))
+        self.assertEqual(sig1, sig2)
+
+
+class TestDeduplicator(unittest.TestCase):
+    def _clock(self):
+        state = {"t": 100.0}
+        return lambda: state["t"], state
+
+    def test_first_not_duplicate(self):
+        clock, state = self._clock()
+        d = Deduplicator(window_seconds=300, clock=clock)
+        self.assertFalse(d.is_duplicate("fp1"))
+
+    def test_duplicate_within_window(self):
+        clock, state = self._clock()
+        d = Deduplicator(window_seconds=300, clock=clock)
+        d.is_duplicate("fp1")
+        state["t"] = 200.0
+        self.assertTrue(d.is_duplicate("fp1"))
+
+    def test_expired_after_window(self):
+        clock, state = self._clock()
+        d = Deduplicator(window_seconds=300, clock=clock)
+        d.is_duplicate("fp1")
+        state["t"] = 500.0
+        self.assertFalse(d.is_duplicate("fp1"))
+
+    def test_mark_and_clear(self):
+        clock, state = self._clock()
+        d = Deduplicator(window_seconds=300, clock=clock)
+        d.mark("fp1")
+        self.assertTrue(d.is_duplicate("fp1"))
+        d.clear()
+        self.assertFalse(d.is_duplicate("fp1"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_integrations.py b/tests/test_integrations.py
index 2e7c81a..c5d60f6 100644
--- a/tests/test_integrations.py
+++ b/tests/test_integrations.py
@@ -1 +1,59 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""可选真实后端适配器测试（Kafka/ES/MySQL/Redis）。"""
+import _bootstrap  # noqa: F401
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+from fault_log_analyzer.integrations import (
+    ElasticsearchSink,
+    KafkaConsumerAdapter,
+    KafkaProducerAdapter,
+    MySQLFaultRepository,
+    RedisDedupCache,
+)
+
+
+class TestAdaptersRequireDependencies(unittest.TestCase):
+    """未安装对应 SDK 时，实例化应抛出明确 RuntimeError。"""
+
+    def test_kafka_producer_requires_kafka(self):
+        with patch.dict("sys.modules", {"kafka": None}):
+            with self.assertRaises(RuntimeError):
+                KafkaProducerAdapter("localhost:9092")
+
+    def test_kafka_consumer_requires_kafka(self):
+        with patch.dict("sys.modules", {"kafka": None}):
+            with self.assertRaises(RuntimeError):
+                KafkaConsumerAdapter("localhost:9092", "g", "logs.raw")
+
+    def test_elasticsearch_requires_es(self):
+        with patch.dict("sys.modules", {"elasticsearch": None}):
+            with self.assertRaises(RuntimeError):
+                ElasticsearchSink("http://localhost:9200")
+
+    def test_mysql_requires_pymysql(self):
+        with patch.dict("sys.modules", {"pymysql": None}):
+            with self.assertRaises(RuntimeError):
+                MySQLFaultRepository("localhost", 3306, "u", "p", "db")
+
+    def test_redis_requires_redis(self):
+        with patch.dict("sys.modules", {"redis": None}):
+            with self.assertRaises(RuntimeError):
+                RedisDedupCache("redis://localhost:6379/0")
+
+
+class TestElasticsearchIndexName(unittest.TestCase):
+    def test_index_name_monthly(self):
+        dt = datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc)
+        self.assertEqual(
+            ElasticsearchSink._index_name("hms-fault-log", dt), "hms-fault-log-2025.01"
+        )
+
+    def test_index_name_default_now(self):
+        name = ElasticsearchSink._index_name("hms-fault-log", None)
+        self.assertTrue(name.startswith("hms-fault-log-"))
+        self.assertEqual(len(name), len("hms-fault-log-") + 7)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 2e7c81a..a1f169e 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -1 +1,68 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""故障日志捕获管道测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+
+from fault_log_analyzer.filters import FaultFilter
+from fault_log_analyzer.models import FaultFilterRule
+from fault_log_analyzer.pipeline import CapturePipeline
+from fault_log_analyzer.storage import MemoryStorage
+
+
+class TestCapturePipeline(unittest.TestCase):
+    def setUp(self):
+        self.storage = MemoryStorage()
+        self.pipeline = CapturePipeline(storage=self.storage)
+
+    def test_error_captured_and_classified(self):
+        log = self.pipeline.process(
+            {"level": "ERROR", "message": "disk full", "host_id": "h-1"}
+        )
+        self.assertIsNotNone(log)
+        self.assertEqual(log.fault_type, "disk_full")
+        self.assertEqual(log.host_id, "h-1")
+        self.assertEqual(self.storage.query_fault_logs()["total"], 1)
+
+    def test_info_skipped(self):
+        self.assertIsNone(self.pipeline.process({"level": "INFO", "message": "hello"}))
+        self.assertEqual(self.storage.query_fault_logs()["total"], 0)
+
+    def test_duplicate_skipped(self):
+        self.pipeline.process({"level": "ERROR", "message": "disk full"})
+        second = self.pipeline.process({"level": "ERROR", "message": "disk full"})
+        self.assertIsNone(second)
+        self.assertEqual(self.pipeline.duplicates, 1)
+        self.assertEqual(self.pipeline.captured, 1)
+
+    def test_producer_callback(self):
+        produced = []
+        pipeline = CapturePipeline(storage=MemoryStorage(), producer=produced.append)
+        log = pipeline.process({"level": "ERROR", "message": "boom"})
+        self.assertEqual(produced, [log])
+
+    def test_process_batch(self):
+        logs = self.pipeline.process_batch(
+            [
+                {"level": "ERROR", "message": "boom"},
+                {"level": "INFO", "message": "skip"},
+                {"level": "FATAL", "message": "fatal thing"},
+            ]
+        )
+        self.assertEqual(len(logs), 2)
+        self.assertEqual(self.pipeline.captured, 2)
+
+    def test_custom_filter(self):
+        f = FaultFilter([FaultFilterRule(name="warn", level="WARN")])
+        pipeline = CapturePipeline(storage=MemoryStorage(), capture_filter=f)
+        self.assertIsNotNone(pipeline.process({"level": "WARN", "message": "x"}))
+        self.assertIsNone(pipeline.process({"level": "ERROR", "message": "x"}))
+
+    def test_raw_line_pipeline(self):
+        line = "2025-01-01 10:18:00 ERROR [svc] no space left on device"
+        log = self.pipeline.process(line)
+        self.assertIsNotNone(log)
+        self.assertEqual(log.service, "svc")
+        self.assertEqual(log.fault_type, "disk_full")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_root_cause.py b/tests/test_root_cause.py
index 2e7c81a..fee6d87 100644
--- a/tests/test_root_cause.py
+++ b/tests/test_root_cause.py
@@ -1 +1,91 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""根因分析测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+from datetime import datetime, timezone
+
+from fault_log_analyzer.models import Event, FaultLog
+from fault_log_analyzer.root_cause import RootCauseAnalyzer
+
+BASE = datetime(2025, 1, 1, 10, 18, 0, tzinfo=timezone.utc)
+
+
+def _log(msg, host="h-1", trace=None, at=None, fid="fl-1"):
+    return FaultLog(
+        fault_log_id=fid,
+        host_id=host,
+        message=msg,
+        trace_id=trace,
+        occurred_at=at or BASE,
+    )
+
+
+def _event(host="h-1", metric="disk_used_percent", value=97.2, threshold=90.0, at=None):
+    return Event(
+        event_id="e-1",
+        host_id=host,
+        metric=metric,
+        value=value,
+        threshold=threshold,
+        fired_at=at or BASE,
+    )
+
+
+class TestLogHints(unittest.TestCase):
+    def test_disk_hint(self):
+        rc = RootCauseAnalyzer().analyze(_log("no space left on device"), [])
+        self.assertEqual(rc.cause_type, "disk_full")
+        self.assertGreater(rc.confidence, 0.0)
+
+    def test_memory_hint(self):
+        rc = RootCauseAnalyzer().analyze(_log("out of memory killed"), [])
+        self.assertEqual(rc.cause_type, "memory_exhausted")
+
+    def test_unknown_when_no_signal(self):
+        rc = RootCauseAnalyzer().analyze(_log("hello world"), [])
+        self.assertEqual(rc.cause_type, "unknown")
+        self.assertEqual(rc.confidence, 0.0)
+
+
+class TestMetricEvents(unittest.TestCase):
+    def test_metric_event_rule(self):
+        rc = RootCauseAnalyzer().analyze(_log("some error"), [_event()])
+        self.assertEqual(rc.cause_type, "disk_full")
+        self.assertGreater(rc.confidence, 0.0)
+
+    def test_event_outside_window_ignored(self):
+        later = datetime(2025, 1, 1, 11, 0, 0, tzinfo=timezone.utc)
+        rc = RootCauseAnalyzer().analyze(_log("some error"), [_event(at=later)])
+        self.assertEqual(rc.cause_type, "unknown")
+
+    def test_different_host_event_ignored(self):
+        rc = RootCauseAnalyzer().analyze(_log("some error"), [_event(host="h-2")])
+        self.assertEqual(rc.cause_type, "unknown")
+
+    def test_memory_metric_rule(self):
+        event = _event(metric="mem_used_percent", value=95.0, threshold=90.0)
+        rc = RootCauseAnalyzer().analyze(_log("some error"), [event])
+        self.assertEqual(rc.cause_type, "memory_exhausted")
+
+
+class TestTraceAssociation(unittest.TestCase):
+    def test_trace_association(self):
+        log = _log("request failed", trace="tr-1")
+        related = FaultLog(
+            fault_log_id="fl-2",
+            host_id="h-1",
+            message="out of memory",
+            trace_id="tr-1",
+            occurred_at=BASE,
+        )
+        rc = RootCauseAnalyzer().analyze(log, [], related_logs=[related])
+        self.assertEqual(rc.cause_type, "memory_exhausted")
+
+    def test_evidence_collected(self):
+        rc = RootCauseAnalyzer().analyze(_log("no space left on device"), [_event()])
+        types = {e["type"] for e in rc.evidence}
+        self.assertIn("log", types)
+        self.assertIn("event", types)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_storage.py b/tests/test_storage.py
index 2e7c81a..db0e957 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -1 +1,93 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""存储抽象与内存实现测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+from datetime import datetime, timezone
+
+from fault_log_analyzer.models import Event, FaultFilterRule, FaultLog, FaultType, RootCause
+from fault_log_analyzer.storage import MemoryStorage
+
+BASE = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+
+
+def _log(fid, host="h-1", msg="boom", level="ERROR", ft="disk_full", at=None):
+    return FaultLog(
+        fault_log_id=fid,
+        host_id=host,
+        message=msg,
+        level=level,
+        fault_type=ft,
+        occurred_at=at or BASE,
+    )
+
+
+class TestFaultLogStorage(unittest.TestCase):
+    def setUp(self):
+        self.s = MemoryStorage()
+
+    def test_save_and_get(self):
+        self.s.save_fault_log(_log("fl-1"))
+        self.assertIsNotNone(self.s.get_fault_log("fl-1"))
+
+    def test_get_missing(self):
+        self.assertIsNone(self.s.get_fault_log("nope"))
+
+    def test_list_sorted_by_time(self):
+        self.s.save_fault_log(_log("fl-1", at=datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)))
+        self.s.save_fault_log(_log("fl-2", at=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone.utc)))
+        self.assertEqual([l.fault_log_id for l in self.s.list_fault_logs()], ["fl-2", "fl-1"])
+
+    def test_query_filters(self):
+        self.s.save_fault_log(_log("fl-1", host="h-1", ft="disk_full", msg="no space"))
+        self.s.save_fault_log(_log("fl-2", host="h-2", ft="memory_exhausted", msg="oom"))
+        self.assertEqual(self.s.query_fault_logs(host_id="h-1")["total"], 1)
+        self.assertEqual(self.s.query_fault_logs(fault_type="memory_exhausted")["total"], 1)
+        self.assertEqual(self.s.query_fault_logs(keyword="OOM")["total"], 1)
+        self.assertEqual(self.s.query_fault_logs(level="error")["total"], 2)
+
+    def test_query_pagination(self):
+        for i in range(5):
+            self.s.save_fault_log(_log(f"fl-{i}"))
+        r = self.s.query_fault_logs(page=2, page_size=2)
+        self.assertEqual(r["total"], 5)
+        self.assertEqual(len(r["items"]), 2)
+
+
+class TestOtherEntities(unittest.TestCase):
+    def setUp(self):
+        self.s = MemoryStorage()
+
+    def test_root_cause(self):
+        self.s.save_root_cause(RootCause(fault_log_id="fl-1", cause_type="disk_full"))
+        self.assertEqual(self.s.get_root_cause("fl-1").cause_type, "disk_full")
+        self.assertIsNone(self.s.get_root_cause("nope"))
+
+    def test_fault_types(self):
+        self.s.add_fault_type(FaultType(fault_type="x", name="X"))
+        self.assertEqual(self.s.get_fault_type("x").name, "X")
+        self.assertEqual(len(self.s.list_fault_types()), 1)
+
+    def test_filter_rules(self):
+        self.s.add_filter_rule(FaultFilterRule(name="r1", level="ERROR"))
+        self.assertEqual(len(self.s.list_filter_rules()), 1)
+
+    def test_events_by_host_window(self):
+        e1 = Event(event_id="e-1", host_id="h-1", metric="m", value=1.0, threshold=0.0, fired_at=BASE)
+        e2 = Event(event_id="e-2", host_id="h-2", metric="m", value=1.0, threshold=0.0, fired_at=BASE)
+        e3 = Event(
+            event_id="e-3",
+            host_id="h-1",
+            metric="m",
+            value=1.0,
+            threshold=0.0,
+            fired_at=datetime(2025, 2, 1, 10, 0, 0, tzinfo=timezone.utc),
+        )
+        for e in (e1, e2, e3):
+            self.s.save_event(e)
+        start = datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+        end = datetime(2025, 1, 31, 23, 59, 59, tzinfo=timezone.utc)
+        got = self.s.list_events_by_host("h-1", start, end)
+        self.assertEqual([e.event_id for e in got], ["e-1"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_workers.py b/tests/test_workers.py
index 2e7c81a..ab6aad1 100644
--- a/tests/test_workers.py
+++ b/tests/test_workers.py
@@ -1 +1,88 @@
-（已提交至远程 main 分支）
\ No newline at end of file
+"""聚类与根因分析 worker 测试。"""
+import _bootstrap  # noqa: F401
+import unittest
+from datetime import datetime, timezone
+
+from fault_log_analyzer.cluster import Clusterer
+from fault_log_analyzer.models import FaultLog
+from fault_log_analyzer.storage import MemoryStorage
+from fault_log_analyzer.workers import ClusterWorker, RootCauseWorker
+
+BASE = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+
+
+def _logs():
+    return [
+        FaultLog(
+            fault_log_id="fl-1",
+            host_id="h-1",
+            message="no space left on device",
+            level="ERROR",
+            occurred_at=BASE,
+        ),
+        FaultLog(
+            fault_log_id="fl-2",
+            host_id="h-1",
+            message="no space left on device /var",
+            level="ERROR",
+            occurred_at=BASE,
+        ),
+        FaultLog(
+            fault_log_id="fl-3",
+            host_id="h-1",
+            message="out of memory error",
+            level="ERROR",
+            occurred_at=BASE,
+        ),
+    ]
+
+
+class TestClusterWorker(unittest.TestCase):
+    def test_empty(self):
+        s = MemoryStorage()
+        self.assertEqual(
+            ClusterWorker(s).run(), {"clustered": 0, "clusters": 0, "noise": 0}
+        )
+
+    def test_run_clusters_and_classifies(self):
+        s = MemoryStorage()
+        for log in _logs():
+            s.save_fault_log(log)
+        worker = ClusterWorker(s, clusterer=Clusterer(eps=0.75, min_samples=2))
+        result = worker.run()
+        self.assertEqual(result["clustered"], 2)
+        self.assertGreaterEqual(result["clusters"], 1)
+        self.assertEqual(result["noise"], 1)
+
+        fl1 = s.get_fault_log("fl-1")
+        self.assertEqual(fl1.fault_type, "disk_full")
+        self.assertEqual(fl1.cluster_id, "c-0")
+        self.assertIsNone(s.get_fault_log("fl-3").fault_type)
+
+
+class TestRootCauseWorker(unittest.TestCase):
+    def test_empty(self):
+        s = MemoryStorage()
+        self.assertEqual(RootCauseWorker(s).run(), [])
+
+    def test_analyzes_and_skips_existing(self):
+        s = MemoryStorage()
+        log = FaultLog(
+            fault_log_id="fl-1",
+            host_id="h-1",
+            message="no space left on device",
+            level="ERROR",
+            occurred_at=BASE,
+        )
+        s.save_fault_log(log)
+        worker = RootCauseWorker(s)
+        results = worker.run()
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0].cause_type, "disk_full")
+        self.assertEqual(s.get_root_cause("fl-1").cause_type, "disk_full")
+        # 再次运行应跳过已分析日志
+        self.assertEqual(worker.run(), [])
+
+
+if __name__ == "__main__":
+    unittest.main()