""" Tests for INDL-51 — Settings: Plot Type Management (real section linking, validation, delete guard). Covers: Create with valid data + section_ids succeeds, response includes sections/ section_ids/in_use Create with duplicate name (case-insensitive) -> 409 Create with invalid capacity/dimensions/price -> 422 Create with empty section_ids -> 422 Create with a section_id belonging to a different tenant -> rejected Update duplicate name (against a different plot type) -> 409; update with same name as itself succeeds Delete with 0 referencing plots succeeds (204) Delete with >=1 referencing plot -> 409, plot type still exists after in_use count in list response reflects actual plot count Migration fragment-matching helper: case-insensitive match + unmatched log """ import logging from uuid import uuid4 import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from src.apps.tenants.models.account import Account from src.apps.auth.models.user import User from src.apps.sections.models.section import Section from src.apps.plots.models.plot import Plot from src.apps.plots.models.plot_type import PlotType from src.core.security import hash_password, create_access_token, build_token_payload # ── helpers ─────────────────────────────────────────────────────────────────── async def _make_account(db: AsyncSession, **kwargs) -> Account: uid = str(uuid4())[-8:] acc = Account( organization_name=kwargs.pop("org_name", f"PlotType Cemetery {uid}"), subdomain=f"plottype-{uid}", contact_email=f"admin-{uid}@test.ca", plan="professional", status="active", **kwargs, ) db.add(acc) await db.flush() return acc async def _make_user(db: AsyncSession, account: Account, *, role: str = "administrator") -> str: user = User( tenant_id=account.id, email=f"user-{role}-{uuid4().hex[:6]}@test.ca", password_hash=hash_password("Test1234!"), first_name="Test", last_name="User", role=role, status="active", ) db.add(user) await db.flush() return create_access_token(build_token_payload(user, account)) async def _make_section(db: AsyncSession, account: Account, *, code: str = "A", name: str = "North Lawn") -> Section: sec = Section( tenant_id=account.id, code=code, name=name, plot_number_prefix=code, next_plot_seq=1, ) db.add(sec) await db.flush() return sec async def _make_plot_type( db: AsyncSession, account: Account, *, name: str = "Standard single", sections=None ) -> PlotType: pt = PlotType( tenant_id=account.id, name=name, default_width_m=1.2, default_length_m=2.4, default_depth_m=1.8, capacity=1, default_price=4200, sections=sections or [], ) db.add(pt) await db.flush() return pt async def _make_plot(db: AsyncSession, account: Account, plot_type: PlotType, *, ref: str) -> Plot: plot = Plot(tenant_id=account.id, plot_ref=ref, plot_type_id=plot_type.id, status="vacant") db.add(plot) await db.flush() return plot def _headers(token: str, account: Account) -> dict: return {"Authorization": f"Bearer {token}", "X-Tenant-ID": str(account.id)} def _valid_payload(section_ids, **overrides) -> dict: payload = { "name": "Cremation bench", "capacity": 2, "default_length_m": "2.4", "default_width_m": "1.2", "default_depth_m": "1.8", "default_price": "4200.00", "section_ids": [str(sid) for sid in section_ids], } payload.update(overrides) return payload # ── create ─────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_create_plot_type_valid(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) sec = await _make_section(db_session, acc) token = await _make_user(db_session, acc) resp = await client.post( "/api/v1/settings/plot-types", json=_valid_payload([sec.id]), headers=_headers(token, acc), ) assert resp.status_code == 201, resp.text data = resp.json()["data"] assert data["name"] == "Cremation bench" assert data["section_ids"] == [str(sec.id)] assert data["sections"] == [{"id": str(sec.id), "code": sec.code, "name": sec.name}] assert data["in_use"] == 0 @pytest.mark.asyncio async def test_create_plot_type_duplicate_name_case_insensitive(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) sec = await _make_section(db_session, acc) await _make_plot_type(db_session, acc, name="Standard Single") token = await _make_user(db_session, acc) resp = await client.post( "/api/v1/settings/plot-types", json=_valid_payload([sec.id], name="standard single"), headers=_headers(token, acc), ) assert resp.status_code == 409 @pytest.mark.parametrize("field,value", [ ("capacity", 0), ("capacity", 21), ("default_length_m", "0"), ("default_width_m", "0"), ("default_depth_m", "0"), ("default_price", "-1"), ]) @pytest.mark.asyncio async def test_create_plot_type_invalid_ranges(client: AsyncClient, db_session: AsyncSession, field, value): acc = await _make_account(db_session) sec = await _make_section(db_session, acc) token = await _make_user(db_session, acc) resp = await client.post( "/api/v1/settings/plot-types", json=_valid_payload([sec.id], **{field: value}), headers=_headers(token, acc), ) assert resp.status_code == 422 @pytest.mark.asyncio async def test_create_plot_type_empty_section_ids(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) token = await _make_user(db_session, acc) resp = await client.post( "/api/v1/settings/plot-types", json=_valid_payload([]), headers=_headers(token, acc), ) assert resp.status_code == 422 @pytest.mark.asyncio async def test_create_plot_type_cross_tenant_section_id_rejected(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) other_acc = await _make_account(db_session) other_sec = await _make_section(db_session, other_acc) token = await _make_user(db_session, acc) resp = await client.post( "/api/v1/settings/plot-types", json=_valid_payload([other_sec.id]), headers=_headers(token, acc), ) # Cross-tenant section reference — rejected as an invalid/unresolvable # reference (422), matching this codebase's ValidationError convention. assert resp.status_code == 422 # ── update ─────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_update_plot_type_duplicate_name_rejected(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) sec = await _make_section(db_session, acc) await _make_plot_type(db_session, acc, name="Standard single", sections=[sec]) pt2 = await _make_plot_type(db_session, acc, name="Companion double", sections=[sec]) token = await _make_user(db_session, acc) resp = await client.patch( f"/api/v1/settings/plot-types/{pt2.id}", json={"name": "standard single"}, headers=_headers(token, acc), ) assert resp.status_code == 409 @pytest.mark.asyncio async def test_update_plot_type_same_name_as_self_succeeds(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) sec = await _make_section(db_session, acc) pt = await _make_plot_type(db_session, acc, name="Standard single", sections=[sec]) token = await _make_user(db_session, acc) resp = await client.patch( f"/api/v1/settings/plot-types/{pt.id}", json={"name": "Standard single", "default_price": "5000.00"}, headers=_headers(token, acc), ) assert resp.status_code == 200, resp.text assert resp.json()["data"]["default_price"] == 5000.0 # ── delete ─────────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_delete_plot_type_not_in_use_succeeds(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) pt = await _make_plot_type(db_session, acc) token = await _make_user(db_session, acc) resp = await client.delete(f"/api/v1/settings/plot-types/{pt.id}", headers=_headers(token, acc)) assert resp.status_code == 204 # The test `client` fixture overrides get_db to reuse `db_session` directly # (no per-request commit like production's get_db), so the router's # `db.delete(pt)` is only pending until flushed here. await db_session.flush() check = await db_session.get(PlotType, pt.id) assert check is None @pytest.mark.asyncio async def test_delete_plot_type_in_use_blocked(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) pt = await _make_plot_type(db_session, acc) await _make_plot(db_session, acc, pt, ref="P-001") token = await _make_user(db_session, acc) resp = await client.delete(f"/api/v1/settings/plot-types/{pt.id}", headers=_headers(token, acc)) assert resp.status_code == 409 check = await db_session.get(PlotType, pt.id) assert check is not None # ── list / in_use ──────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_plot_types_in_use_count(client: AsyncClient, db_session: AsyncSession): acc = await _make_account(db_session) pt = await _make_plot_type(db_session, acc) for i in range(3): await _make_plot(db_session, acc, pt, ref=f"P-{i}") token = await _make_user(db_session, acc) resp = await client.get("/api/v1/settings/plot-types", headers=_headers(token, acc)) assert resp.status_code == 200 items = resp.json()["data"] row = next(i for i in items if i["id"] == str(pt.id)) assert row["in_use"] == 3 # ── data-migration fragment matching ──────────────────────────────────────── def test_split_fragments_basic(): from importlib import import_module mod = import_module( "src.database.migrations.versions.0064_plot_type_sections_link" ) assert mod._split_fragments("A, b , C") == ["A", "b", "C"] assert mod._split_fragments("") == [] assert mod._split_fragments(None) == [] def test_fragment_matching_case_insensitive_and_unmatched_logged(caplog): """ Exercises the same case-insensitive code/name matching logic the 0064 data migration uses, against an in-memory model of a tenant's sections (A, B) and free-text fragments "A, b , C" — "C" has no match and must be logged as a warning, not raise. """ from importlib import import_module mod = import_module( "src.database.migrations.versions.0064_plot_type_sections_link" ) sections = [ {"id": "sec-a", "code": "A", "name": "North Lawn"}, {"id": "sec-b", "code": "B", "name": "Garden Court"}, ] def find_section(fragment: str): for s in sections: if s["code"].lower() == fragment.lower() or s["name"].lower() == fragment.lower(): return s return None fragments = mod._split_fragments("A, b , C") matched = [] unmatched = [] logger = logging.getLogger("test.fragment_matching") for frag in fragments: section = find_section(frag) if section: matched.append(section["id"]) else: unmatched.append(frag) logger.warning("fragment %r did not match any section", frag) assert matched == ["sec-a", "sec-b"] assert unmatched == ["C"]