diff --git a/tests/test_integrations.py b/tests/test_integrations.py
new file mode 100644
index 0000000..1b94d1e
--- /dev/null
+++ b/tests/test_integrations.py
@@ -0,0 +1,94 @@
+import _bootstrap  # noqa: F401
+
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+from fault_log_analyzer.integrations import (
+    ElasticsearchSink,
+    KafkaMessageBus,
+    MySQLFaultLogRepository,
+    MySQLFaultTypeRepository,
+    MySQLFilterRuleRepository,
+    MySQLResultEventRepository,
+    MySQLRootCauseRepository,
+    RedisDedupCache,
+)
+
+
+class TestElasticsearchSinkIndexName(unittest.TestCase):
+    """ElasticsearchSink 的索引命名逻辑（不依赖 ES 客户端即可测试）。"""
+
+    def test_index_name_monthly_rollover(self):
+        sink = object.__new__(ElasticsearchSink)
+        sink._index_pattern = "hms-fault-log-{yyyy.MM}"
+        self.assertEqual(
+            sink._index_name(datetime(2025, 1, 15, tzinfo=timezone.utc)),
+            "hms-fault-log-2025.01",
+        )
+        self.assertEqual(
+            sink._index_name(datetime(2025, 12, 1, tzinfo=timezone.utc)),
+            "hms-fault-log-2025.12",
+        )
+
+    def test_index_name_custom_pattern(self):
+        sink = object.__new__(ElasticsearchSink)
+        sink._index_pattern = "my-index-{yyyy.MM}"
+        self.assertEqual(
+            sink._index_name(datetime(2025, 6, 30, tzinfo=timezone.utc)),
+            "my-index-2025.06",
+        )
+
+
+class TestLazyImportBackends(unittest.TestCase):
+    """可选后端依赖（kafka/elasticsearch/pymysql/redis）未安装时，实例化应抛出
+    明确的 RuntimeError，而非影响核心逻辑与单元测试。"""
+
+    def _assert_missing_dependency_raises_runtime_error(self, factory):
+        with patch("builtins.__import__", side_effect=ImportError("no optional dependency")):
+            with self.assertRaises(RuntimeError):
+                factory()
+
+    def test_kafka_message_bus(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: KafkaMessageBus("localhost:9092", "fault-log-analyzer")
+        )
+
+    def test_elasticsearch_sink(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: ElasticsearchSink(["http://localhost:9200"])
+        )
+
+    def test_mysql_fault_log_repository(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: MySQLFaultLogRepository("h", 3306, "u", "p", "hms")
+        )
+
+    def test_mysql_root_cause_repository(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: MySQLRootCauseRepository("h", 3306, "u", "p", "hms")
+        )
+
+    def test_mysql_fault_type_repository(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: MySQLFaultTypeRepository("h", 3306, "u", "p", "hms")
+        )
+
+    def test_mysql_filter_rule_repository(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: MySQLFilterRuleRepository("h", 3306, "u", "p", "hms")
+        )
+
+    def test_mysql_result_event_repository(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: MySQLResultEventRepository("h", 3306, "u", "p", "hms")
+        )
+
+    def test_redis_dedup_cache(self):
+        self._assert_missing_dependency_raises_runtime_error(
+            lambda: RedisDedupCache("redis://localhost:6379/0")
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()