AgentLab/backend/tests/test_analysis_scope.py

"""报告文档里系统写的「分析范围」（engine/analysis_scope.py，00-总方案 §3.6，契约 contract.ScopeBlock）。

- 现象、拆过（按下钻链）、排除（分散、差异可能只是样本波动）、算不了、检验次数
- 没拆：候选维度确定地从数据目录算——拆解所用查询的事实表上标成属性、状态、不遮罩、取值不多的列，加上它关联的维度表；
  拆解按的列对不上目录、读不到目录、认不出事实表时只说核对不了，不乱列
- agent 写的假设不进这一段
"""
from __future__ import annotations

import sqlite3
import uuid

import pytest

from app.analysis.testing import fake_result
from app.data import catalog as data_catalog
from app.data.catalog import make_item
from app.data.catalog_view import CatalogView
from app.data.introspect import introspect
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.engine import analysis_scope as scope
from app.engine.evidence import make_eid
from tests.fixtures.sources import drop_source

SQL = "SELECT o.region, o.week, SUM(o.amount) AS gmv FROM orders o JOIN stores s ON s.id = o.store_id GROUP BY 1, 2"
SNAP = "1" * 64
SNAPSHOTS = {SNAP: {"source": "shop", "sql": SQL, "columns": ["region", "week", "gmv"], "rows": []}}


def codes(*values: str, complete: bool = False) -> dict:
    item = make_item({v: "" for v in values}, "profile", "verified")
    return {**item, "complete": True} if complete else item


def column(label: str, measure: str | None = None, codes_item: dict | None = None) -> dict:
    out = {"label": make_item(label, "human")}
    if measure:
        out["measure"] = make_item(measure, "human")
    if codes_item:
        out["codes"] = codes_item
    return out


NOTES = {
    "orders": {
        "kind": make_item("fact", "human"),
        "columns": {
            "region": column("地区", "attribute", codes("华东", "华南", "华北", "西南")),
            "channel": column("渠道", "attribute", codes("线上", "线下")),
            "status": column("订单状态", "status", codes("1", "2", "9", complete=True)),
            "member_level": column("会员等级", "attribute"),                       # 没有码值：取值多不多说不清
            "customer_id": column("客户编号", "identifier", codes("c1", "c2")),
            "phone": column("手机号", "attribute", codes("1", "2")),               # 遮罩列
            "amount": column("实收金额", "flow"),
            "week": column("周", "attribute", codes(*(f"W{i}" for i in range(60)))),  # 取值太多
        },
        "relations": [{"id": "r1", "columns": ["store_id"], "to_table": "stores", "to_columns": ["id"],
                       "cardinality": "many_to_one", "source": "fk", "status": "verified"}],
    },
    "stores": {
        "kind": make_item("dimension", "human"),
        "columns": {"store_type": column("门店类型", "attribute", codes("旗舰", "社区")),
                    "city": column("城市", "attribute", codes("上海", "杭州"))},
    },
}
SOURCES = {"shop": {"tables": NOTES, "masked": ["phone"]}}


def entry(alias: str, result: dict) -> dict:
    artifact = "a" * 63 + alias[-1]
    return {"alias": alias, "kind": "analysis", "artifact": artifact, "eid": make_eid("analysis", artifact, {}),
            "locator": {}, "result": result}


def data(snapshot: str = SNAP) -> list[dict]:
    return [{"ref": "Q1", "artifact": snapshot, "role": "data", "columns": ["region", "week", "gmv"], "rows": "all"}]


def build(catalog: dict, *, sources=SOURCES, snapshots=SNAPSHOTS):
    def load(artifact):
        if artifact not in snapshots:
            raise ValueError("取不回来")
        return snapshots[artifact]

    return scope.build(catalog, sources=sources, load=load)


def test_no_conclusions_no_scope():
    assert scope.build({"m:gmv": {"kind": "metric"}}, sources=SOURCES) is None


def test_the_sections_follow_the_verdicts():
    region = fake_result(inputs=data())
    channel = fake_result(verdict="dispersed", inputs=[*data(), {"ref": "A1", "artifact": "a" * 63 + "1",
                                                                 "role": "parent"}],
                          params={"by": ["channel"], "label": "销售额"}, hypothesis="华东是不是线上拖累的")
    channel["table"]["columns"][0] = "channel"
    channel["table"]["labels"] = {"channel": "渠道"}
    catalog = {
        "A1": entry("A1", region),
        "A2": entry("A2", fake_result("baseline", None, "outside_band")),
        "A3": entry("A3", channel),
        "A4": entry("A4", fake_result("differ", "rate", "no_evidence")),
        "A5": entry("A5", fake_result("trend", None, None, status="refused", reason_code="history_short")),
        "A6": entry("A6", fake_result("differ", "rate", "differs")),
    }
    block = build(catalog)
    assert [i["alias"] for i in block["phenomena"]] == ["A2"]
    assert [(i["alias"], i["parent"], i["dims"], i["verdict"]) for i in block["decomposed"]] == [
        ("A1", None, ["地区"], "driver"), ("A3", "A1", ["渠道"], "dispersed")]
    assert [i["alias"] for i in block["excluded"]] == ["A3", "A4"]
    assert [(i["alias"], i["reason"]) for i in block["refused"]] == [("A5", "历史只有 5 期，至少要 8 期")]
    assert block["tests"] == 2 and block["corrected"] is False
    assert block["decomposed"][0]["phrase"] == region["phrase"]
    assert "华东是不是线上拖累的" not in str(block), "agent 写的假设不进分析范围"


def test_drill_chains_keep_each_parent_ahead_of_its_children():
    def child(of: str) -> dict:
        return fake_result(inputs=[*data(), {"ref": of, "artifact": "a" * 63 + of[-1], "role": "parent"}])

    catalog = {"A1": entry("A1", fake_result(inputs=data())), "A2": entry("A2", fake_result(inputs=data())),
               "A3": entry("A3", child("A1")), "A4": entry("A4", child("A3"))}
    assert [i["alias"] for i in build(catalog)["decomposed"]] == ["A1", "A3", "A4", "A2"]


def test_candidate_dimensions_come_from_the_fact_table_and_its_dimensions():
    block = build({"A1": entry("A1", fake_result(inputs=data()))})
    # 地区拆过了；会员等级没有码值、客户编号是标识、手机号遮罩、金额是流量、周取值太多：都不是候选
    assert block["not_decomposed"] == ["渠道", "订单状态", "门店类型", "城市"] and block["note"] is None
    assert block["tests"] == 0 and block["corrected"] is None


def test_a_dimension_that_does_not_match_the_catalog_is_reported_not_guessed():
    result = fake_result(inputs=data(), params={"by": ["地区名"], "label": "销售额"})
    result["table"]["columns"][0] = "地区名"
    result["table"]["labels"] = {"地区名": "地区名"}
    block = build({"A1": entry("A1", result)})
    assert block["not_decomposed"] is None and block["note"] == "候选维度核对不了：拆解按的「地区名」对不上数据目录里的列"


@pytest.mark.parametrize("kwargs, note", [
    ({"sources": None}, scope.NOTE_NO_CATALOG),
    ({"sources": {"shop": {"tables": {}, "masked": []}}}, scope.NOTE_NO_NOTES),
    ({"sources": {}}, scope.NOTE_NO_CATALOG),
    ({"snapshots": {}}, scope.NOTE_NO_QUERY),
    ({"sources": {"shop": {"tables": {"orders": {"columns": NOTES["orders"]["columns"]},
                                      "stores": NOTES["stores"]}, "masked": []}}}, scope.NOTE_NO_FACT),
])
def test_what_cannot_be_checked_is_said_plainly(kwargs, note):
    block = build({"A1": entry("A1", fake_result(inputs=data()))}, **kwargs)
    assert block["not_decomposed"] is None and block["note"] == note
    assert [i["alias"] for i in block["decomposed"]] == ["A1"], "核对不了的只是候选维度，别的照写"


def test_a_single_unlabelled_table_counts_as_the_fact_table():
    notes = {"orders": {k: v for k, v in NOTES["orders"].items() if k != "kind"}}
    sql = "SELECT region, SUM(amount) AS gmv FROM orders GROUP BY region"
    block = build({"A1": entry("A1", fake_result(inputs=data()))}, sources={"shop": {"tables": notes, "masked": []}},
                  snapshots={SNAP: {**SNAPSHOTS[SNAP], "sql": sql}})
    assert block["not_decomposed"] == ["渠道", "订单状态", "手机号"]


def test_without_a_dimension_decomposition_the_candidates_are_not_checked():
    block = build({"A1": entry("A1", fake_result("baseline", None, "within_band", inputs=data()))})
    assert block["not_decomposed"] is None and block["note"] == scope.NOTE_NO_DECOMPOSE


@pytest.fixture
async def shop(tmp_path):
    path = tmp_path / "scope_shop.db"
    db = sqlite3.connect(path)
    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, channel TEXT, store_id INTEGER, "
                     "phone TEXT, amount REAL, week TEXT);"
                     "CREATE TABLE stores (id INTEGER PRIMARY KEY, store_type TEXT, city TEXT);")
    db.commit()
    db.close()
    async with SessionLocal() as session:
        row = DataSource(name=f"scope_{uuid.uuid4().hex[:6]}", kind="sqlite", database=str(path),
                         options={"mask_columns": "phone"})
        session.add(row)
        await session.commit()
        row.schema_cache = await introspect(row)
        await session.commit()
        await session.refresh(row)
        for table, notes in NOTES.items():
            await data_catalog.write_entry(session, row.id, table, notes, if_version=0, actor=None)
        session.expunge(row)
    yield row
    await drop_source(row.id)


async def test_the_report_reads_the_catalog_through_the_run_view(shop):
    facts = await scope.source_facts([shop.name, "nope"])
    assert set(facts) == {shop.name} and facts[shop.name]["masked"] == ["phone"]
    assert set(facts[shop.name]["tables"]) == {"orders", "stores"}
    assert await scope.source_facts([shop.name], CatalogView(mode="off")) is None

    snapshot = {**SNAPSHOTS[SNAP], "source": shop.name}
    catalog = {"Q1": {"alias": "Q1", "kind": "query", "artifact": SNAP, "source": shop.name},
               "A1": entry("A1", fake_result(inputs=data()))}
    block = await scope.scope_of(catalog, None, load=lambda a: snapshot)
    assert block["not_decomposed"] == ["渠道", "订单状态", "门店类型", "城市"]
    block = await scope.scope_of(catalog, {"mode": "off"}, load=lambda a: snapshot)
    assert block["not_decomposed"] is None and block["note"] == scope.NOTE_NO_CATALOG


def test_an_unmaintained_single_table_catalog_says_so_instead_of_a_column_mismatch():
    """V15（验收复核）：只读一张表、目录还没维护时，_fact_tables 会兜底认出那张表，于是「对不上数据目录里的列」这条
    把「目录没维护」说成了数据对不上。目录一列说明都没有时照实说目录没维护。"""
    from app.engine.analysis_scope import NOTE_NOT_MAINTAINED, _undecomposed

    result = {"op": "decompose", "mode": "additive", "inputs": [{"role": "data", "artifact": "a" * 64}],
              "params": {"by": ["channel"]}, "table": {"columns": ["channel", "base"], "labels": {"channel": "渠道"}}}
    sources = {"源": {"tables": {"orders": {}}, "masked": []}}
    snapshots = {"a" * 64: {"source": "源", "sql": "SELECT channel, SUM(amount) AS amt FROM orders GROUP BY channel",
                            "columns": ["channel", "amt"]}}
    assert _undecomposed([result], {}, sources, snapshots.get) == (None, NOTE_NOT_MAINTAINED)