hub-core/tests/test_legacy_topic_aliases.py

"""Legacy topic alias seed and the RM_SLICE_TOPICSPINE A5 rollback switch.

The ``test_a5_*`` cases run the State Hub A5 comparison packet
(STATE-WP-0098-T01, cases A5-C01..C10) locally against checked-in fixtures:
the legacy side is the 2026-10-13 ``GET /topics/?status=active`` capture and
the candidate side is the in-memory runtime fed with the contract fixture page.
"""

from __future__ import annotations

import json
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from jsonschema import Draft202012Validator, FormatChecker

from hub_core.contracts import repository_navigation_contract_root
from hub_core.runtime.app import create_app
from hub_core.runtime.config import RuntimeSettings
from hub_core.runtime.legacy_topic_aliases import (
    LegacyTopicAliasError,
    load_legacy_topic_aliases,
    parse_legacy_topic_aliases,
)
from hub_core.runtime.repository_navigation import RepositoryNavigationService
from hub_core.runtime.store import InMemoryPortStore

LEGACY_CAPTURE = Path(__file__).parent / "fixtures" / "a5-legacy-topics-2026-10-13.json"
SEED = Path(__file__).parents[1] / "hub_core" / "runtime" / "data" / "legacy_topic_aliases.json"
REPOSITORIES = "/ports/projections/repository-navigation/repositories"


def legacy_topics() -> list[dict[str, Any]]:
    return json.loads(LEGACY_CAPTURE.read_text(encoding="utf-8"))["topics"]


def seed_document() -> dict[str, Any]:
    return json.loads(SEED.read_text(encoding="utf-8"))


def fixture_page() -> dict[str, Any]:
    return json.loads(
        repository_navigation_contract_root()
        .joinpath("fixtures", "repository-classification-page.json")
        .read_text(encoding="utf-8")
    )


def output_validator() -> Draft202012Validator:
    schema = json.loads(
        repository_navigation_contract_root()
        .joinpath("schemas", "repository-navigation-projection.schema.json")
        .read_text(encoding="utf-8")
    )
    return Draft202012Validator(schema, format_checker=FormatChecker())


class PageClient:
    def __init__(self, page: dict[str, Any]) -> None:
        self.page = page

    async def fetch_classification_page(self, cursor: str | None) -> dict[str, Any]:
        return deepcopy(self.page)


def runtime(mode: str | None) -> TestClient:
    kwargs = {} if mode is None else {"rm_slice_topicspine": mode}
    settings = RuntimeSettings(
        environment="test", backend="memory", allow_ephemeral=True, **kwargs
    )
    return TestClient(
        create_app(
            settings=settings,
            port_store=InMemoryPortStore(),
            repo_projection_client=PageClient(fixture_page()),
        )
    )


def as_legacy_rows(aliases: list[dict[str, Any]]) -> set[tuple[str, str, str]]:
    return {
        (alias["legacy_topic_id"], alias["legacy_topic_slug"], alias["facet_value"])
        for alias in aliases
    }


def expected_rows() -> set[tuple[str, str, str]]:
    # Normalisation documented in the seed: 1.0.0 slugs forbid underscores.
    return {
        (topic["id"], topic["slug"].replace("_", "-"), topic["domain_slug"])
        for topic in legacy_topics()
    }


# --- seed ------------------------------------------------------------------


def test_seed_is_versioned_non_authoritative_and_expiring() -> None:
    document = seed_document()
    alias_set = load_legacy_topic_aliases()

    assert document["authority"] is False
    assert alias_set.alias_set_version == document["alias_set_version"]
    assert len(alias_set.aliases) == 16
    assert all(alias.facet_kind == "primary_domain" for alias in alias_set.aliases)
    assert all(alias.residual_owner == "STATE-WP-0079" for alias in alias_set.aliases)
    assert all(
        alias.expires_at == datetime(2027, 3, 31, 23, 59, 59, tzinfo=timezone.utc)
        for alias in alias_set.aliases
    )


def test_seed_keeps_legacy_topic_ids_and_spelling_stable() -> None:
    alias_set = load_legacy_topic_aliases()
    by_id = {alias.legacy_topic_id: alias for alias in alias_set.aliases}

    for topic in legacy_topics():
        alias = by_id[topic["id"]]
        assert alias.legacy_topic_slug == topic["slug"]
        assert alias.facet_value == topic["domain_slug"]
    assert alias_set.resolve("inter_hub") is alias_set.resolve("inter-hub")
    assert alias_set.resolve("cee7bedf-2b48-46ef-8601-006474f2ad7a").facet_value == "infotech"
    assert alias_set.resolve("no-such-topic") is None


def test_alias_status_turns_expired_at_expires_at() -> None:
    alias_set = load_legacy_topic_aliases()
    before = alias_set.to_contract(now=datetime(2027, 3, 31, 23, 59, 58, tzinfo=timezone.utc))
    after = alias_set.to_contract(now=datetime(2027, 3, 31, 23, 59, 59, tzinfo=timezone.utc))

    assert {alias["status"] for alias in before} == {"active"}
    assert {alias["status"] for alias in after} == {"expired"}


@pytest.mark.parametrize(
    ("mutate", "message"),
    [
        (lambda doc: doc.update(authority=True), "authority"),
        (lambda doc: doc.pop("expires_at"), "expires_at"),
        (lambda doc: doc.update(residual_owner=""), "legacyTopicAlias"),
        (lambda doc: doc["aliases"].append(dict(doc["aliases"][0])), "duplicate"),
        (lambda doc: doc["aliases"][0].update(facet_value="no-such-domain"), "invalid facet"),
        (lambda doc: doc["aliases"][0].update(legacy_topic_id="not-a-uuid"), "legacyTopicAlias"),
        (lambda doc: doc.update(aliases=[]), "non-empty"),
    ],
)
def test_invalid_seed_is_rejected(mutate, message) -> None:
    document = seed_document()
    mutate(document)
    with pytest.raises(LegacyTopicAliasError, match=message):
        parse_legacy_topic_aliases(document)


# --- RM_SLICE_TOPICSPINE ---------------------------------------------------


def test_flag_defaults_off_and_reads_the_documented_env_name(monkeypatch) -> None:
    monkeypatch.delenv("RM_SLICE_TOPICSPINE", raising=False)
    assert RuntimeSettings().rm_slice_topicspine == "off"
    assert RuntimeSettings.from_env().rm_slice_topicspine == "off"
    monkeypatch.setenv("RM_SLICE_TOPICSPINE", " Shadow ")
    assert RuntimeSettings.from_env().rm_slice_topicspine == "shadow"
    monkeypatch.setenv("RM_SLICE_TOPICSPINE", "true")
    with pytest.raises(ValueError, match="RM_SLICE_TOPICSPINE"):
        RuntimeSettings.from_env()
    with pytest.raises(ValueError, match="RM_SLICE_TOPICSPINE"):
        RuntimeSettings(rm_slice_topicspine="maybe")


def test_service_never_serves_aliases_while_off() -> None:
    service = RepositoryNavigationService(
        client=None,
        store=InMemoryPortStore(),
        legacy_topic_aliases=load_legacy_topic_aliases(),
    )
    assert service.topic_spine_mode == "off"
    assert service.legacy_topic_aliases is None
    with pytest.raises(ValueError, match="alias set"):
        RepositoryNavigationService(client=None, store=InMemoryPortStore(), topic_spine_mode="on")


def test_flag_off_leaves_navigation_output_unchanged() -> None:
    with runtime(None) as default, runtime("off") as off:
        responses = [client.get(REPOSITORIES) for client in (default, off)]
        ready = off.get("/readyz").json()["checks"]

    for response in responses:
        assert response.status_code == 200
        assert response.json()["legacy_topic_aliases"] == []
        assert "x-rm-slice-topicspine" not in response.headers
        assert "x-legacy-topic-alias-set" not in response.headers
    strip = lambda body: {k: v for k, v in body.items() if k not in {"rebuilt_at", "source_checked_at"}}
    assert strip(responses[0].json()) == strip(responses[1].json())
    assert "rm_slice_topicspine" not in ready


@pytest.mark.parametrize("mode", ["shadow", "on"])
def test_flag_shadow_or_on_serves_contract_valid_aliases(mode: str) -> None:
    with runtime(mode) as client:
        listing = client.get(REPOSITORIES)
        facet = client.get(
            "/ports/projections/repository-navigation/facets/primary_domain/infotech"
        )
        ready = client.get("/readyz").json()["checks"]

    for response in (listing, facet):
        assert response.status_code == 200
        output_validator().validate(response.json())
        assert response.headers["x-rm-slice-topicspine"] == mode
        assert response.headers["x-legacy-topic-alias-set"] == "2026-10-13.1"
        assert len(response.json()["legacy_topic_aliases"]) == 16
    # Classification rows are untouched by the alias metadata.
    assert [row["slug"] for row in listing.json()["repositories"]] == ["hub-core", "repo-manager"]
    assert ready["rm_slice_topicspine"] == mode


# --- A5 comparison cases run locally against fixtures -----------------------


def test_a5_c01_topic_list_maps_to_aliases() -> None:
    with runtime("shadow") as client:
        aliases = client.get(REPOSITORIES, params={"limit": 500}).json()["legacy_topic_aliases"]

    assert as_legacy_rows(aliases) == expected_rows()
    assert {alias["facet_kind"] for alias in aliases} == {"primary_domain"}
    assert all(alias["expires_at"] and alias["residual_owner"] for alias in aliases)


def test_a5_c02_and_c08_each_topic_id_resolves_to_one_alias() -> None:
    with runtime("shadow") as client:
        aliases = client.get(REPOSITORIES, params={"limit": 500}).json()["legacy_topic_aliases"]

    for topic in legacy_topics():
        matches = [alias for alias in aliases if alias["legacy_topic_id"] == topic["id"]]
        assert len(matches) == 1
        assert matches[0]["facet_value"] == topic["domain_slug"]
    custodian = next(a for a in aliases if a["legacy_topic_id"] == "cee7bedf-2b48-46ef-8601-006474f2ad7a")
    assert (custodian["legacy_topic_slug"], custodian["facet_value"]) == ("custodian", "infotech")


def test_a5_c03_unknown_facet_value_is_rejected_not_empty() -> None:
    # Packet expected 200 + zero repositories (unverified); the runtime
    # validates the controlled domain vocabulary and answers 400 instead.
    with runtime("shadow") as client:
        unknown = client.get(
            "/ports/projections/repository-navigation/facets/primary_domain/no-such-domain"
        )
        empty = client.get(
            "/ports/projections/repository-navigation/facets/primary_domain/space"
        )

    assert unknown.status_code == 400
    assert empty.status_code == 200
    assert empty.json()["repositories"] == []


def test_a5_c10_rollback_to_off_withdraws_aliases() -> None:
    with runtime("on") as switched:
        assert len(switched.get(REPOSITORIES).json()["legacy_topic_aliases"]) == 16
    with runtime("off") as rolled_back:
        response = rolled_back.get(REPOSITORIES)
    assert response.json()["legacy_topic_aliases"] == []
    assert "x-rm-slice-topicspine" not in response.headers