"""PR3 tests — AI narratives, department savings rollup, IT digest guard.""" from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession import app.domains.bailian.llm_client as llm_mod from app.domains.cmdb import efficiency_ai, efficiency_service from app.domains.cmdb.models import CmdbServer, CmdbServerEfficiency from app.identity.person_models import IdentityOrgMembership, IdentityPerson from app.integrations.models import Integration, PlatformSetting async def _integration(db: AsyncSession) -> Integration: integ = Integration( id=uuid.uuid4(), kind="cloud", provider="aws", name="aws-pr3", slug=f"aws-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers", "metrics"], ) db.add(integ) await db.flush() return integ async def _person(db: AsyncSession, name: str, org_path: str | None) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=f"{uuid.uuid4().hex[:8]}@pingcap.com", display_name=name, employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.flush() if org_path: db.add(IdentityOrgMembership( id=uuid.uuid4(), person_id=p.id, org_node_source_id=f"node-{uuid.uuid4().hex[:8]}", org_node_type="department", org_node_name=org_path.split("/")[-1].strip(), org_path=org_path, is_primary=True, membership_status="active", )) await db.flush() return p async def _server_eff( db, integ, *, owner_id=None, verdict="idle", reco="stop", savings=100, cost=100, cpu_p95=2.0, ) -> tuple[CmdbServer, CmdbServerEfficiency]: now = datetime.now(UTC) srv = CmdbServer( id=uuid.uuid4(), instance_id=f"i-{uuid.uuid4().hex[:10]}", name="box", status="running", region="us-east-1", integration_id=str(integ.id), vcpu=4, memory_mb=16384, instance_type="m5.xlarge", owner_person_id=owner_id, first_seen_at=now - timedelta(days=90), last_synced_at=now, ) db.add(srv) await db.flush() eff = CmdbServerEfficiency( id=uuid.uuid4(), server_id=str(srv.id), integration_id=str(integ.id), window_days=14, data_days=14, cpu_avg=cpu_p95 - 1, cpu_p95=cpu_p95, cpu_max=cpu_p95 + 5, verdict=verdict, recommendation=reco, monthly_cost_usd=Decimal(cost), estimated_monthly_savings_usd=Decimal(savings), cost_currency="USD", rule_version="v1", computed_at=now, ) db.add(eff) await db.flush() return srv, eff # ── department rollup ──────────────────────────────────────────────── @pytest.mark.asyncio async def test_savings_by_department_rolls_up_subtree(db_session: AsyncSession): integ = await _integration(db_session) p_be = await _person(db_session, "BE", "PingCAP / Engineering / Backend") p_fe = await _person(db_session, "FE", "PingCAP / Engineering / Frontend") await _server_eff(db_session, integ, owner_id=str(p_be.id), savings=100) await _server_eff(db_session, integ, owner_id=str(p_fe.id), savings=40) await db_session.commit() rows = await efficiency_service.savings_by_department(db_session) by = {r["department"]: r for r in rows} # Parent shows subtree total; leaves show their own. assert by["Engineering"]["estimated_monthly_savings_usd"] == 140.0 assert by["Engineering"]["servers"] == 2 assert by["Backend"]["estimated_monthly_savings_usd"] == 100.0 assert by["Frontend"]["estimated_monthly_savings_usd"] == 40.0 @pytest.mark.asyncio async def test_savings_by_department_unassigned_bucket(db_session: AsyncSession): integ = await _integration(db_session) await _server_eff(db_session, integ, owner_id=None, savings=25) await db_session.commit() rows = await efficiency_service.savings_by_department(db_session) assert any(r["department"] == "__unassigned__" for r in rows) async def _scope_with_cost_fact(db, integ, owner, srv): """Subscription owned by ``owner`` + a cost-fact line attributing ``srv`` to it — the standard server→scope billing link.""" from datetime import date from app.domains.cmdb.models import CmdbCloudScope, CmdbCostFact scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-{uuid.uuid4().hex[:8]}", display_name="sub", owner_person_id=str(owner.id), ) db.add(scope) await db.flush() db.add(CmdbCostFact( fact_key=uuid.uuid4().hex, date=date.today(), integration_id=str(integ.id), scope_id=str(scope.id), resource_id=srv.instance_id, resource_type="Instance", service="AmazonEC2", amount_usd=5, currency="USD", )) await db.flush() return scope @pytest.mark.asyncio async def test_savings_by_department_inherits_scope_owner( db_session: AsyncSession, ): """A server with NO stored owner attributes to the department of the owner it inherits from its cloud subscription — it must not land in the unassigned bucket when the subscription has an owner.""" integ = await _integration(db_session) scope_owner = await _person( db_session, "SO", "PingCAP / Engineering / Cloud", ) srv, _ = await _server_eff(db_session, integ, owner_id=None, savings=77) await _scope_with_cost_fact(db_session, integ, scope_owner, srv) await db_session.commit() rows = await efficiency_service.savings_by_department(db_session) by = {r["department"]: r for r in rows} assert by["Cloud"]["estimated_monthly_savings_usd"] == 77.0 assert by["Engineering"]["estimated_monthly_savings_usd"] == 77.0 assert "__unassigned__" not in by @pytest.mark.asyncio async def test_savings_by_department_explicit_owner_beats_inheritance( db_session: AsyncSession, ): """A transferred (explicitly-assigned) owner re-homes the machine even when the subscription owner sits in a different department.""" integ = await _integration(db_session) scope_owner = await _person( db_session, "SO", "PingCAP / Engineering / Cloud", ) explicit = await _person(db_session, "EX", "PingCAP / Product / Growth") srv, _ = await _server_eff( db_session, integ, owner_id=str(explicit.id), savings=33, ) await _scope_with_cost_fact(db_session, integ, scope_owner, srv) await db_session.commit() rows = await efficiency_service.savings_by_department(db_session) by = {r["department"]: r for r in rows} assert by["Growth"]["estimated_monthly_savings_usd"] == 33.0 assert "Cloud" not in by # ── AI narratives ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_generate_narratives_no_key_skips(db_session: AsyncSession): integ = await _integration(db_session) await _server_eff(db_session, integ, verdict="idle", savings=100) await db_session.commit() res = await efficiency_ai.generate_narratives(db_session, str(integ.id)) assert res["status"] == "skipped" assert res["reason"] == "llm_not_configured" @pytest.mark.asyncio async def test_generate_narratives_writes_cache( db_session: AsyncSession, monkeypatch, ): db_session.add(PlatformSetting(key="ai_spending.llm.api_key", value="sk-test")) await db_session.flush() async def _fake(self, *, system, user, max_tokens=300, temperature=0.4): return "这台机器 CPU 长期低于 3%,几乎无网络流量,建议关停,预计每月省 $100。" monkeypatch.setattr(llm_mod.BailianLLMClient, "chat_text", _fake) integ = await _integration(db_session) srv, _ = await _server_eff(db_session, integ, verdict="idle", savings=100) healthy_srv, _ = await _server_eff(db_session, integ, verdict="healthy", reco="keep", savings=0) await db_session.commit() res = await efficiency_ai.generate_narratives(db_session, str(integ.id)) assert res["status"] == "ok" assert res["generated"] == 1 # only the actionable (idle) one idle_eff = (await db_session.execute( select(CmdbServerEfficiency).where(CmdbServerEfficiency.server_id == str(srv.id)) )).scalar_one() assert idle_eff.ai_narrative and "关停" in idle_eff.ai_narrative assert idle_eff.ai_narrative_at is not None healthy_eff = (await db_session.execute( select(CmdbServerEfficiency).where(CmdbServerEfficiency.server_id == str(healthy_srv.id)) )).scalar_one() assert healthy_eff.ai_narrative is None # non-actionable → no narrative @pytest.mark.asyncio async def test_generate_narratives_skips_fresh(db_session: AsyncSession, monkeypatch): db_session.add(PlatformSetting(key="ai_spending.llm.api_key", value="sk-test")) await db_session.flush() calls = {"n": 0} async def _fake(self, *, system, user, max_tokens=300, temperature=0.4): calls["n"] += 1 return "narrative" monkeypatch.setattr(llm_mod.BailianLLMClient, "chat_text", _fake) integ = await _integration(db_session) srv, eff = await _server_eff(db_session, integ, verdict="idle", savings=100) eff.ai_narrative = "existing fresh" eff.ai_narrative_at = datetime.now(UTC) # fresh await db_session.commit() res = await efficiency_ai.generate_narratives(db_session, str(integ.id)) assert res["generated"] == 0 assert calls["n"] == 0 # fresh → no LLM call # ── IT digest guard ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_digest_skips_when_no_actionable(client, db_session: AsyncSession): # ``client`` fixture monkeypatches app.core.database.async_session_factory # to the test engine — run_efficiency_digest opens its own session via # that factory, so without this it would hit the real app DB. from app.scheduled_jobs.handlers import run_efficiency_digest integ = await _integration(db_session) # Only a healthy server → zero actionable savings. await _server_eff(db_session, integ, verdict="healthy", reco="keep", savings=0) await db_session.commit() res = await run_efficiency_digest(db_session, {}) assert res["status"] == "skipped" assert "no actionable" in res["reason"]