"""asset_service 的聚焦测试：CRUD + 列表 + 到期查询（含失败/边界分支） 此前 app/services/asset_service.py 只被间接覆盖（其它测试单独 import 了 export_assets/import_assets/get_overview），核心 CRUD 面 0 直接引用。本文件 以模块名 `asset_service` 直接调用，补齐关键行为与失败/边界分支： - create/get/update/delete 正常路径 + 失败分支（缺详情 400 / 账号悬空 400 / 不存在 404） - delete 的联动清理（详情 + subdomains/site_certs 子表），且 metrics 清理失败可容忍 - VPS 凭证在库内加密存储、输出只给 has_* 标记、更新时凭证为 None 保留原值 - list_assets 的类型/状态/归档筛选、名称与平台搜索、provider slug 与 provider_id、 非法排序字段回退 expiry_date - get_expiring 的到期口径（含已过期、排除已注销、按剩余天数升序） 全程 tmp_path + create_engine 临时库，免凭证、免网络：VPS 凭证加密用一个临时生成的 Fernet key（不读 .env 的真实 MASTER_KEY）。 运行：.venv/bin/pytest tests/test_asset_service.py -v """ from datetime import date, timedelta import pytest from cryptography.fernet import Fernet from fastapi import HTTPException from sqlmodel import Session, SQLModel, create_engine, select import app.models.asset # noqa: F401 确保所有表注册进 SQLModel.metadata 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.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.models.ssl import SiteCert, Subdomain from app.schemas.asset import ( AssetCreate, AssetUpdate, DomainDetailIn, VPSDetailIn, ) from app.services import asset_service @pytest.fixture() def engine(tmp_path, monkeypatch): """临时资产库；metrics_engine 也指过来，让 delete 的监控清理落在临时库而非本地 data/""" 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() # 清掉可能已用旧 key 构建的缓存实例 yield crypto._get_fernet.cache_clear() def _vps_create(name="web-01", ip="1.2.3.4", **kw): detail = kw.pop("vps_detail", VPSDetailIn(ip_address=ip)) return AssetCreate( name=name, asset_type=AssetType.VPS, provider="vultr", vps_detail=detail, **kw ) # ----------------------------- create / get ----------------------------- # def test_create_vps_returns_read_with_detail(engine): with Session(engine) as s: read = asset_service.create_asset(s, _vps_create()) assert read.id is not None assert read.name == "web-01" assert read.vps_detail is not None assert read.vps_detail.ip_address == "1.2.3.4" # 未提供凭证时不应误报存在 assert read.vps_detail.has_ssh_key is False assert read.vps_detail.has_password is False def test_create_vps_encrypts_credentials_at_rest(engine, master_key): """明文凭证必须加密入库，输出只暴露 has_* 标记，绝不回传明文""" detail = VPSDetailIn(ip_address="1.2.3.4", ssh_key="TOPSECRET", password="PW") with Session(engine) as s: read = asset_service.create_asset(s, _vps_create(vps_detail=detail)) stored = s.exec(select(VPSDetail).where(VPSDetail.asset_id == read.id)).one() assert stored.ssh_key_encrypted != "TOPSECRET" # 不是明文 assert crypto.decrypt(stored.ssh_key_encrypted) == "TOPSECRET" # 可还原 assert read.vps_detail.has_ssh_key is True assert "ssh_key" not in read.vps_detail.model_dump() # 输出模型无明文字段 def test_create_asset_missing_detail_raises_400(engine): """asset_type=vps 却不带 vps_detail：失败分支，必须 400 而不是写入残缺资产""" data = AssetCreate(name="no-detail", asset_type=AssetType.VPS, provider="vultr") with Session(engine) as s: with pytest.raises(HTTPException) as ei: asset_service.create_asset(s, data) assert ei.value.status_code == 400 def test_create_asset_invalid_account_raises_400(engine): """account_id 指向不存在的账号：防外键悬空，400""" data = _vps_create(account_id=999) with Session(engine) as s: with pytest.raises(HTTPException) as ei: asset_service.create_asset(s, data) assert ei.value.status_code == 400 def test_create_asset_links_valid_account_and_resolves_name(engine): with Session(engine) as s: acc = Account(name="acc@example.com", platform="vultr") s.add(acc) s.commit() s.refresh(acc) read = asset_service.create_asset(s, _vps_create(account_id=acc.id)) assert read.account_name == "acc@example.com" def test_get_asset_missing_raises_404(engine): with Session(engine) as s: with pytest.raises(HTTPException) as ei: asset_service.get_asset(s, 123456) assert ei.value.status_code == 404 def test_get_asset_computes_days_to_expiry(engine): expiry = date.today() + timedelta(days=7) with Session(engine) as s: created = asset_service.create_asset(s, _vps_create(expiry_date=expiry)) got = asset_service.get_asset(s, created.id) assert got.days_to_expiry == 7 # ----------------------------- update ----------------------------- # def test_update_asset_changes_main_fields(engine): with Session(engine) as s: created = asset_service.create_asset(s, _vps_create()) updated = asset_service.update_asset( s, created.id, AssetUpdate(name="renamed", cost=12.5) ) assert updated.name == "renamed" assert updated.cost == 12.5 def test_update_vps_preserves_credential_when_none(engine, master_key): """更新详情时未携带凭证（None）不能把已存凭证清空""" detail = VPSDetailIn(ip_address="1.2.3.4", ssh_key="KEEPME") with Session(engine) as s: created = asset_service.create_asset(s, _vps_create(vps_detail=detail)) # 只改 IP，不带 ssh_key asset_service.update_asset( s, created.id, AssetUpdate(vps_detail=VPSDetailIn(ip_address="9.9.9.9")) ) stored = s.exec(select(VPSDetail).where(VPSDetail.asset_id == created.id)).one() assert stored.ip_address == "9.9.9.9" assert crypto.decrypt(stored.ssh_key_encrypted) == "KEEPME" def test_update_asset_missing_raises_404(engine): with Session(engine) as s: with pytest.raises(HTTPException) as ei: asset_service.update_asset(s, 123456, AssetUpdate(name="x")) assert ei.value.status_code == 404 # ----------------------------- delete ----------------------------- # def test_delete_asset_removes_detail_and_child_rows(engine): """删除域名资产时，详情与 subdomains/site_certs 子表行都要清掉，避免孤儿数据""" with Session(engine) as s: created = asset_service.create_asset( s, AssetCreate( name="example.com", asset_type=AssetType.DOMAIN, provider="namesilo", domain_detail=DomainDetailIn(domain_name="example.com"), ), ) aid = created.id s.add(Subdomain(asset_id=aid, host="www")) s.add(SiteCert(hostname="www.example.com", asset_id=aid)) s.commit() asset_service.delete_asset(s, aid) assert s.get(Asset, aid) is None assert s.exec(select(DomainDetail).where(DomainDetail.asset_id == aid)).first() is None assert s.exec(select(Subdomain).where(Subdomain.asset_id == aid)).first() is None assert s.exec(select(SiteCert).where(SiteCert.asset_id == aid)).first() is None def test_delete_asset_missing_raises_404(engine): with Session(engine) as s: with pytest.raises(HTTPException) as ei: asset_service.delete_asset(s, 123456) assert ei.value.status_code == 404 def test_delete_asset_tolerates_metrics_cleanup_failure(engine, monkeypatch): """metrics.db 可能被 Agent 短暂锁定：清理失败不能让已成功的删除返回 500""" def _boom(_asset_id): raise RuntimeError("metrics.db is locked") monkeypatch.setattr(asset_service, "_cleanup_metrics_for_asset", _boom) with Session(engine) as s: created = asset_service.create_asset(s, _vps_create()) asset_service.delete_asset(s, created.id) # 不应抛异常 assert s.get(Asset, created.id) is None # ----------------------------- list ----------------------------- # def _seed(engine, name, *, asset_type=AssetType.VPS, provider="vultr", status=AssetStatus.ACTIVE, is_archived=False, provider_id=None): with Session(engine) as s: a = Asset( name=name, asset_type=asset_type, provider=provider, status=status, is_archived=is_archived, provider_id=provider_id, ) s.add(a) s.commit() s.refresh(a) return a.id def test_list_assets_filters_by_type_status_and_archive(engine): _seed(engine, "vps-active", asset_type=AssetType.VPS) _seed(engine, "dom-active", asset_type=AssetType.DOMAIN, provider="namesilo") _seed(engine, "vps-stopped", status=AssetStatus.STOPPED) _seed(engine, "vps-archived", is_archived=True) with Session(engine) as s: vps = asset_service.list_assets(s, asset_type=AssetType.VPS) stopped = asset_service.list_assets(s, asset_status=AssetStatus.STOPPED) archived = asset_service.list_assets(s, is_archived=True) assert {a.name for a in vps} == {"vps-active", "vps-stopped", "vps-archived"} assert {a.name for a in stopped} == {"vps-stopped"} assert {a.name for a in archived} == {"vps-archived"} def test_list_assets_search_and_provider_slug_and_id(engine): with Session(engine) as s: p = Provider(slug="vultr", name="Vultr", category="vps") s.add(p) s.commit() s.refresh(p) pid = p.id _seed(engine, "blog-vps", provider="vultr", provider_id=pid) _seed(engine, "shop-domain", provider="namesilo", asset_type=AssetType.DOMAIN) with Session(engine) as s: by_name = asset_service.list_assets(s, q="blog") by_slug = asset_service.list_assets(s, provider="vultr") by_id = asset_service.list_assets(s, provider=str(pid)) assert {a.name for a in by_name} == {"blog-vps"} assert {a.name for a in by_slug} == {"blog-vps"} assert {a.name for a in by_id} == {"blog-vps"} # provider_name 由 provider_id 关联解析 assert by_id[0].provider_name == "Vultr" def test_list_assets_invalid_sort_falls_back_and_order_desc(engine): """非白名单排序字段必须回退 expiry_date，不能因未知列报错""" with Session(engine) as s: s.add(Asset(name="near", asset_type=AssetType.VPS, provider="vultr", expiry_date=date.today() + timedelta(days=1))) s.add(Asset(name="far", asset_type=AssetType.VPS, provider="vultr", expiry_date=date.today() + timedelta(days=90))) s.commit() asc = asset_service.list_assets(s, sort="; DROP TABLE assets", order="asc") desc = asset_service.list_assets(s, sort="expiry_date", order="desc") assert [a.name for a in asc] == ["near", "far"] assert [a.name for a in desc] == ["far", "near"] # ----------------------------- expiring ----------------------------- # def test_get_expiring_includes_overdue_excludes_cancelled_sorted(engine): today = date.today() _seed(engine, "overdue") _seed(engine, "soon") _seed(engine, "far") _seed(engine, "cancelled") with Session(engine) as s: for name, expiry, status in ( ("overdue", today - timedelta(days=5), AssetStatus.ACTIVE), ("soon", today + timedelta(days=10), AssetStatus.ACTIVE), ("far", today + timedelta(days=100), AssetStatus.ACTIVE), ("cancelled", today + timedelta(days=1), AssetStatus.CANCELLED), ): a = s.exec(select(Asset).where(Asset.name == name)).one() a.expiry_date = expiry a.status = status s.add(a) s.commit() result = asset_service.get_expiring(s, days=30) names = [a.name for a in result] # 已过期(-5) 排在 30 天内(+10) 之前；far 超窗口、cancelled 已注销都不进 assert names == ["overdue", "soon"] assert result[0].days_to_expiry == -5