worldquant-alpha-system/backend/tests/test_research_deletion.py

"""Deleting workspace items hides them without severing research provenance."""

import importlib.util
from pathlib import Path

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations

from tests.test_backtests import execute, setup, start
from tests.test_catalog import catalog as catalog_fixture
from tests.test_research_workspace import expansion, template
from tests.test_research_workspace import research_input as research_input_fixture

catalog = catalog_fixture
research_input = research_input_fixture


@pytest.mark.parametrize("kind", ["template", "feature"])
async def test_asset_deletion_checks_version_and_preserves_revisions(logged_in, research_input, kind):
    content = template() if kind == "template" else {
        "name": "待删除特征",
        "hypothesis": "排名比较",
        "input_ids": [research_input["id"]],
        "template": template(),
    }
    response = await logged_in.post("/api/v1/research/assets", json={"kind": kind, "content": content})
    assert response.status_code == 201, response.text
    original = response.json()
    url = f"/api/v1/research/assets/{original['id']}"
    updated = await logged_in.put(url, json={"kind": kind, "version": 1, "content": {**content, "name": "新版"}})
    assert updated.status_code == 200
    assert (await logged_in.delete(f"{url}?version=1")).status_code == 409
    assert not (await logged_in.get(url)).json()["archived"]
    assert (await logged_in.delete(f"{url}?version=2")).status_code == 200
    listing = (await logged_in.get(f"/api/v1/research/assets?kind={kind}")).json()
    assert listing["total"] == 0 and listing["items"] == []
    historical = (await logged_in.get(f"{url}?version=1")).json()
    assert historical["archived"] is True
    assert historical["content"] == original["content"]
    assert historical["provenance"] == original["provenance"]
    latest = (await logged_in.get(url)).json()
    assert latest["version"] == 3
    assert (await logged_in.delete(f"{url}?version=3")).status_code == 409
    assert (await logged_in.put(url, json={"kind": kind, "version": 3, "content": content})).status_code == 409
    assert (await logged_in.get(url)).json()["version"] == 3


async def test_experiment_deletion_preserves_backtest_and_child_sources(app, logged_in, research_input):
    created = await logged_in.post("/api/v1/research/experiments", json=expansion(research_input["id"]))
    assert created.status_code == 201, created.text
    parent = created.json()
    generated = await logged_in.post(
        "/api/v1/research/experiments",
        json=expansion(research_input["id"], parent_experiment_ids=[parent["id"]]),
    )
    assert generated.status_code == 201, generated.text
    variant = generated.json()
    assert variant["kind"] == "variant"
    child = await logged_in.post(
        "/api/v1/research/experiments",
        json=expansion(research_input["id"], parent_experiment_ids=[variant["id"]]),
    )
    assert child.status_code == 201
    _, lane = await setup(app)
    url = f"/api/v1/research/experiments/{variant['id']}"
    preview = await logged_in.post(f"{url}/preview", json={})
    assert preview.status_code == 201, preview.text
    run = await start(logged_in, preview.json(), "delete-variant-provenance")
    await execute(app, lane, run["backtest_run_id"])
    before = (await logged_in.get(url)).json()
    assert before["backtest_run_ids"] == [run["backtest_run_id"]]

    assert (await logged_in.delete(url)).status_code == 200
    assert (await logged_in.delete(url)).status_code == 200
    all_items = (await logged_in.get("/api/v1/research/experiments")).json()
    assert all_items["total"] == 2
    assert variant["id"] not in {item["id"] for item in all_items["items"]}
    variants = (await logged_in.get("/api/v1/research/experiments?kind=variant&limit=1&offset=1")).json()
    assert variants["total"] == 1 and variants["items"] == []
    after = (await logged_in.get(url)).json()
    assert after == {**before, "archived": True}
    saved_child = (await logged_in.get(f"/api/v1/research/experiments/{child.json()['id']}")).json()
    assert saved_child["parents"] == child.json()["parents"]
    results = (await logged_in.get(f"/api/v1/backtests/runs/{run['backtest_run_id']}/results")).json()
    alpha_id = results["items"][0]["alpha_id"]
    sources = (await logged_in.get(f"/api/v1/alphas/{alpha_id}/sources")).json()
    assert sources["items"][0]["source"]["research_id"] == variant["id"]
    assert (await logged_in.delete(f"/api/v1/research/experiments/{parent['id']}")).status_code == 200
    assert (await logged_in.get("/api/v1/research/experiments")).json()["total"] == 1


async def test_delete_research_requires_auth_and_known_id(client):
    for url in ("assets/missing?version=1", "experiments/missing"):
        assert (await client.delete(f"/api/v1/research/{url}")).status_code == 401


async def test_delete_unknown_experiment(logged_in):
    assert (await logged_in.delete("/api/v1/research/experiments/missing")).status_code == 404


def test_archive_migration_preserves_existing_rows(tmp_path):
    path = Path(__file__).parents[1] / "migrations/versions/0014_research_experiment_archive.py"
    spec = importlib.util.spec_from_file_location("research_archive_migration", path)
    migration = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(migration)
    engine = sa.create_engine(f"sqlite:///{tmp_path}/migration.db")
    with engine.begin() as connection:
        connection.execute(sa.text("CREATE TABLE research_experiments (id VARCHAR(36) PRIMARY KEY, evidence JSON NOT NULL)"))
        connection.execute(sa.text("INSERT INTO research_experiments VALUES ('existing', :evidence)"), {"evidence": '{"source":"old"}'})
        with Operations.context(MigrationContext.configure(connection)):
            migration.upgrade()
            assert connection.execute(sa.text("SELECT archived FROM research_experiments")).scalar_one() == 0
            connection.execute(sa.text("UPDATE research_experiments SET archived = 1"))
            migration.downgrade()
            migration.upgrade()
        assert connection.execute(sa.text("SELECT evidence FROM research_experiments")).scalar_one() == '{"source":"old"}'
        column = next(c for c in sa.inspect(connection).get_columns("research_experiments") if c["name"] == "archived")
        assert column["nullable"] is False
    engine.dispose()