"""Datasource field rules on every write path (app/crud/field_constraints.py): default values, unique values and auto-numbers.""" import io import json import uuid from datetime import datetime, timezone import pytest from pydantic import ValidationError from app.crud.field_constraints import format_autonumber, resolve_default from app.schemas.master import MasterFieldDef FIELDS = [ {"name": "code", "type": "text", "unique": True}, {"name": "title", "type": "text", "required": True}, {"name": "status", "type": "text", "default_value": "Open"}, {"name": "qty", "type": "integer", "default_value": "1"}, {"name": "opened_on", "type": "date", "default_value": "{{today}}"}, {"name": "ticket_no", "type": "autonumber", "autonumber_format": "TCK-{YYYY}-{SEQ:4}"}, ] async def _master(client, auth_headers, fields=FIELDS): resp = await client.post("/masters", headers=auth_headers, json={"name": f"Tickets {uuid.uuid4().hex[:6]}", "field_schema": fields}) assert resp.status_code == 201, resp.text return resp.json()["id"] async def test_defaults_unique_and_autonumbers_on_the_record_routes(client, auth_headers): master_id = await _master(client, auth_headers) url = f"/masters/{master_id}/records" year = datetime.now(timezone.utc).year first = await client.post(url, headers=auth_headers, json={"field_values": {"code": "A-1", "title": "Leak", "ticket_no": "HACKED"}}) assert first.status_code == 201, first.text v = first.json()["field_values"] assert v["status"] == "Open" and v["qty"] == 1 and len(v["opened_on"]) == 10 assert v["ticket_no"] == f"TCK-{year}-0001" # given out, not taken from the caller second = await client.post(url, headers=auth_headers, json={"field_values": {"code": "A-2", "title": "Door", "status": "Closed"}}) assert second.json()["field_values"]["ticket_no"] == f"TCK-{year}-0002" assert second.json()["field_values"]["status"] == "Closed" # a value given wins over the default # Unique: case and spaces don't make a value different. clash = await client.post(url, headers=auth_headers, json={"field_values": {"code": " a-1 ", "title": "Again"}}) assert clash.status_code == 422 and "must be unique" in str(clash.json()["detail"]) # An edit keeps its own value, can't take another's, and can't change its number. rid = second.json()["id"] ok = await client.patch(f"{url}/{rid}", headers=auth_headers, json={"field_values": {"code": "A-2", "title": "Door 2", "ticket_no": "X"}}) assert ok.status_code == 200, ok.text assert ok.json()["field_values"]["ticket_no"] == f"TCK-{year}-0002" taken = await client.patch(f"{url}/{rid}", headers=auth_headers, json={"field_values": {"code": "A-1", "title": "Door"}}) assert taken.status_code == 422 # A refused record uses no number. third = await client.post(url, headers=auth_headers, json={"field_values": {"code": "A-3", "title": "Roof"}}) assert third.json()["field_values"]["ticket_no"] == f"TCK-{year}-0003" async def test_import_checks_unique_values_within_the_file_too(client, auth_headers): master_id = await _master(client, auth_headers) await client.post(f"/masters/{master_id}/records", headers=auth_headers, json={"field_values": {"code": "A-1", "title": "Existing"}}) csv = "Code,Title\nA-1,Clashes with a record\nB-1,Fine\nb-1,Clashes with the row above\nC-1,Fine\n" resp = await client.post( f"/masters/{master_id}/bulk-import", headers=auth_headers, files={"file": ("tickets.csv", io.BytesIO(csv.encode()), "text/csv")}, data={"column_mapping": json.dumps({"Code": "code", "Title": "title"}), "commit": "true"}, ) assert resp.status_code == 200, resp.text result = resp.json() assert result["valid_rows"] == 2 assert sorted(e["row"] for e in result["row_errors"]) == [1, 3] rows = (await client.get(f"/masters/{master_id}/records", headers=auth_headers)).json() numbers = sorted(r["field_values"]["ticket_no"] for r in rows) assert len(numbers) == 3 and len(set(numbers)) == 3 # the imported rows got numbers too assert all(r["field_values"]["status"] == "Open" for r in rows) async def test_a_task_data_step_follows_the_rules(test_tenant): from app.crud.tasks import _exec_data_update from app.models.master import Master from tests.conftest import AsyncSessionLocal tenant_id = test_tenant["tenant_id"] async with AsyncSessionLocal() as db: async with db.begin(): master = Master(tenant_id=tenant_id, name=f"T {uuid.uuid4().hex[:6]}", field_schema=FIELDS) db.add(master) await db.flush() node = {"datasource_type": "master", "master_id": str(master.id), "input_params": []} ok, detail, _ = await _exec_data_update(db, tenant_id, node, {"code": "Z-1", "title": "From a task"}) assert ok, detail ok, detail, _ = await _exec_data_update(db, tenant_id, node, {"code": "z-1", "title": "Twice"}) assert not ok and "must be unique" in detail def test_formats_and_defaults(): now = datetime(2026, 10, 23, 9, 30) assert format_autonumber("APP/{YY}{MM}/{SEQ:5}", 42, now) == "APP/2610/00042" assert format_autonumber(None, 7, now) == "7" assert resolve_default({"type": "number", "default_value": "2.5"}, now, None) == 2.5 assert resolve_default({"type": "boolean", "default_value": "yes"}, now, None) is True assert resolve_default({"type": "text", "default_value": "{{now}}"}, now, None) == "2026-10-23T09:30:00" class Person: name, email, id = "Asha", "asha@example.com", "u1" assert resolve_default({"type": "text", "default_value": "{{current_user.name}}"}, now, Person()) == "Asha" assert resolve_default({"type": "text", "default_value": "{{current_user.name}}"}, now, None) is None def test_field_definition_rules(): assert MasterFieldDef(name="n", type="autonumber").autonumber_format == "{SEQ}" for bad in [ {"name": "n", "type": "autonumber", "autonumber_format": "APP-{YYYY}"}, # no running number {"name": "n", "type": "autonumber", "required": True}, {"name": "n", "type": "text", "autonumber_format": "{SEQ}"}, {"name": "n", "type": "formula", "formula_expression": "1", "unique": True}, {"name": "n", "type": "json", "unique": True}, ]: with pytest.raises(ValidationError): MasterFieldDef(**bad)