diff --git a/tests/test_sync_service.py b/tests/test_sync_service.py
new file mode 100644
index 0000000..f3bd348
--- /dev/null
+++ b/tests/test_sync_service.py
@@ -0,0 +1,336 @@
+"""sync_service 核心同步流程测试
+
+覆盖 sync_account 的三个关键场景：
+1. 正常同步入库：适配器返回 VPS + 域名，验证资产正确写入数据库
+2. (provider_id, external_id) 去重：同 provider_id + external_id 的资产再次同步时更新而非新建
+3. 适配器异常容错：VPS 同步抛异常时不中断整体流程，域名仍可同步，错误记录在结果中
+
+全程 tmp_path + create_engine 临时库，mock 适配器，免凭证免网络。
+
+运行：.venv/bin/python -m pytest tests/test_sync_service.py -v
+"""
+
+from datetime import date, timedelta
+
+import pytest
+from cryptography.fernet import Fernet
+from sqlmodel import Session, SQLModel, create_engine, select
+
+import app.models.asset  # noqa: F401
+import app.models.credential  # noqa: F401
+import app.models.monitor  # noqa: F401
+import app.models.provider  # noqa: F401
+import app.models.ssl  # noqa: F401
+from app.adapters.base import (
+    AccountInfo,
+    BaseAdapter,
+    NormalizedDomain,
+    NormalizedVPS,
+)
+from app.core import crypto
+from app.core.config import settings
+from app.models.asset import Account, Asset, AssetStatus, AssetType, DomainDetail, VPSDetail
+from app.models.provider import Provider
+from app.services import sync_service
+
+
+# ----------------------------- fixtures ----------------------------- #
+
+
+@pytest.fixture()
+def engine(tmp_path, monkeypatch):
+    """临时资产库"""
+    eng = create_engine(
+        f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False}
+    )
+    SQLModel.metadata.create_all(eng)
+    monkeypatch.setattr("app.database.metrics_engine", eng)
+    yield eng
+    eng.dispose()
+
+
+@pytest.fixture()
+def master_key(monkeypatch):
+    """临时 Fernet key，不依赖 .env 里的真实 MASTER_KEY"""
+    monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode())
+    crypto._get_fernet.cache_clear()
+    yield
+    crypto._get_fernet.cache_clear()
+
+
+# ----------------------------- mock adapter ----------------------------- #
+
+
+class MockAdapter(BaseAdapter):
+    """可配置返回值的测试适配器"""
+
+    required_config = ["api_key"]
+
+    def __init__(self, config=None, *, vps_list=None, domain_list=None, account_info=None,
+                 vps_raises=None, domain_raises=None):
+        super().__init__(config)
+        self._vps_list = vps_list or []
+        self._domain_list = domain_list or []
+        self._account_info = account_info
+        self._vps_raises = vps_raises
+        self._domain_raises = domain_raises
+
+    def test_connection(self):
+        return {"ok": True, "message": "mock"}
+
+    def list_vps(self):
+        if self._vps_raises:
+            raise self._vps_raises
+        return self._vps_list
+
+    def list_domains(self):
+        if self._domain_raises:
+            raise self._domain_raises
+        return self._domain_list
+
+    def get_account(self):
+        if self._account_info:
+            return self._account_info
+        return AccountInfo(balance=100.0, currency="USD")
+
+    def capabilities(self):
+        return {"list_vps": True, "list_domains": True, "get_account": True}
+
+
+def _seed_provider(session, slug="vultr", name="Vultr", sdk_type="vultr-api"):
+    """在库中种入一个 Provider 并返回"""
+    p = Provider(slug=slug, name=name, sdk_type=sdk_type, category="vps")
+    session.add(p)
+    session.commit()
+    session.refresh(p)
+    return p
+
+
+def _seed_account(session, platform="vultr", name="test@example.com", api_config=None):
+    """在库中种入一个 Account 并返回"""
+    from app.core.crypto import encrypt
+    config_enc = encrypt(api_config or '{"api_key": "fake-key"}')
+    acc = Account(name=name, platform=platform, api_config_encrypted=config_enc)
+    session.add(acc)
+    session.commit()
+    session.refresh(acc)
+    return acc
+
+
+def _patch_adapter(monkeypatch, adapter):
+    """让 _build_adapter_for_account 返回固定的 mock 适配器"""
+    def _fake_build(session, account):
+        provider = session.exec(select(Provider).where(Provider.slug == account.platform)).first()
+        return provider, adapter
+
+    monkeypatch.setattr(sync_service, "_build_adapter_for_account", _fake_build)
+
+
+# ----------------------------- 场景 1：正常同步入库 ----------------------------- #
+
+
+def test_sync_account_normal路径_vps_and_domains_created(engine, master_key, monkeypatch):
+    """适配器返回 2 台 VPS + 1 个域名，同步后全部正确入库"""
+    today = date.today()
+    vps_list = [
+        NormalizedVPS(
+            external_id="vps-001", name="web-server",
+            ip_address="1.2.3.4", region="Tokyo", os="Debian 12",
+            cpu_cores=2, memory_gb=4, disk_gb=80,
+            status="active", monthly_cost=12.5, currency="USD",
+            expiry_date=today + timedelta(days=30),
+        ),
+        NormalizedVPS(
+            external_id="vps-002", name="db-server",
+            ip_address="5.6.7.8", region="Singapore",
+            status="active",
+        ),
+    ]
+    domain_list = [
+        NormalizedDomain(
+            external_id="dom-001", domain_name="example.com",
+            registrar="namesilo", expiry_date=today + timedelta(days=365),
+            status="active",
+        ),
+    ]
+
+    adapter = MockAdapter(config={"api_key": "fake"}, vps_list=vps_list, domain_list=domain_list)
+    _patch_adapter(monkeypatch, adapter)
+
+    with Session(engine) as s:
+        provider = _seed_provider(s)
+        account = _seed_account(s)
+
+        result = sync_service.sync_account(s, account.id)
+
+        # 验证返回结构
+        assert result["provider"] == "vultr"
+        assert result["vps"]["created"] == 2
+        assert result["vps"]["updated"] == 0
+        assert result["domains"]["created"] == 1
+        assert result["domains"]["updated"] == 0
+        assert "vps_error" not in result
+        assert "domains_error" not in result
+        assert result["last_synced_at"] is not None
+
+        # 验证数据库中的资产
+        all_assets = s.exec(select(Asset)).all()
+        assert len(all_assets) == 3  # 2 VPS + 1 domain
+
+        vps_assets = [a for a in all_assets if a.asset_type == AssetType.VPS]
+        dom_assets = [a for a in all_assets if a.asset_type == AssetType.DOMAIN]
+        assert len(vps_assets) == 2
+        assert len(dom_assets) == 1
+
+        # 验证 VPS 详情
+        web = next(a for a in vps_assets if a.external_id == "vps-001")
+        assert web.name == "web-server"
+        assert web.status == AssetStatus.ACTIVE
+        assert web.cost == 12.5
+        assert web.account_id == account.id
+
+        detail = s.exec(select(VPSDetail).where(VPSDetail.asset_id == web.id)).first()
+        assert detail is not None
+        assert detail.ip_address == "1.2.3.4"
+        assert detail.region == "Tokyo"
+        assert detail.cpu_cores == 2
+
+        # 验证域名详情
+        dom = dom_assets[0]
+        assert dom.external_id == "dom-001"
+        dom_detail = s.exec(select(DomainDetail).where(DomainDetail.asset_id == dom.id)).first()
+        assert dom_detail.domain_name == "example.com"
+        assert dom_detail.registrar == "namesilo"
+
+
+# ----------------------------- 场景 2：去重逻辑 ----------------------------- #
+
+
+def test_sync_account_去重_provider_id_external_id(engine, master_key, monkeypatch):
+    """相同 (provider_id, external_id, asset_type) 的资产再次同步时更新而非新建"""
+    today = date.today()
+
+    # 第一次同步：创建 1 台 VPS
+    first_vps = [
+        NormalizedVPS(
+            external_id="vps-001", name="web-server",
+            ip_address="1.2.3.4", status="active",
+            expiry_date=today + timedelta(days=30),
+        ),
+    ]
+    adapter1 = MockAdapter(config={"api_key": "fake"}, vps_list=first_vps)
+    _patch_adapter(monkeypatch, adapter1)
+
+    with Session(engine) as s:
+        provider = _seed_provider(s)
+        account = _seed_account(s)
+        result1 = sync_service.sync_account(s, account.id)
+        assert result1["vps"]["created"] == 1
+        assert result1["vps"]["updated"] == 0
+
+    # 第二次同步：同一 external_id，名称和 IP 变了
+    second_vps = [
+        NormalizedVPS(
+            external_id="vps-001", name="web-server-renamed",
+            ip_address="9.9.9.9", status="active",
+            expiry_date=today + timedelta(days=60),
+        ),
+    ]
+    adapter2 = MockAdapter(config={"api_key": "fake"}, vps_list=second_vps)
+    _patch_adapter(monkeypatch, adapter2)
+
+    with Session(engine) as s:
+        result2 = sync_service.sync_account(s, account.id)
+        assert result2["vps"]["created"] == 0
+        assert result2["vps"]["updated"] == 1
+
+        # 验证没有新建资产，而是更新了已有资产
+        all_assets = s.exec(select(Asset)).all()
+        assert len(all_assets) == 1  # 仍然只有 1 条
+
+        asset = all_assets[0]
+        assert asset.name == "web-server-renamed"
+        assert asset.expiry_date == today + timedelta(days=60)
+
+        # 验证 VPS 详情也更新了
+        detail = s.exec(select(VPSDetail).where(VPSDetail.asset_id == asset.id)).first()
+        assert detail.ip_address == "9.9.9.9"
+
+
+# ----------------------------- 场景 3：适配器异常容错 ----------------------------- #
+
+
+def test_sync_account_vps异常_域名仍可同步(engine, master_key, monkeypatch):
+    """VPS 同步抛异常时，域名同步不受影响，错误记录在结果中"""
+    today = date.today()
+    domain_list = [
+        NormalizedDomain(
+            external_id="dom-001", domain_name="example.com",
+            registrar="namesilo", expiry_date=today + timedelta(days=365),
+            status="active",
+        ),
+    ]
+
+    # VPS 同步会抛异常，但域名正常
+    adapter = MockAdapter(
+        config={"api_key": "fake"},
+        vps_raises=RuntimeError("API rate limit exceeded"),
+        domain_list=domain_list,
+    )
+    _patch_adapter(monkeypatch, adapter)
+
+    with Session(engine) as s:
+        provider = _seed_provider(s)
+        account = _seed_account(s)
+
+        result = sync_service.sync_account(s, account.id)
+
+        # VPS 失败应记录错误
+        assert "vps_error" in result
+        assert "rate limit" in result["vps_error"].lower()
+        assert "vps" not in result or result.get("vps") is None
+
+        # 域名应正常同步
+        assert result["domains"]["created"] == 1
+        assert result["domains"]["updated"] == 0
+        assert "domains_error" not in result
+
+        # 数据库中只有域名，没有 VPS
+        all_assets = s.exec(select(Asset)).all()
+        assert len(all_assets) == 1
+        assert all_assets[0].asset_type == AssetType.DOMAIN
+
+
+def test_sync_account_domain异常_vps仍可同步(engine, master_key, monkeypatch):
+    """域名同步抛异常时，VPS 同步不受影响（对称验证）"""
+    vps_list = [
+        NormalizedVPS(
+            external_id="vps-001", name="web-server",
+            ip_address="1.2.3.4", status="active",
+        ),
+    ]
+
+    adapter = MockAdapter(
+        config={"api_key": "fake"},
+        vps_list=vps_list,
+        domain_raises=ConnectionError("DNS resolution failed"),
+    )
+    _patch_adapter(monkeypatch, adapter)
+
+    with Session(engine) as s:
+        provider = _seed_provider(s)
+        account = _seed_account(s)
+
+        result = sync_service.sync_account(s, account.id)
+
+        # VPS 正常
+        assert result["vps"]["created"] == 1
+        assert "vps_error" not in result
+
+        # 域名失败
+        assert "domains_error" in result
+        assert "DNS resolution" in result["domains_error"]
+
+        all_assets = s.exec(select(Asset)).all()
+        assert len(all_assets) == 1
+        assert all_assets[0].asset_type == AssetType.VPS