release: bump version to 0.60.0

backend/tests/test_api.py

@@ -5,10 +5,12 @@ from datetime import datetime
 from unittest.mock import patch, AsyncMock
 from httpx import AsyncClient, ASGITransport
 
+from app.api.v1 import earth as earth_api
 from app.main import app
 from app.core.config import settings
 from app.core.security import create_access_token
 from app.db.session import get_db
+from app.models.system_setting import SystemSetting
 from app.models.user import User
 from app.schemas.ai import (
     AIProviderStatusResponse,
@@ -61,6 +63,148 @@ async def test_root_endpoint():
     assert data["version"] == settings.VERSION
 
 
+class _ScalarOneOrNoneResult:
+    def __init__(self, value=None):
+        self._value = value
+
+    def scalar_one_or_none(self):
+        return self._value
+
+
+class _FakeEarthBrandSession:
+    def __init__(self, record=None):
+        self.record = record
+        self.added = None
+        self.deleted = False
+        self.committed = False
+
+    async def execute(self, statement):
+        if statement.__class__.__name__ == "Delete":
+            self.deleted = True
+            self.record = None
+            return _ScalarOneOrNoneResult(None)
+        return _ScalarOneOrNoneResult(self.record)
+
+    def add(self, record):
+        self.added = record
+        self.record = record
+
+    async def commit(self):
+        self.committed = True
+
+    async def refresh(self, _record):
+        return None
+
+
+def _override_admin_user():
+    return User(id=1, username="testuser", email="test@example.com", role="admin", is_active=True)
+
+
+@pytest.mark.asyncio
+async def test_get_earth_brand_returns_static_defaults():
+    async def override_get_db():
+        yield _FakeEarthBrandSession()
+
+    app.dependency_overrides[get_db] = override_get_db
+    try:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.get("/api/v1/earth/brand")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["is_default"] is True
+        assert data["brand"]["logo_src"] == "/earth/assets/brand/earth-logo.png"
+        assert data["brand"]["title_src"] == "/earth/assets/brand/title-zh.png"
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_update_and_reset_earth_brand(auth_headers):
+    session = _FakeEarthBrandSession()
+
+    async def override_get_db():
+        yield session
+
+    app.dependency_overrides.update(
+        {
+            __import__("app.core.security", fromlist=["get_current_user"]).get_current_user: _override_admin_user,
+            get_db: override_get_db,
+        }
+    )
+    try:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            update_response = await client.put(
+                "/api/v1/earth/brand",
+                headers=auth_headers,
+                json={
+                    "logo_src": "/earth-brand-assets/custom.png",
+                    "title_src": "",
+                    "title_text": "Custom Earth",
+                    "subtitle": "Custom subtitle",
+                    "description": "Custom description",
+                    "aria_label": "",
+                    "title_alt": "",
+                },
+            )
+            reset_response = await client.delete("/api/v1/earth/brand", headers=auth_headers)
+
+        assert update_response.status_code == 200
+        updated = update_response.json()
+        assert updated["is_default"] is False
+        assert updated["brand"]["title_text"] == "Custom Earth"
+        assert updated["brand"]["aria_label"] == "Custom Earth"
+        assert isinstance(session.added, SystemSetting)
+        assert reset_response.status_code == 200
+        reset = reset_response.json()
+        assert reset["is_default"] is True
+        assert reset["brand"]["logo_src"] == "/earth/assets/brand/earth-logo.png"
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_upload_earth_brand_asset_rejects_invalid_type(auth_headers):
+    app.dependency_overrides[
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user
+    ] = _override_admin_user
+    try:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.post(
+                "/api/v1/earth/brand/assets",
+                headers=auth_headers,
+                files={"file": ("brand.txt", b"nope", "text/plain")},
+            )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "unsupported_file_type"
+    finally:
+        app.dependency_overrides.clear()
+
+
+@pytest.mark.asyncio
+async def test_upload_earth_brand_asset_saves_file(auth_headers, tmp_path, monkeypatch):
+    monkeypatch.setattr(earth_api, "EARTH_BRAND_ASSET_DIR", tmp_path)
+    app.dependency_overrides[
+        __import__("app.core.security", fromlist=["get_current_user"]).get_current_user
+    ] = _override_admin_user
+    try:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.post(
+                "/api/v1/earth/brand/assets",
+                headers=auth_headers,
+                files={"file": ("brand.png", b"png-bytes", "image/png")},
+            )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["url"].startswith("/earth-brand-assets/")
+        assert (tmp_path / data["filename"]).read_bytes() == b"png-bytes"
+    finally:
+        app.dependency_overrides.clear()
+
+
 @pytest.mark.asyncio
 async def test_dashboard_stats_without_auth():
     """Test dashboard stats requires authentication"""

backend/tests/test_situational_alert_ai_brief.py

@@ -0,0 +1,71 @@
+import pytest
+
+from app.services.situational_alert_ai_brief import build_situational_alert_brief_request
+
+
+class _SingleUseScalarResult:
+    def __init__(self, value=0, rows=None):
+        self.value = value
+        self.rows = rows or []
+        self.scalar_calls = 0
+
+    def scalar(self):
+        self.scalar_calls += 1
+        if self.scalar_calls > 1:
+            raise AssertionError("scalar result was consumed more than once")
+        return self.value
+
+    def fetchall(self):
+        return self.rows
+
+    def scalar_one_or_none(self):
+        return None
+
+    def scalars(self):
+        rows = self.rows
+
+        class _Scalars:
+            def all(self):
+                return rows
+
+        return _Scalars()
+
+
+class _FakeBriefSession:
+    def __init__(self):
+        self._results = [
+            _SingleUseScalarResult(3),
+            _SingleUseScalarResult(2),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(4),
+            _SingleUseScalarResult(1),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(5),
+            _SingleUseScalarResult(2),
+            _SingleUseScalarResult(rows=[]),
+            _SingleUseScalarResult(),
+        ]
+
+    async def execute(self, _query):
+        return self._results.pop(0)
+
+
+@pytest.mark.asyncio
+async def test_situational_alert_brief_builder_reuses_counts_without_reconsuming_results(monkeypatch):
+    monkeypatch.setattr(
+        "app.services.situational_alert_ai_brief.get_latest_bgp_brief_record",
+        lambda: None,
+    )
+
+    request, facts, context = await build_situational_alert_brief_request(_FakeBriefSession())
+
+    assert request.title == "态势告警 AI 简报"
+    assert "总告警 3 条，active 2 条" in facts[0]
+    assert "累计 incidents 4 条，active incidents 1 条" in facts[1]
+    assert "累计 anomalies 5 条，active anomalies 2 条" in facts[2]
+    assert context["active_system_alerts"] == 2
+    assert context["active_bgp_incidents"] == 1
+    assert context["active_bgp_anomalies"] == 2