#!/usr/bin/env python3 """Test script for the delete document endpoint on Airflow webserver. Usage: # Quick connectivity check python scripts/test_delete_document_endpoint.py --check # Delete a document by ID python scripts/test_delete_document_endpoint.py delete DOCUMENT_ID # Delete with custom host python scripts/test_delete_document_endpoint.py --host http://localhost:8081 delete DOCUMENT_ID # Test invalid UUID handling python scripts/test_delete_document_endpoint.py delete not-a-uuid # Test non-existent document (expect 404) python scripts/test_delete_document_endpoint.py delete 00000000-0000-0000-0000-000000000000 Requires HMAC_KEY_RING env var (or HMAC_KEY_RING_FILE) to be set. """ import argparse import json import sys import uuid from dotenv import load_dotenv load_dotenv() import requests from textiq_hmac import load_key_ring, sign_context_headers DEFAULT_TENANT_ID = "28aa3971-48af-4f2a-aa73-9e4f89fbfba3" DEFAULT_NODE_ID = "22222222-2222-2222-2222-222222222222" DEFAULT_USER_ID = "33333333-3333-3333-3333-333333333333" DEFAULT_ROLE = "content_specialist" def sign_request(tenant_id, node_id, user_id, role): """Generate HMAC headers using textiq_hmac package.""" key_ring = load_key_ring() headers = { "X-Tenant-ID": tenant_id, "X-Node-ID": node_id, "X-User-ID": user_id, "X-Role": role, } return sign_context_headers(headers, key_ring) def _print_response(r): """Print response details.""" print(f" Status: {r.status_code}") print(f" X-Correlation-ID: {r.headers.get('X-Correlation-ID', 'NOT PRESENT')}") print(f" X-Request-ID: {r.headers.get('X-Request-ID', 'NOT PRESENT')}") try: body = r.json() print(f" Body: {json.dumps(body, indent=2)}") except Exception: print(f" Body: {r.text[:500]}") def check_endpoint(host): """Quick connectivity check — GET should return 405 if route exists.""" url = f"{host}/api/v1/documents/00000000-0000-0000-0000-000000000000" print(f"GET {url}") try: r = requests.get(url, timeout=5) print(f" Status: {r.status_code}") if r.status_code in (405, 401): print(f" -> Route exists ({r.status_code} = expected for unauthenticated GET)") return True elif r.status_code == 404: print(" -> Route NOT found. Plugin may not be loaded.") print(" Run: docker exec textiq-airflow-webserver airflow plugins") return False else: print(f" -> Unexpected status. Body: {r.text[:200]}") return False except requests.ConnectionError as e: print(e) print(f" -> Cannot connect to {host}. Is the webserver running?") return False def test_no_auth(host): """DELETE without auth headers — should return 401.""" url = f"{host}/api/v1/documents/00000000-0000-0000-0000-000000000000" print(f"\nDELETE {url} (no auth headers)") r = requests.delete(url, timeout=10) print(f" Status: {r.status_code}") print(f" Body: {r.text[:200]}") if r.status_code == 401: print(" -> PASS: Got 401 as expected") else: print(f" -> FAIL: Expected 401, got {r.status_code}") def test_invalid_uuid(host, tenant_id, node_id, user_id, role): """DELETE with invalid UUID — should return 422.""" url = f"{host}/api/v1/documents/not-a-valid-uuid" headers = sign_request(tenant_id, node_id, user_id, role) headers["X-Correlation-ID"] = str(uuid.uuid4()) print(f"\nDELETE {url} (invalid UUID)") r = requests.delete(url, headers=headers, timeout=10) _print_response(r) if r.status_code == 422: print(" -> PASS: Got 422 as expected") else: print(f" -> FAIL: Expected 422, got {r.status_code}") def test_not_found(host, tenant_id, node_id, user_id, role): """DELETE with non-existent document — should return 404.""" fake_id = "00000000-0000-0000-0000-000000000000" url = f"{host}/api/v1/documents/{fake_id}" headers = sign_request(tenant_id, node_id, user_id, role) headers["X-Correlation-ID"] = str(uuid.uuid4()) print(f"\nDELETE {url} (non-existent document)") r = requests.delete(url, headers=headers, timeout=10) _print_response(r) if r.status_code == 404: print(" -> PASS: Got 404 as expected") else: print(f" -> FAIL: Expected 404, got {r.status_code}") def delete_document(host, document_id, tenant_id, node_id, user_id, role): """DELETE /api/v1/documents/{document_id}""" url = f"{host}/api/v1/documents/{document_id}" headers = sign_request(tenant_id, node_id, user_id, role) headers["X-Correlation-ID"] = str(uuid.uuid4()) print(f"\nDELETE {url}") print(f" Tenant: {tenant_id}") print(f" Node: {node_id}") print(f" Correlation-ID: {headers['X-Correlation-ID']}") r = requests.delete(url, headers=headers, timeout=30) _print_response(r) if r.status_code == 200: body = r.json() deleted = body.get("deleted", {}) print(f" -> PASS: Document deleted") print(f" Chunks: {deleted.get('chunks', '?')}") print(f" Domain terms: {deleted.get('domain_terms', '?')}") print(f" S3 files: {deleted.get('s3_files', '?')}") elif r.status_code == 404: print(" -> Document not found (404)") elif r.status_code == 422: print(" -> Invalid UUID (422)") elif r.status_code == 500: print(" -> Server error (500) — check Milvus connectivity") else: print(f" -> FAIL: Unexpected status {r.status_code}") def main(): parser = argparse.ArgumentParser(description="Test DE delete document endpoint") parser.add_argument("--host", default="http://localhost:8081", help="Airflow webserver URL") parser.add_argument("--check", action="store_true", help="Quick connectivity check + edge case tests") parser.add_argument("--tenant-id", default=DEFAULT_TENANT_ID) parser.add_argument("--node-id", default=DEFAULT_NODE_ID) parser.add_argument("--user-id", default=DEFAULT_USER_ID) parser.add_argument("--role", default=DEFAULT_ROLE) subparsers = parser.add_subparsers(dest="command") sp_delete = subparsers.add_parser("delete", help="Delete a document") sp_delete.add_argument("document_id", help="Document UUID to delete") args = parser.parse_args() print(f"=== Delete Document Endpoint Test ({args.host}) ===\n") alive = check_endpoint(args.host) if not alive: sys.exit(1) if args.check: test_no_auth(args.host) test_invalid_uuid(args.host, args.tenant_id, args.node_id, args.user_id, args.role) test_not_found(args.host, args.tenant_id, args.node_id, args.user_id, args.role) return if not args.command: print("\nNo command specified. Running auth + edge case tests.\n") test_no_auth(args.host) test_invalid_uuid(args.host, args.tenant_id, args.node_id, args.user_id, args.role) test_not_found(args.host, args.tenant_id, args.node_id, args.user_id, args.role) return if args.command == "delete": delete_document( args.host, args.document_id, args.tenant_id, args.node_id, args.user_id, args.role, ) if __name__ == "__main__": main()