fitguide/tests/test_danger_zone.py

import pytest

from app.crud import audit as audit_crud
from app.models.attachment_point import AttachmentPoint
from app.models.attachment_point_name import AttachmentPointName
from app.models.augmented_part import AugmentedPart, AugmentedPartNote
from app.models.clamp_connection import ClampConnection
from app.models.fit import Fit, FitLabel, FitNote
from app.models.imported_part import ImportedPart
from app.models.imported_part_config import ImportedPartConfig
from app.models.iso_thread import IsoThread
from app.models.manufacturer import Manufacturer
from app.models.part_colour import PartColour
from app.models.part_type import PartType
from app.models.scope_info import ScopeInfo
from app.models.scope_model import ScopeModel
from app.models.static_connection import StaticConnection
from app.models.sub_part_type import SubPartType
from app.models.unc_thread import UncSeries, UncThread


async def _seed_domain_data(db):
    manufacturer = Manufacturer(name="Acme", url="https://acme.test")
    colour = PartColour(name="Black")
    scope_t = PartType(name="Scope")
    cover_t = PartType(name="Flip Cover")
    apn = AttachmentPointName(name="Objective")
    db.add_all([manufacturer, colour, scope_t, cover_t, apn])
    await db.flush()

    sub_t = SubPartType(name="Riflescope", part_type_id=scope_t.id)
    scope_model = ScopeModel(name="Alpha", manufacturer_id=manufacturer.id)
    iso = IsoThread(diameter_mm=52.0, pitch_mm=0.75)
    unc = UncThread(diameter_fraction="1/4", tpi=20, series=UncSeries.UNC)
    clamp = ClampConnection(min_mm=25.0, max_mm=30.0)
    static = StaticConnection(od_mm=50.0, od_in=1.97)
    db.add_all([sub_t, scope_model, iso, unc, clamp, static])
    await db.flush()

    scope = AugmentedPart(
        description="Alpha 3-18x50",
        sku="ALP-318",
        colour_id=colour.id,
        manufacturer_id=manufacturer.id,
        part_type_id=scope_t.id,
        scope_model_id=scope_model.id,
        sub_part_type_id=sub_t.id,
    )
    cover = AugmentedPart(
        description="Flip cover 50mm",
        colour_id=colour.id,
        manufacturer_id=manufacturer.id,
        part_type_id=cover_t.id,
    )
    db.add_all([scope, cover])
    await db.flush()

    db.add(
        AugmentedPartNote(augmented_part_id=scope.id, username="tester", body="note")
    )
    db.add(
        ScopeInfo(augmented_part_id=scope.id, magnification_min=3, magnification_max=18)
    )

    ap_s = AttachmentPoint(
        augmented_part_id=scope.id,
        attachment_point_name_id=apn.id,
        iso_thread_id=iso.id,
    )
    ap_c = AttachmentPoint(
        augmented_part_id=cover.id,
        attachment_point_name_id=apn.id,
        iso_thread_id=iso.id,
    )
    db.add_all([ap_s, ap_c])
    await db.flush()

    fit = Fit(
        attachment_point_a_id=ap_s.id,
        attachment_point_b_id=ap_c.id,
        augmented_part_a_id=scope.id,
        augmented_part_b_id=cover.id,
        label=FitLabel.POSSIBLE,
    )
    db.add(fit)
    await db.flush()
    db.add(FitNote(fit_id=fit.id, username="tester", body="fits fine"))

    imported = ImportedPart(InventoryID="INV-1", Description="Alpha 3-18x50")
    db.add(imported)

    config = ImportedPartConfig(
        id=1, odata_url="https://example.test/odata", username="svc"
    )
    db.add(config)

    await db.commit()


async def _web_login(client, email, password="secret123"):
    resp = await client.post("/login", data={"email": email, "password": password})
    assert resp.status_code == 303


@pytest.mark.asyncio
async def test_non_superuser_forbidden(client, seeded, make_user):
    # "admin" role holds every permission but is not a superuser.
    await make_user("admin@example.com", role_ids=[seeded["roles"]["admin"].id])
    await _web_login(client, "admin@example.com")

    assert (await client.get("/admin/danger-zone")).status_code == 403
    assert (await client.post("/admin/danger-zone/purge")).status_code == 403


@pytest.mark.asyncio
async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
    client, seeded, make_user, db_sessionmaker
):
    async with db_sessionmaker() as db:
        await _seed_domain_data(db)

    await make_user("root@example.com", is_superuser=True)
    await _web_login(client, "root@example.com")

    assert (await client.get("/admin/danger-zone")).status_code == 200

    resp = await client.post("/admin/danger-zone/purge")
    assert resp.status_code == 200
    assert "Purge complete" in resp.text

    async with db_sessionmaker() as db:
        for Model in (
            AugmentedPart,
            AugmentedPartNote,
            ScopeInfo,
            AttachmentPoint,
            Fit,
            FitNote,
            ImportedPart,
            ScopeModel,
            AttachmentPointName,
            IsoThread,
            UncThread,
            ClampConnection,
            StaticConnection,
            Manufacturer,
        ):
            rows = (await db.execute(Model.__table__.select())).all()
            assert rows == [], f"{Model.__name__} should be empty after purge"

        # Kept: users, RBAC, the OData connection config, and part taxonomy.
        from app.crud import user as user_crud

        assert await user_crud.get_by_email(db, "root@example.com") is not None
        config = await db.get(ImportedPartConfig, 1)
        assert config is not None
        assert config.odata_url == "https://example.test/odata"

        for Model in (PartColour, PartType, SubPartType):
            rows = (await db.execute(Model.__table__.select())).all()
            assert rows != [], f"{Model.__name__} should survive the purge"

        entries = await audit_crud.list_recent(db)
        assert any("Purged all domain data" in e.description for e in entries)