diff --git a/app/api/__init__.py b/app/api/__init__.py index 9936d7f4..25344332 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -34,6 +34,7 @@ from app.api.plans import router as plans_router from app.api.process import router as process_router from app.api.profile import router as profile_router from app.api.queue import router as queue_router +from app.api.routing_rules import router as routing_rules_router from app.api.saved_searches import router as saved_searches_router from app.api.scheduled_jobs import router as scheduled_jobs_router from app.api.search import router as search_router @@ -85,6 +86,7 @@ router.include_router(onboarding_router) router.include_router(billing_router) router.include_router(pipelines_router) router.include_router(profile_router) +router.include_router(routing_rules_router) router.include_router(imap_accounts_router) router.include_router(imap_profiles_router) router.include_router(integrations_router) diff --git a/app/api/routing_rules.py b/app/api/routing_rules.py new file mode 100644 index 00000000..98162c09 --- /dev/null +++ b/app/api/routing_rules.py @@ -0,0 +1,468 @@ +"""Routing rules API endpoints. + +Provides full CRUD for pipeline routing rules that conditionally assign +documents to pipelines based on document properties (file type, category, +metadata fields, size, etc.). + +Rules are evaluated in ascending ``position`` order. The first rule whose +condition matches wins and routes the document to the specified target +pipeline. If no rule matches, the caller falls back to the owner's (or +system) default pipeline. +""" + +import logging +from typing import Annotated, Any + +from fastapi import APIRouter, Body, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.auth import require_login +from app.database import get_db +from app.models import Pipeline, PipelineRoutingRule +from app.utils.routing_engine import ( + BUILTIN_FIELDS, + VALID_OPERATORS, + _evaluate_condition, + _resolve_field, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/routing-rules", tags=["routing-rules"]) + +DbSession = Annotated[Session, Depends(get_db)] + +MAX_RULES_PER_OWNER = 100 +MAX_NAME_LENGTH = 255 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _get_user_id(request: Request) -> str: + """Return the authenticated user identifier.""" + user = getattr(request.state, "user", None) + if user: + if isinstance(user, dict): + return user.get("sub", user.get("email", "anonymous")) + return getattr(user, "sub", getattr(user, "email", "anonymous")) + return "anonymous" + + +def _is_admin(request: Request) -> bool: + """Return ``True`` when the current user has admin privileges.""" + user = getattr(request.state, "user", None) + if not user: + return False + groups = user.get("groups", []) if isinstance(user, dict) else getattr(user, "groups", []) + return "admin" in groups + + +def _can_access_rule(rule: PipelineRoutingRule, user_id: str, admin: bool) -> bool: + """Check whether the user is allowed to read this rule.""" + if admin: + return True + return rule.owner_id == user_id + + +def _can_write_rule(rule: PipelineRoutingRule, user_id: str, admin: bool) -> bool: + """Check whether the user is allowed to modify this rule.""" + if rule.owner_id is None: + return admin + return rule.owner_id == user_id + + +def _validate_field(field: str) -> None: + """Raise 422 if the field name is invalid.""" + if field in BUILTIN_FIELDS: + return + if field.startswith("metadata.") and len(field) > len("metadata."): + return + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=( + f"Invalid field '{field}'. " + f"Valid built-in fields: {sorted(BUILTIN_FIELDS)}. " + "For AI metadata, use 'metadata.'." + ), + ) + + +def _validate_operator(operator: str) -> None: + """Raise 422 if the operator is not recognised.""" + if operator not in VALID_OPERATORS: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Invalid operator '{operator}'. Valid operators: {sorted(VALID_OPERATORS)}", + ) + + +def _serialize_rule(rule: PipelineRoutingRule) -> dict[str, Any]: + """Serialize a routing rule to a JSON-compatible dict.""" + return { + "id": rule.id, + "owner_id": rule.owner_id, + "name": rule.name, + "position": rule.position, + "field": rule.field, + "operator": rule.operator, + "value": rule.value, + "target_pipeline_id": rule.target_pipeline_id, + "is_active": rule.is_active, + "created_at": rule.created_at.isoformat() if rule.created_at else None, + "updated_at": rule.updated_at.isoformat() if rule.updated_at else None, + } + + +# --------------------------------------------------------------------------- +# Pydantic request models +# --------------------------------------------------------------------------- + + +class RoutingRuleCreate(BaseModel): + """Request body for creating a routing rule.""" + + name: str = Field(..., min_length=1, max_length=MAX_NAME_LENGTH) + field: str = Field(..., min_length=1, max_length=255) + operator: str = Field(..., min_length=1, max_length=50) + value: str = Field(..., max_length=1024) + target_pipeline_id: int + position: int | None = None + is_active: bool = True + + +class RoutingRuleUpdate(BaseModel): + """Request body for updating a routing rule.""" + + name: str | None = Field(None, min_length=1, max_length=MAX_NAME_LENGTH) + field: str | None = Field(None, min_length=1, max_length=255) + operator: str | None = Field(None, min_length=1, max_length=50) + value: str | None = Field(None, max_length=1024) + target_pipeline_id: int | None = None + position: int | None = None + is_active: bool | None = None + + +class RoutingRuleEvaluateRequest(BaseModel): + """Request body for dry-run rule evaluation.""" + + file_type: str | None = None + filename: str | None = None + size: int | None = None + document_type: str | None = None + metadata: dict[str, Any] | None = None + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.get("") +@require_login +def list_routing_rules(request: Request, db: DbSession) -> list[dict[str, Any]]: + """List all routing rules accessible by the current user. + + Returns the user's own rules plus any system-wide rules (``owner_id=NULL``). + Rules are sorted by position. + """ + user_id = _get_user_id(request) + + rules = ( + db.query(PipelineRoutingRule) + .filter((PipelineRoutingRule.owner_id == user_id) | (PipelineRoutingRule.owner_id.is_(None))) + .order_by( + PipelineRoutingRule.owner_id.is_(None).asc(), + PipelineRoutingRule.position.asc(), + ) + .all() + ) + + return [_serialize_rule(r) for r in rules] + + +@router.post("", status_code=status.HTTP_201_CREATED) +@require_login +def create_routing_rule(request: Request, db: DbSession, body: RoutingRuleCreate) -> dict[str, Any]: + """Create a new routing rule for the current user. + + Returns: + The created routing rule. + + Raises: + HTTPException 422: If the field or operator is invalid. + HTTPException 404: If the target pipeline does not exist. + HTTPException 409: If the maximum number of rules is reached. + """ + user_id = _get_user_id(request) + + _validate_field(body.field) + _validate_operator(body.operator) + + # Verify target pipeline exists and is accessible. + pipeline = db.query(Pipeline).filter(Pipeline.id == body.target_pipeline_id).first() + if not pipeline: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Target pipeline {body.target_pipeline_id} not found", + ) + + # Enforce per-owner limit. + count = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.owner_id == user_id).count() + if count >= MAX_RULES_PER_OWNER: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Maximum of {MAX_RULES_PER_OWNER} routing rules per user reached", + ) + + # Auto-assign position if not specified. + position = body.position + if position is None: + max_pos = ( + db.query(PipelineRoutingRule.position) + .filter(PipelineRoutingRule.owner_id == user_id) + .order_by(PipelineRoutingRule.position.desc()) + .first() + ) + position = (max_pos[0] + 1) if max_pos else 0 + + rule = PipelineRoutingRule( + owner_id=user_id, + name=body.name.strip(), + position=position, + field=body.field, + operator=body.operator, + value=body.value, + target_pipeline_id=body.target_pipeline_id, + is_active=body.is_active, + ) + + try: + db.add(rule) + db.commit() + db.refresh(rule) + except Exception: + db.rollback() + logger.exception("Failed to create routing rule for user=%s", user_id) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to create routing rule", + ) + + logger.info("Routing rule created: id=%s, user=%s", rule.id, user_id) + return _serialize_rule(rule) + + +@router.get("/operators") +def list_operators() -> dict[str, Any]: + """Return the list of supported operators and fields. + + This is a public endpoint (no auth required) so that UIs can populate + dropdowns without hard-coding the catalogue. + """ + return { + "operators": sorted(VALID_OPERATORS), + "builtin_fields": sorted(BUILTIN_FIELDS), + "metadata_prefix": "metadata.", + } + + +@router.post("/evaluate") +@require_login +def evaluate_rules(request: Request, db: DbSession, body: RoutingRuleEvaluateRequest) -> dict[str, Any]: + """Dry-run rule evaluation against the provided document properties. + + Returns the first matching rule and target pipeline (if any), or + indicates that no rule matched (default pipeline will be used). + """ + user_id = _get_user_id(request) + + doc_props: dict[str, Any] = { + "file_type": body.file_type, + "filename": body.filename, + "size": body.size, + "document_type": body.document_type, + "metadata": body.metadata or {}, + } + + rules = ( + db.query(PipelineRoutingRule) + .filter( + PipelineRoutingRule.is_active.is_(True), + (PipelineRoutingRule.owner_id == user_id) | (PipelineRoutingRule.owner_id.is_(None)), + ) + .order_by( + PipelineRoutingRule.owner_id.is_(None).asc(), + PipelineRoutingRule.position.asc(), + ) + .all() + ) + + for rule in rules: + actual = _resolve_field(rule.field, doc_props) + if _evaluate_condition(actual, rule.operator, rule.value): + pipeline = db.query(Pipeline).filter(Pipeline.id == rule.target_pipeline_id).first() + return { + "matched": True, + "rule": _serialize_rule(rule), + "target_pipeline": { + "id": pipeline.id, + "name": pipeline.name, + "is_active": pipeline.is_active, + } + if pipeline + else None, + } + + return {"matched": False, "rule": None, "target_pipeline": None} + + +@router.put("/reorder") +@require_login +def reorder_routing_rules( + request: Request, + db: DbSession, + rule_ids: list[int] = Body(..., embed=True), +) -> list[dict[str, Any]]: + """Reorder the caller's routing rules. + + Expects a JSON body ``{"rule_ids": [3, 1, 2]}`` where the list + contains the IDs of the caller's rules in the desired order. + """ + user_id = _get_user_id(request) + + rules = ( + db.query(PipelineRoutingRule) + .filter(PipelineRoutingRule.owner_id == user_id, PipelineRoutingRule.id.in_(rule_ids)) + .all() + ) + + rule_map = {r.id: r for r in rules} + + if len(rule_map) != len(rule_ids) or set(rule_map.keys()) != set(rule_ids): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="rule_ids must contain exactly the IDs of your routing rules", + ) + + for pos, rid in enumerate(rule_ids): + rule_map[rid].position = pos + + try: + db.commit() + except Exception: + db.rollback() + logger.exception("Failed to reorder routing rules for user=%s", user_id) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to reorder routing rules", + ) + + ordered = sorted(rules, key=lambda r: r.position) + return [_serialize_rule(r) for r in ordered] + + +@router.get("/{rule_id}") +@require_login +def get_routing_rule(rule_id: int, request: Request, db: DbSession) -> dict[str, Any]: + """Return a single routing rule by ID.""" + user_id = _get_user_id(request) + admin = _is_admin(request) + + rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first() + if not rule or not _can_access_rule(rule, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found") + + return _serialize_rule(rule) + + +@router.put("/{rule_id}") +@require_login +def update_routing_rule(rule_id: int, request: Request, db: DbSession, body: RoutingRuleUpdate) -> dict[str, Any]: + """Update a routing rule. + + Only the fields present in the request body are updated. + """ + user_id = _get_user_id(request) + admin = _is_admin(request) + + rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first() + if not rule or not _can_access_rule(rule, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found") + + if not _can_write_rule(rule, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this rule") + + if body.field is not None: + _validate_field(body.field) + rule.field = body.field + + if body.operator is not None: + _validate_operator(body.operator) + rule.operator = body.operator + + if body.value is not None: + rule.value = body.value + + if body.name is not None: + rule.name = body.name.strip() + + if body.target_pipeline_id is not None: + pipeline = db.query(Pipeline).filter(Pipeline.id == body.target_pipeline_id).first() + if not pipeline: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Target pipeline {body.target_pipeline_id} not found", + ) + rule.target_pipeline_id = body.target_pipeline_id + + if body.position is not None: + rule.position = body.position + + if body.is_active is not None: + rule.is_active = body.is_active + + try: + db.commit() + db.refresh(rule) + except Exception: + db.rollback() + logger.exception("Failed to update routing rule id=%s", rule_id) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to update routing rule", + ) + + logger.info("Routing rule updated: id=%s, user=%s", rule_id, user_id) + return _serialize_rule(rule) + + +@router.delete("/{rule_id}", status_code=status.HTTP_204_NO_CONTENT) +@require_login +def delete_routing_rule(rule_id: int, request: Request, db: DbSession) -> None: + """Delete a routing rule.""" + user_id = _get_user_id(request) + admin = _is_admin(request) + + rule = db.query(PipelineRoutingRule).filter(PipelineRoutingRule.id == rule_id).first() + if not rule or not _can_access_rule(rule, user_id, admin): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Routing rule not found") + + if not _can_write_rule(rule, user_id, admin): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot modify this rule") + + try: + db.delete(rule) + db.commit() + except Exception: + db.rollback() + logger.exception("Failed to delete routing rule id=%s", rule_id) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to delete routing rule", + ) + + logger.info("Routing rule deleted: id=%s, user=%s", rule_id, user_id) diff --git a/app/models.py b/app/models.py index bb1ee985..b752fdb6 100644 --- a/app/models.py +++ b/app/models.py @@ -7,6 +7,7 @@ from app.database import Base # Foreign key constants _FILES_ID_FK = "files.id" _PIPELINES_ID_FK = "pipelines.id" +_ROUTING_RULES_TABLE = "pipeline_routing_rules" class DocumentMetadata(Base): @@ -970,3 +971,53 @@ class ComplianceTemplate(Base): applied_by = Column(String(255), nullable=True) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class PipelineRoutingRule(Base): + """Conditional routing rule that assigns documents to pipelines. + + Rules are evaluated in ascending ``position`` order for a given owner. + The first rule whose condition matches the document properties wins and + the document is routed to ``target_pipeline_id``. If no rule matches, + the caller falls back to the owner's (or system) default pipeline. + + Supported fields: + file_type, document_type, category, filename, size, and any key + inside the AI-extracted metadata JSON (prefixed ``metadata.``). + + Supported operators: + equals, not_equals, contains, not_contains, regex, gt, lt, gte, lte. + """ + + __tablename__ = _ROUTING_RULES_TABLE + + id = Column(Integer, primary_key=True, index=True) + + # Owner of this rule. NULL = system-wide rule (admin only). + owner_id = Column(String, nullable=True, index=True) + + # Human-readable label for the rule. + name = Column(String(255), nullable=False) + + # Evaluation order (lower = earlier). First matching rule wins. + position = Column(Integer, nullable=False, default=0) + + # The document property to evaluate. + # Built-in: file_type, document_type, category, filename, size. + # For AI metadata fields, use the "metadata." prefix. + field = Column(String(255), nullable=False) + + # Comparison operator. + operator = Column(String(50), nullable=False) + + # Value to compare against (always stored as text; cast as needed). + value = Column(String(1024), nullable=False) + + # Target pipeline when the condition matches. + target_pipeline_id = Column(Integer, ForeignKey(_PIPELINES_ID_FK), nullable=False, index=True) + + # Soft-disable without deleting. + is_active = Column(Boolean, nullable=False, default=True) + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/app/utils/routing_engine.py b/app/utils/routing_engine.py new file mode 100644 index 00000000..b65c3696 --- /dev/null +++ b/app/utils/routing_engine.py @@ -0,0 +1,223 @@ +"""Routing engine for conditional document-to-pipeline assignment. + +Evaluates a set of :class:`PipelineRoutingRule` rows against document +properties and returns the first matching target pipeline (if any). + +Supported document fields +------------------------- +* ``file_type`` – MIME type of the file (e.g. ``application/pdf``) +* ``filename`` – original filename +* ``size`` – file size in bytes (numeric comparison) +* ``document_type`` – AI-classified document type (e.g. ``Invoice``) +* ``category`` – alias for ``document_type`` +* ``metadata.`` – arbitrary key inside the AI-extracted JSON metadata + +Supported comparison operators +------------------------------ +* ``equals`` / ``not_equals`` +* ``contains`` / ``not_contains`` (substring match, case-insensitive) +* ``regex`` (Python ``re`` full-match, case-insensitive) +* ``gt`` / ``lt`` / ``gte`` / ``lte`` (numeric comparison) +""" + +import json +import logging +import re +from typing import Any + +from sqlalchemy.orm import Session + +from app.models import Pipeline, PipelineRoutingRule + +logger = logging.getLogger(__name__) + +# Operators recognised by the engine. +VALID_OPERATORS: frozenset[str] = frozenset( + { + "equals", + "not_equals", + "contains", + "not_contains", + "regex", + "gt", + "lt", + "gte", + "lte", + } +) + +# Fields that are resolved directly from the FileRecord. +BUILTIN_FIELDS: frozenset[str] = frozenset( + { + "file_type", + "filename", + "size", + "document_type", + "category", + } +) + + +def _resolve_field(field: str, doc_props: dict[str, Any]) -> Any: + """Resolve a *field* name to its actual value from *doc_props*. + + ``doc_props`` is expected to contain top-level keys that mirror the + built-in field names **plus** a ``metadata`` dict with the parsed + AI metadata JSON. + """ + if field == "category": + # ``category`` is an alias for ``document_type``. + field = "document_type" + + if field.startswith("metadata."): + meta_key = field[len("metadata.") :] + metadata = doc_props.get("metadata") or {} + return metadata.get(meta_key) + + return doc_props.get(field) + + +def _to_float(value: Any) -> float | None: + """Try to convert *value* to a float for numeric comparison.""" + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _evaluate_condition(actual: Any, operator: str, expected: str) -> bool: + """Return ``True`` when *actual* satisfies *operator* against *expected*. + + All string comparisons are case-insensitive. Numeric operators (``gt``, + ``lt``, ``gte``, ``lte``) attempt to cast both sides to ``float``. + """ + if actual is None: + # If the document property is missing, the rule cannot match + # (except for ``not_equals`` / ``not_contains`` which should match). + if operator == "not_equals": + return True + if operator == "not_contains": + return True + return False + + actual_str = str(actual).lower() + expected_lower = expected.lower() + + if operator == "equals": + return actual_str == expected_lower + if operator == "not_equals": + return actual_str != expected_lower + if operator == "contains": + return expected_lower in actual_str + if operator == "not_contains": + return expected_lower not in actual_str + if operator == "regex": + try: + return bool(re.fullmatch(expected, str(actual), flags=re.IGNORECASE)) + except re.error: + logger.warning("Invalid regex in routing rule: %s", expected) + return False + + # Numeric operators + actual_num = _to_float(actual) + expected_num = _to_float(expected) + if actual_num is None or expected_num is None: + return False + + if operator == "gt": + return actual_num > expected_num + if operator == "lt": + return actual_num < expected_num + if operator == "gte": + return actual_num >= expected_num + if operator == "lte": + return actual_num <= expected_num + + return False + + +def build_document_properties(file_record: Any) -> dict[str, Any]: + """Build the property dict that the engine evaluates against. + + Args: + file_record: A :class:`FileRecord` instance (or any object with the + same attributes). + + Returns: + A dict with ``file_type``, ``filename``, ``size``, ``document_type``, + and ``metadata`` keys. + """ + metadata: dict[str, Any] = {} + raw_meta = getattr(file_record, "ai_metadata", None) + if raw_meta: + try: + metadata = json.loads(raw_meta) if isinstance(raw_meta, str) else raw_meta + except (json.JSONDecodeError, TypeError): + metadata = {} + + return { + "file_type": getattr(file_record, "mime_type", None), + "filename": getattr(file_record, "original_filename", None), + "size": getattr(file_record, "file_size", None), + "document_type": metadata.get("document_type"), + "metadata": metadata, + } + + +def evaluate_routing_rules( + db: Session, + owner_id: str | None, + doc_props: dict[str, Any], +) -> Pipeline | None: + """Evaluate routing rules and return the first matching pipeline. + + Rules are fetched for the given *owner_id* **plus** any system-wide rules + (``owner_id IS NULL``). Owner rules are evaluated first (by position), + then system rules. + + Args: + db: Active database session. + owner_id: The document owner's identifier (may be ``None``). + doc_props: Document property dict as produced by + :func:`build_document_properties`. + + Returns: + The first matching :class:`Pipeline`, or ``None`` when no rule + matches (caller should fall back to the default pipeline). + """ + # Fetch active rules for the owner + system rules, ordered by position. + rules = ( + db.query(PipelineRoutingRule) + .filter( + PipelineRoutingRule.is_active.is_(True), + (PipelineRoutingRule.owner_id == owner_id) | (PipelineRoutingRule.owner_id.is_(None)), + ) + .order_by( + # Owner-specific rules take priority over system rules. + PipelineRoutingRule.owner_id.is_(None).asc(), + PipelineRoutingRule.position.asc(), + ) + .all() + ) + + for rule in rules: + actual = _resolve_field(rule.field, doc_props) + if _evaluate_condition(actual, rule.operator, rule.value): + pipeline = db.query(Pipeline).filter(Pipeline.id == rule.target_pipeline_id).first() + if pipeline and pipeline.is_active: + logger.info( + "Routing rule matched: rule_id=%s, name=%s, target_pipeline=%s", + rule.id, + rule.name, + rule.target_pipeline_id, + ) + return pipeline + logger.warning( + "Routing rule %s matched but target pipeline %s is inactive or missing", + rule.id, + rule.target_pipeline_id, + ) + + return None diff --git a/docs/API.md b/docs/API.md index 9515df4c..3d4c7842 100644 --- a/docs/API.md +++ b/docs/API.md @@ -1960,6 +1960,160 @@ Pass no `pipeline_id` query parameter (or omit it) to clear the assignment. ``` +## Routing Rules + +Routing rules let you conditionally assign documents to different pipelines +based on file properties such as type, size, filename, or AI-extracted +metadata. Rules are evaluated in **position order** (lowest first); the first +rule that matches wins. If no rule matches, the system falls back to the +owner's (or global) default pipeline. + +### Supported operators and fields + +```bash +GET /api/routing-rules/operators +``` + +Returns the catalogue of valid operators and built-in fields so UIs can +populate dropdowns without hard-coding values. + +**Response (200):** +```json +{ + "operators": ["contains", "equals", "gt", "gte", "lt", "lte", "not_contains", "not_equals", "regex"], + "builtin_fields": ["category", "document_type", "file_type", "filename", "size"], + "metadata_prefix": "metadata." +} +``` + +> **Tip:** For AI metadata fields use the `metadata.` prefix, e.g. +> `metadata.sender`, `metadata.amount`. + +### List routing rules + +```bash +GET /api/routing-rules +``` + +Returns the current user's rules **plus** any system-wide rules +(`owner_id = null`), ordered by position. + +**Response (200):** +```json +[ + { + "id": 1, + "owner_id": "alice", + "name": "Route invoices", + "position": 0, + "field": "document_type", + "operator": "equals", + "value": "Invoice", + "target_pipeline_id": 3, + "is_active": true, + "created_at": "2026-03-09T12:00:00+00:00", + "updated_at": "2026-03-09T12:00:00+00:00" + } +] +``` + +### Create routing rule + +```bash +POST /api/routing-rules +Content-Type: application/json + +{ + "name": "Route invoices", + "field": "document_type", + "operator": "equals", + "value": "Invoice", + "target_pipeline_id": 3 +} +``` + +Optional fields: `position` (auto-assigned if omitted), `is_active` (default `true`). + +**Response (201 Created):** The created rule object. + +### Get routing rule + +```bash +GET /api/routing-rules/{rule_id} +``` + +**Response (200):** A single rule object. + +### Update routing rule + +```bash +PUT /api/routing-rules/{rule_id} +Content-Type: application/json + +{ "name": "Renamed rule", "operator": "contains", "is_active": false } +``` + +Only the supplied fields are updated. + +**Response (200):** The updated rule object. + +### Delete routing rule + +```bash +DELETE /api/routing-rules/{rule_id} +``` + +Returns **204 No Content**. + +### Reorder routing rules + +```bash +PUT /api/routing-rules/reorder +Content-Type: application/json + +{ "rule_ids": [3, 1, 2] } +``` + +Provide the complete ordered list of your rule IDs. Positions are reassigned +0, 1, 2, … in the given order. + +### Evaluate rules (dry run) + +```bash +POST /api/routing-rules/evaluate +Content-Type: application/json + +{ + "file_type": "application/pdf", + "filename": "invoice_2024.pdf", + "size": 204800, + "document_type": "Invoice", + "metadata": { "sender": "Acme Corp" } +} +``` + +Tests which rule (if any) would match the given properties **without** +actually routing a document. + +**Response (200) – match found:** +```json +{ + "matched": true, + "rule": { "id": 1, "name": "Route invoices", "..." : "..." }, + "target_pipeline": { "id": 3, "name": "Invoice Pipeline", "is_active": true } +} +``` + +**Response (200) – no match:** +```json +{ + "matched": false, + "rule": null, + "target_pipeline": null +} +``` + + ## API Tokens Personal API tokens allow programmatic access to the DocuElevate API without diff --git a/docs/UserGuide.md b/docs/UserGuide.md index 9906a89c..957d005d 100644 --- a/docs/UserGuide.md +++ b/docs/UserGuide.md @@ -626,6 +626,58 @@ Pass no `pipeline_id` to clear the assignment and fall back to the system defaul Admins can create **system pipelines** that appear in every user's pipeline list. These can be set as the global default so all users benefit from a consistent processing baseline. Navigate to **Pipelines** and check the **System pipeline** box when creating a new one (admin only). +### Conditional routing rules + +Routing rules automatically assign incoming documents to the right pipeline +based on their properties — no manual pipeline selection required. + +**How it works:** + +1. Define one or more routing rules via the API + (`POST /api/routing-rules`). +2. Each rule specifies a **field** to inspect, an **operator** (condition), + a **value** to compare against, and a **target pipeline**. +3. When a document is processed, rules are evaluated **in position order** + (lowest first). The first matching rule wins and the document is routed + to that pipeline. +4. If no rule matches, the document is processed by the default pipeline. + +**Available fields:** + +| Field | Description | +|-------|-------------| +| `file_type` | MIME type, e.g. `application/pdf` | +| `filename` | Original filename | +| `size` | File size in bytes | +| `document_type` | AI-classified type (Invoice, Contract, …) | +| `category` | Alias for `document_type` | +| `metadata.` | Any key from the AI-extracted metadata JSON | + +**Available operators:** + +| Operator | Description | +|----------|-------------| +| `equals` / `not_equals` | Exact match (case-insensitive) | +| `contains` / `not_contains` | Substring match (case-insensitive) | +| `regex` | Full Python regex match (case-insensitive) | +| `gt` / `lt` / `gte` / `lte` | Numeric comparison (greater/less than) | + +**Example:** Route invoices to one pipeline and large files to another: + +``` +Rule 1: field=document_type, operator=equals, value=Invoice, target_pipeline=3 +Rule 2: field=size, operator=gt, value=1048576, target_pipeline=5 +``` + +With first-match-wins logic, an invoice of any size matches Rule 1 and is +routed to pipeline 3. A non-invoice file larger than 1 MB matches Rule 2 +and is routed to pipeline 5. Everything else falls back to the default +pipeline. + +You can test your rules without actually routing a document using the +**evaluate** endpoint (`POST /api/routing-rules/evaluate`). For the full +API reference, see [API Documentation](API.md#routing-rules). + ## API Access For programmatic access, DocuElevate provides a comprehensive REST API: diff --git a/migrations/env.py b/migrations/env.py index 5bd09c43..a0bc2f99 100644 --- a/migrations/env.py +++ b/migrations/env.py @@ -33,6 +33,7 @@ from app.models import ( # noqa: F401 LocalUser, MobileDevice, Pipeline, + PipelineRoutingRule, PipelineStep, ProcessingLog, SavedSearch, diff --git a/migrations/versions/035_add_routing_rules.py b/migrations/versions/035_add_routing_rules.py new file mode 100644 index 00000000..d15648ea --- /dev/null +++ b/migrations/versions/035_add_routing_rules.py @@ -0,0 +1,46 @@ +"""Add pipeline_routing_rules table for conditional document routing. + +Revision ID: 035_add_routing_rules +Revises: 034_add_user_profile_settings +Create Date: 2026-03-09 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "035_add_routing_rules" +down_revision: Union[str, None] = "034_add_user_profile_settings" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create pipeline_routing_rules table.""" + op.create_table( + "pipeline_routing_rules", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("owner_id", sa.String(), nullable=True), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("position", sa.Integer(), nullable=False, server_default="0"), + sa.Column("field", sa.String(255), nullable=False), + sa.Column("operator", sa.String(50), nullable=False), + sa.Column("value", sa.String(1024), nullable=False), + sa.Column("target_pipeline_id", sa.Integer(), nullable=False), + sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.PrimaryKeyConstraint("id"), + sa.ForeignKeyConstraint(["target_pipeline_id"], ["pipelines.id"]), + ) + op.create_index("ix_routing_rules_id", "pipeline_routing_rules", ["id"]) + op.create_index("ix_routing_rules_owner_id", "pipeline_routing_rules", ["owner_id"]) + op.create_index("ix_routing_rules_target_pipeline_id", "pipeline_routing_rules", ["target_pipeline_id"]) + + +def downgrade() -> None: + """Drop pipeline_routing_rules table.""" + op.drop_index("ix_routing_rules_target_pipeline_id", "pipeline_routing_rules") + op.drop_index("ix_routing_rules_owner_id", "pipeline_routing_rules") + op.drop_index("ix_routing_rules_id", "pipeline_routing_rules") + op.drop_table("pipeline_routing_rules") diff --git a/tests/conftest.py b/tests/conftest.py index 6befe3d8..fef85bd9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -66,6 +66,7 @@ from app.models import ( # noqa: F401, E402 DocumentMetadata, FileRecord, Pipeline, + PipelineRoutingRule, PipelineStep, ProcessingLog, SavedSearch, diff --git a/tests/test_routing_rules.py b/tests/test_routing_rules.py new file mode 100644 index 00000000..3598c3e0 --- /dev/null +++ b/tests/test_routing_rules.py @@ -0,0 +1,803 @@ +"""Tests for the routing rules API and routing engine. + +Covers CRUD operations for routing rules, rule evaluation (dry-run and engine), +operator logic, access control, and edge cases. +""" + +import json + +import pytest + +from app.models import FileRecord, Pipeline, PipelineRoutingRule +from app.utils.routing_engine import ( + BUILTIN_FIELDS, + VALID_OPERATORS, + _evaluate_condition, + _resolve_field, + _to_float, + build_document_properties, + evaluate_routing_rules, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_pipeline(db_session, name="Test Pipeline", owner_id="testuser", is_active=True): + """Insert a minimal Pipeline and return it.""" + p = Pipeline(owner_id=owner_id, name=name, is_default=False, is_active=is_active) + db_session.add(p) + db_session.commit() + db_session.refresh(p) + return p + + +def _make_rule(db_session, target_pipeline_id, **kwargs): + """Insert a PipelineRoutingRule and return it.""" + defaults = { + "owner_id": "testuser", + "name": "Test Rule", + "position": 0, + "field": "file_type", + "operator": "equals", + "value": "application/pdf", + "is_active": True, + } + defaults.update(kwargs) + defaults["target_pipeline_id"] = target_pipeline_id + rule = PipelineRoutingRule(**defaults) + db_session.add(rule) + db_session.commit() + db_session.refresh(rule) + return rule + + +def _make_file_record(db_session, **kwargs): + """Insert a minimal FileRecord and return it.""" + defaults = { + "owner_id": "testuser", + "filehash": "abc123", + "original_filename": "test.pdf", + "local_filename": "/tmp/test.pdf", + "file_size": 1024, + "mime_type": "application/pdf", + } + defaults.update(kwargs) + fr = FileRecord(**defaults) + db_session.add(fr) + db_session.commit() + db_session.refresh(fr) + return fr + + +# =========================================================================== +# Unit tests – routing engine +# =========================================================================== + + +@pytest.mark.unit +class TestResolveField: + """Tests for _resolve_field().""" + + def test_builtin_field(self): + """Built-in fields are resolved directly from the dict.""" + props = {"file_type": "application/pdf", "size": 1024} + assert _resolve_field("file_type", props) == "application/pdf" + assert _resolve_field("size", props) == 1024 + + def test_category_alias(self): + """'category' is an alias for 'document_type'.""" + props = {"document_type": "Invoice"} + assert _resolve_field("category", props) == "Invoice" + + def test_metadata_field(self): + """'metadata.' resolves from the nested metadata dict.""" + props = {"metadata": {"sender": "Acme Corp", "amount": 100.50}} + assert _resolve_field("metadata.sender", props) == "Acme Corp" + assert _resolve_field("metadata.amount", props) == 100.50 + + def test_missing_metadata_key(self): + """Missing metadata key returns None.""" + props = {"metadata": {"sender": "Acme"}} + assert _resolve_field("metadata.missing_key", props) is None + + def test_missing_metadata_dict(self): + """Missing metadata dict returns None.""" + props = {} + assert _resolve_field("metadata.sender", props) is None + + def test_missing_builtin_field(self): + """Missing built-in field returns None.""" + props = {} + assert _resolve_field("file_type", props) is None + + +@pytest.mark.unit +class TestToFloat: + """Tests for _to_float().""" + + def test_int_value(self): + assert _to_float(42) == 42.0 + + def test_float_value(self): + assert _to_float(3.14) == 3.14 + + def test_string_number(self): + assert _to_float("100") == 100.0 + + def test_none_returns_none(self): + assert _to_float(None) is None + + def test_non_numeric_string_returns_none(self): + assert _to_float("not-a-number") is None + + +@pytest.mark.unit +class TestEvaluateCondition: + """Tests for _evaluate_condition().""" + + def test_equals_match(self): + assert _evaluate_condition("application/pdf", "equals", "application/pdf") is True + + def test_equals_case_insensitive(self): + assert _evaluate_condition("Application/PDF", "equals", "application/pdf") is True + + def test_equals_no_match(self): + assert _evaluate_condition("image/png", "equals", "application/pdf") is False + + def test_not_equals_match(self): + assert _evaluate_condition("image/png", "not_equals", "application/pdf") is True + + def test_not_equals_no_match(self): + assert _evaluate_condition("application/pdf", "not_equals", "application/pdf") is False + + def test_contains_match(self): + assert _evaluate_condition("invoice_2024.pdf", "contains", "invoice") is True + + def test_contains_case_insensitive(self): + assert _evaluate_condition("INVOICE_2024.PDF", "contains", "invoice") is True + + def test_contains_no_match(self): + assert _evaluate_condition("receipt.pdf", "contains", "invoice") is False + + def test_not_contains_match(self): + assert _evaluate_condition("receipt.pdf", "not_contains", "invoice") is True + + def test_not_contains_no_match(self): + assert _evaluate_condition("invoice_2024.pdf", "not_contains", "invoice") is False + + def test_regex_match(self): + assert _evaluate_condition("invoice_2024.pdf", "regex", r"invoice_\d+\.pdf") is True + + def test_regex_no_match(self): + assert _evaluate_condition("receipt.pdf", "regex", r"invoice_\d+\.pdf") is False + + def test_regex_case_insensitive(self): + assert _evaluate_condition("INVOICE_2024.PDF", "regex", r"invoice_\d+\.pdf") is True + + def test_regex_invalid_pattern(self): + """Invalid regex should return False, not raise.""" + assert _evaluate_condition("test", "regex", r"[invalid") is False + + def test_gt(self): + assert _evaluate_condition(2048, "gt", "1024") is True + assert _evaluate_condition(1024, "gt", "1024") is False + + def test_lt(self): + assert _evaluate_condition(512, "lt", "1024") is True + assert _evaluate_condition(1024, "lt", "1024") is False + + def test_gte(self): + assert _evaluate_condition(1024, "gte", "1024") is True + assert _evaluate_condition(2048, "gte", "1024") is True + assert _evaluate_condition(512, "gte", "1024") is False + + def test_lte(self): + assert _evaluate_condition(1024, "lte", "1024") is True + assert _evaluate_condition(512, "lte", "1024") is True + assert _evaluate_condition(2048, "lte", "1024") is False + + def test_none_actual_returns_false(self): + """When the actual value is None, most operators return False.""" + assert _evaluate_condition(None, "equals", "test") is False + assert _evaluate_condition(None, "contains", "test") is False + assert _evaluate_condition(None, "regex", "test") is False + assert _evaluate_condition(None, "gt", "10") is False + + def test_none_actual_not_equals_returns_true(self): + """not_equals should return True when actual is None.""" + assert _evaluate_condition(None, "not_equals", "test") is True + + def test_none_actual_not_contains_returns_true(self): + """not_contains should return True when actual is None.""" + assert _evaluate_condition(None, "not_contains", "test") is True + + def test_non_numeric_gt_returns_false(self): + """Non-numeric values should return False for numeric operators.""" + assert _evaluate_condition("abc", "gt", "100") is False + + def test_unknown_operator_returns_false(self): + """Unknown operator should return False.""" + assert _evaluate_condition("test", "unknown_op", "test") is False + + +@pytest.mark.unit +class TestBuildDocumentProperties: + """Tests for build_document_properties().""" + + def test_basic_properties(self): + """Properties are extracted from FileRecord attributes.""" + fr = _MockFileRecord( + mime_type="application/pdf", + original_filename="test.pdf", + file_size=2048, + ai_metadata=json.dumps({"document_type": "Invoice", "sender": "Acme"}), + ) + props = build_document_properties(fr) + assert props["file_type"] == "application/pdf" + assert props["filename"] == "test.pdf" + assert props["size"] == 2048 + assert props["document_type"] == "Invoice" + assert props["metadata"]["sender"] == "Acme" + + def test_no_metadata(self): + """When ai_metadata is None, metadata is an empty dict.""" + fr = _MockFileRecord(mime_type="image/png", original_filename="img.png", file_size=512, ai_metadata=None) + props = build_document_properties(fr) + assert props["metadata"] == {} + assert props["document_type"] is None + + def test_invalid_metadata_json(self): + """Invalid JSON in ai_metadata should result in empty metadata.""" + fr = _MockFileRecord( + mime_type="application/pdf", + original_filename="test.pdf", + file_size=1024, + ai_metadata="not-json", + ) + props = build_document_properties(fr) + assert props["metadata"] == {} + + +class _MockFileRecord: + """Lightweight stand-in for FileRecord in unit tests.""" + + def __init__(self, **kwargs): + for k, v in kwargs.items(): + setattr(self, k, v) + + +# =========================================================================== +# Unit tests – evaluate_routing_rules (DB-backed) +# =========================================================================== + + +@pytest.mark.unit +class TestEvaluateRoutingRules: + """Tests for evaluate_routing_rules() with a real DB session.""" + + def test_first_match_wins(self, db_session): + """The first matching rule (by position) is used.""" + p1 = _make_pipeline(db_session, name="Pipeline A") + p2 = _make_pipeline(db_session, name="Pipeline B") + + _make_rule(db_session, p1.id, position=0, field="file_type", operator="equals", value="application/pdf") + _make_rule(db_session, p2.id, position=1, field="file_type", operator="equals", value="application/pdf") + + doc = {"file_type": "application/pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is not None + assert result.id == p1.id + + def test_no_match_returns_none(self, db_session): + """When no rule matches, None is returned (caller uses default).""" + p = _make_pipeline(db_session, name="Pipeline A") + _make_rule(db_session, p.id, field="file_type", operator="equals", value="image/png") + + doc = {"file_type": "application/pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is None + + def test_inactive_rule_skipped(self, db_session): + """Inactive rules are not evaluated.""" + p = _make_pipeline(db_session, name="Pipeline A") + _make_rule(db_session, p.id, is_active=False) + + doc = {"file_type": "application/pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is None + + def test_inactive_pipeline_skipped(self, db_session): + """Matching rule with inactive target pipeline is skipped.""" + p = _make_pipeline(db_session, name="Inactive Pipeline", is_active=False) + _make_rule(db_session, p.id) + + doc = {"file_type": "application/pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is None + + def test_system_rules_evaluated_after_user_rules(self, db_session): + """System rules (owner_id=NULL) are evaluated after user-specific rules.""" + p_user = _make_pipeline(db_session, name="User Pipeline") + p_system = _make_pipeline(db_session, name="System Pipeline", owner_id=None) + + # System rule at position 0, user rule at position 1 — user should still win. + _make_rule( + db_session, + p_system.id, + owner_id=None, + position=0, + field="file_type", + operator="equals", + value="application/pdf", + name="System Rule", + ) + _make_rule( + db_session, + p_user.id, + owner_id="testuser", + position=1, + field="file_type", + operator="equals", + value="application/pdf", + name="User Rule", + ) + + doc = {"file_type": "application/pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is not None + assert result.id == p_user.id + + def test_metadata_field_routing(self, db_session): + """Rules can match on metadata.* fields.""" + p = _make_pipeline(db_session, name="Invoice Pipeline") + _make_rule( + db_session, + p.id, + field="metadata.sender", + operator="contains", + value="acme", + ) + + doc = {"file_type": "application/pdf", "metadata": {"sender": "Acme Corporation"}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is not None + assert result.id == p.id + + def test_size_routing(self, db_session): + """Rules can match on file size with numeric comparison.""" + p = _make_pipeline(db_session, name="Large File Pipeline") + _make_rule( + db_session, + p.id, + field="size", + operator="gt", + value="1048576", + ) + + # 2 MB file + doc = {"size": 2097152, "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is not None + assert result.id == p.id + + def test_regex_routing(self, db_session): + """Rules can match using regex on filename.""" + p = _make_pipeline(db_session, name="Invoice Pipeline") + _make_rule( + db_session, + p.id, + field="filename", + operator="regex", + value=r"invoice_\d{4}.*", + ) + + doc = {"filename": "invoice_2024_q1.pdf", "metadata": {}} + result = evaluate_routing_rules(db_session, "testuser", doc) + assert result is not None + assert result.id == p.id + + +# =========================================================================== +# API tests – operators endpoint (public) +# =========================================================================== + + +@pytest.mark.unit +class TestOperatorsEndpoint: + """Tests for the /api/routing-rules/operators catalogue endpoint.""" + + def test_operators_returns_lists(self, client): + """GET /api/routing-rules/operators returns operators and fields.""" + r = client.get("/api/routing-rules/operators") + assert r.status_code == 200 + data = r.json() + assert "operators" in data + assert "builtin_fields" in data + assert "metadata_prefix" in data + assert set(data["operators"]) == VALID_OPERATORS + assert set(data["builtin_fields"]) == BUILTIN_FIELDS + + +# =========================================================================== +# API tests – CRUD +# =========================================================================== + + +@pytest.mark.unit +class TestRoutingRuleCRUD: + """Tests for the routing rules CRUD API endpoints.""" + + def test_create_rule(self, client, db_session): + """POST /api/routing-rules creates a new rule.""" + p = _make_pipeline(db_session, name="Target Pipeline") + r = client.post( + "/api/routing-rules", + json={ + "name": "Route PDFs", + "field": "file_type", + "operator": "equals", + "value": "application/pdf", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 201 + data = r.json() + assert data["name"] == "Route PDFs" + assert data["field"] == "file_type" + assert data["operator"] == "equals" + assert data["target_pipeline_id"] == p.id + assert data["is_active"] is True + + def test_create_rule_invalid_field(self, client, db_session): + """POST with an invalid field returns 422.""" + p = _make_pipeline(db_session, name="Target") + r = client.post( + "/api/routing-rules", + json={ + "name": "Bad field", + "field": "invalid_field", + "operator": "equals", + "value": "test", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 422 + + def test_create_rule_invalid_operator(self, client, db_session): + """POST with an invalid operator returns 422.""" + p = _make_pipeline(db_session, name="Target") + r = client.post( + "/api/routing-rules", + json={ + "name": "Bad op", + "field": "file_type", + "operator": "invalid_op", + "value": "test", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 422 + + def test_create_rule_missing_pipeline(self, client): + """POST referencing a nonexistent pipeline returns 404.""" + r = client.post( + "/api/routing-rules", + json={ + "name": "No pipeline", + "field": "file_type", + "operator": "equals", + "value": "application/pdf", + "target_pipeline_id": 99999, + }, + ) + assert r.status_code == 404 + + def test_list_rules(self, client, db_session): + """GET /api/routing-rules returns the user's rules.""" + p = _make_pipeline(db_session, name="Pipeline") + _make_rule(db_session, p.id, name="Rule 1", position=0, owner_id="anonymous") + _make_rule(db_session, p.id, name="Rule 2", position=1, owner_id="anonymous") + + r = client.get("/api/routing-rules") + assert r.status_code == 200 + data = r.json() + assert len(data) >= 2 + names = [d["name"] for d in data] + assert "Rule 1" in names + assert "Rule 2" in names + + def test_get_rule(self, client, db_session): + """GET /api/routing-rules/{id} returns a specific rule.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, name="My Rule", owner_id="anonymous") + + r = client.get(f"/api/routing-rules/{rule.id}") + assert r.status_code == 200 + assert r.json()["name"] == "My Rule" + + def test_get_rule_not_found(self, client): + """GET /api/routing-rules/{id} returns 404 for nonexistent rule.""" + r = client.get("/api/routing-rules/99999") + assert r.status_code == 404 + + def test_update_rule(self, client, db_session): + """PUT /api/routing-rules/{id} updates a rule.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, name="Old Name", owner_id="anonymous") + + r = client.put( + f"/api/routing-rules/{rule.id}", + json={"name": "New Name", "operator": "contains"}, + ) + assert r.status_code == 200 + data = r.json() + assert data["name"] == "New Name" + assert data["operator"] == "contains" + + def test_update_rule_invalid_field(self, client, db_session): + """PUT with invalid field returns 422.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, owner_id="anonymous") + + r = client.put(f"/api/routing-rules/{rule.id}", json={"field": "bad_field"}) + assert r.status_code == 422 + + def test_update_rule_invalid_operator(self, client, db_session): + """PUT with invalid operator returns 422.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, owner_id="anonymous") + + r = client.put(f"/api/routing-rules/{rule.id}", json={"operator": "bad_op"}) + assert r.status_code == 422 + + def test_update_rule_missing_pipeline(self, client, db_session): + """PUT with nonexistent target pipeline returns 404.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, owner_id="anonymous") + + r = client.put(f"/api/routing-rules/{rule.id}", json={"target_pipeline_id": 99999}) + assert r.status_code == 404 + + def test_delete_rule(self, client, db_session): + """DELETE /api/routing-rules/{id} removes a rule.""" + p = _make_pipeline(db_session, name="Pipeline") + rule = _make_rule(db_session, p.id, owner_id="anonymous") + + r = client.delete(f"/api/routing-rules/{rule.id}") + assert r.status_code == 204 + + # Verify it's gone + r = client.get(f"/api/routing-rules/{rule.id}") + assert r.status_code == 404 + + def test_delete_rule_not_found(self, client): + """DELETE for nonexistent rule returns 404.""" + r = client.delete("/api/routing-rules/99999") + assert r.status_code == 404 + + +# =========================================================================== +# API tests – reorder +# =========================================================================== + + +@pytest.mark.unit +class TestReorderRoutingRules: + """Tests for the PUT /api/routing-rules/reorder endpoint.""" + + def test_reorder_rules(self, client, db_session): + """Reordering updates the position of rules.""" + p = _make_pipeline(db_session, name="Pipeline") + r1 = _make_rule(db_session, p.id, name="A", position=0, owner_id="anonymous") + r2 = _make_rule(db_session, p.id, name="B", position=1, owner_id="anonymous") + + r = client.put( + "/api/routing-rules/reorder", + json={"rule_ids": [r2.id, r1.id]}, + ) + assert r.status_code == 200 + data = r.json() + assert data[0]["id"] == r2.id + assert data[0]["position"] == 0 + assert data[1]["id"] == r1.id + assert data[1]["position"] == 1 + + def test_reorder_invalid_ids(self, client, db_session): + """Reorder with invalid IDs returns 422.""" + r = client.put( + "/api/routing-rules/reorder", + json={"rule_ids": [99999]}, + ) + assert r.status_code == 422 + + +# =========================================================================== +# API tests – evaluate (dry-run) +# =========================================================================== + + +@pytest.mark.unit +class TestEvaluateEndpoint: + """Tests for the POST /api/routing-rules/evaluate dry-run endpoint.""" + + def test_evaluate_match(self, client, db_session): + """Evaluate returns the matching rule and target pipeline.""" + p = _make_pipeline(db_session, name="Invoice Pipeline") + _make_rule( + db_session, + p.id, + name="PDF Route", + field="file_type", + operator="equals", + value="application/pdf", + owner_id="anonymous", + ) + + r = client.post( + "/api/routing-rules/evaluate", + json={"file_type": "application/pdf"}, + ) + assert r.status_code == 200 + data = r.json() + assert data["matched"] is True + assert data["rule"]["name"] == "PDF Route" + assert data["target_pipeline"]["id"] == p.id + + def test_evaluate_no_match(self, client, db_session): + """Evaluate returns matched=False when no rule applies.""" + p = _make_pipeline(db_session, name="Pipeline") + _make_rule( + db_session, + p.id, + name="PNG Route", + field="file_type", + operator="equals", + value="image/png", + owner_id="anonymous", + ) + + r = client.post( + "/api/routing-rules/evaluate", + json={"file_type": "application/pdf"}, + ) + assert r.status_code == 200 + data = r.json() + assert data["matched"] is False + assert data["rule"] is None + + def test_evaluate_with_metadata(self, client, db_session): + """Evaluate with metadata.* fields works.""" + p = _make_pipeline(db_session, name="Invoice Pipeline") + _make_rule( + db_session, + p.id, + name="Invoice Route", + field="metadata.sender", + operator="contains", + value="acme", + owner_id="anonymous", + ) + + r = client.post( + "/api/routing-rules/evaluate", + json={"metadata": {"sender": "Acme Corporation"}}, + ) + assert r.status_code == 200 + data = r.json() + assert data["matched"] is True + + def test_evaluate_with_size(self, client, db_session): + """Evaluate with size comparisons works.""" + p = _make_pipeline(db_session, name="Large Pipeline") + _make_rule( + db_session, + p.id, + name="Large Files", + field="size", + operator="gt", + value="1000000", + owner_id="anonymous", + ) + + r = client.post( + "/api/routing-rules/evaluate", + json={"size": 2000000}, + ) + assert r.status_code == 200 + assert r.json()["matched"] is True + + +# =========================================================================== +# API tests – metadata field validation +# =========================================================================== + + +@pytest.mark.unit +class TestFieldValidation: + """Tests for field validation in routing rule creation.""" + + def test_metadata_prefix_accepted(self, client, db_session): + """Fields with 'metadata.' prefix are valid.""" + p = _make_pipeline(db_session, name="Pipeline") + r = client.post( + "/api/routing-rules", + json={ + "name": "Metadata Rule", + "field": "metadata.sender", + "operator": "equals", + "value": "test", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 201 + + def test_bare_metadata_rejected(self, client, db_session): + """Just 'metadata.' without a key suffix is invalid.""" + p = _make_pipeline(db_session, name="Pipeline") + r = client.post( + "/api/routing-rules", + json={ + "name": "Bad Metadata", + "field": "metadata.", + "operator": "equals", + "value": "test", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 422 + + @pytest.mark.parametrize("field", sorted(BUILTIN_FIELDS)) + def test_builtin_fields_accepted(self, client, db_session, field): + """All built-in fields are accepted.""" + p = _make_pipeline(db_session, name=f"Pipeline for {field}") + r = client.post( + "/api/routing-rules", + json={ + "name": f"Rule for {field}", + "field": field, + "operator": "equals", + "value": "test", + "target_pipeline_id": p.id, + }, + ) + assert r.status_code == 201 + + +# =========================================================================== +# API tests – auto-position +# =========================================================================== + + +@pytest.mark.unit +class TestAutoPosition: + """Tests for automatic position assignment.""" + + def test_auto_position_increments(self, client, db_session): + """Rules created without position get incrementing positions.""" + p = _make_pipeline(db_session, name="Pipeline") + + r1 = client.post( + "/api/routing-rules", + json={ + "name": "First", + "field": "file_type", + "operator": "equals", + "value": "application/pdf", + "target_pipeline_id": p.id, + }, + ) + r2 = client.post( + "/api/routing-rules", + json={ + "name": "Second", + "field": "file_type", + "operator": "equals", + "value": "image/png", + "target_pipeline_id": p.id, + }, + ) + assert r1.status_code == 201 + assert r2.status_code == 201 + assert r2.json()["position"] > r1.json()["position"]