From e658dec83c961c314a3ad26f8a93e4ac67f2c9f0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 11 Feb 2026 15:16:53 +0000 Subject: [PATCH] feat(test): add comprehensive external API integration tests with PDF generation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add fpdf2 dependency for dynamic test PDF generation - Create tests/test_external_integrations.py with end-to-end pipeline tests: - OpenAI: key validation, metadata extraction via chat completion - Azure Document Intelligence: admin connectivity, full OCR on generated PDF - S3: bucket access, upload/download/delete pipeline - Dropbox: token refresh, upload/download/delete pipeline - OneDrive: token refresh, upload/download/delete pipeline - Authentik: OIDC discovery endpoint, credential consistency - Full pipeline: Azure OCR → OpenAI metadata extraction - Update tests/conftest.py to capture original env vars before test overrides - Add has_real_env() helper and original_env fixture for credential detection - All external tests use @pytest.mark.requires_external and skipif guards - Test files are dynamically generated with unique content per run - Uploaded test files are cleaned up in finally blocks Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- requirements-dev.txt | 1 + tests/conftest.py | 53 +- tests/test_external_integrations.py | 810 ++++++++++++++++++++++++++++ 3 files changed, 863 insertions(+), 1 deletion(-) create mode 100644 tests/test_external_integrations.py diff --git a/requirements-dev.txt b/requirements-dev.txt index b20fdb6b..765e2ae7 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -8,6 +8,7 @@ pytest-asyncio>=0.23.0 pytest-mock>=3.12.0 httpx>=0.26.0 # For async test client testcontainers>=3.7.1 # For integration tests with real containers +fpdf2>=2.8.0 # For generating test PDF documents in integration tests minio>=7.1.0 # For MinIO/S3 integration tests redis>=4.5.0 # For Redis integration tests boto3>=1.26.0 # For S3 integration tests diff --git a/tests/conftest.py b/tests/conftest.py index 03d8b6f4..20f395b3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,7 +4,7 @@ Pytest configuration and shared fixtures for DocuElevate tests. import os import tempfile -from typing import Generator +from typing import Dict, Generator, Optional import pytest from fastapi.testclient import TestClient @@ -12,6 +12,37 @@ from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool +# Capture original environment variables before overriding with test defaults. +# This allows integration tests to detect when real API credentials are available +# (e.g., injected via GitHub Actions secrets) and run live API verification. +_EXTERNAL_API_ENV_KEYS = [ + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "AZURE_AI_KEY", + "AZURE_ENDPOINT", + "AZURE_REGION", + "AWS_ACCESS_KEY_ID", + "AWS_SECRET_ACCESS_KEY", + "S3_BUCKET_NAME", + "S3_FOLDER_PREFIX", + "DROPBOX_APP_KEY", + "DROPBOX_APP_SECRET", + "DROPBOX_REFRESH_TOKEN", + "ONEDRIVE_CLIENT_ID", + "ONEDRIVE_CLIENT_SECRET", + "ONEDRIVE_REFRESH_TOKEN", + "ONEDRIVE_TENANT_ID", + "ONEDRIVE_FOLDER_PATH", + "GOOGLE_DRIVE_CREDENTIALS_JSON", + "GOOGLE_DRIVE_FOLDER_ID", + "AUTHENTIK_CLIENT_ID", + "AUTHENTIK_CLIENT_SECRET", + "AUTHENTIK_CONFIG_URL", + "SESSION_SECRET", +] +_PLACEHOLDER_VALUES = {"test-key", "test", "", "NOT_SET"} +_original_env: Dict[str, Optional[str]] = {key: os.environ.get(key) for key in _EXTERNAL_API_ENV_KEYS} + # Set test environment variables before importing app os.environ["DATABASE_URL"] = "sqlite:///:memory:" os.environ["REDIS_URL"] = "redis://localhost:6379/1" @@ -167,6 +198,26 @@ def mock_azure_response(): return {"analyzeResult": {"content": "Test document content extracted by OCR", "pages": [{"pageNumber": 1}]}} +def has_real_env(*keys: str) -> bool: + """Check if real (non-placeholder) environment variables were set before test overrides. + + Returns True only if ALL specified keys had non-placeholder values in the + original environment. Used by integration tests to decide whether to skip + when real credentials are unavailable. + """ + for key in keys: + value = _original_env.get(key) + if value is None or value in _PLACEHOLDER_VALUES: + return False + return True + + +@pytest.fixture(scope="session") +def original_env() -> Dict[str, Optional[str]]: + """Provide access to the original environment variables captured before test overrides.""" + return dict(_original_env) + + # Markers for categorizing tests def pytest_configure(config): """Configure custom pytest markers.""" diff --git a/tests/test_external_integrations.py b/tests/test_external_integrations.py new file mode 100644 index 00000000..8813483c --- /dev/null +++ b/tests/test_external_integrations.py @@ -0,0 +1,810 @@ +""" +Integration tests for external API services using real credentials. + +These tests verify that external API integrations work correctly when real +credentials are provided via environment variables (e.g., GitHub Actions secrets). + +Each test is guarded by ``pytest.mark.skipif`` so it is skipped automatically +when the required environment variables are absent or still set to placeholder +values. All tests carry the ``@pytest.mark.requires_external`` marker so they +can be run (or excluded) with:: + + pytest -m requires_external # run only external tests + pytest -m "not requires_external" # skip external tests + +**Pipeline coverage:** + +The tests exercise real end-to-end flows wherever credentials allow: + +- *OpenAI*: key validation **and** metadata extraction via chat completion. +- *Azure Document Intelligence*: admin connectivity **and** OCR of a generated PDF. +- *S3*: bucket access, file upload, download verification, and cleanup. +- *Dropbox*: token refresh, file upload, download verification, and cleanup. +- *OneDrive*: token refresh, file upload, download verification, and cleanup. +- *Authentik/OIDC*: discovery endpoint and credential consistency. + +Each pipeline test dynamically generates a unique PDF with ``fpdf2`` so every +run operates on fresh data. Uploaded test files are cleaned up in ``finally`` +blocks to avoid polluting external storage. +""" + +import json +import logging +import os +import tempfile +import uuid +from typing import Optional + +import pytest + +from tests.conftest import has_real_env + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Test-PDF generator helper +# --------------------------------------------------------------------------- +_TEST_PREFIX = "docuelevate_test_" + + +def generate_test_pdf( + content: Optional[str] = None, + filename_prefix: str = _TEST_PREFIX, +) -> str: + """Generate a unique test PDF with embedded text. + + Creates a one-page PDF containing *content* (or a random invoice stub) + and returns the path to the temporary file. The caller is responsible + for deleting the file when done. + + Args: + content: Optional text to embed. When ``None`` a realistic + invoice-style document is generated. + filename_prefix: Prefix for the temp filename. + + Returns: + Absolute path to the generated PDF file. + """ + from fpdf import FPDF + + unique_id = uuid.uuid4().hex[:8] + + if content is None: + content = ( + f"Invoice #{unique_id}\n" + f"Date: 2024-06-15\n" + f"From: Acme Integration Testing GmbH\n" + f"To: DocuElevate QA Department\n" + f"Amount: EUR 1,234.56\n\n" + f"Description: Annual subscription renewal for cloud document\n" + f"processing services. Reference: REF-{unique_id}.\n\n" + f"Payment terms: Net 30 days.\n" + f"Bank: Deutsche Bank, IBAN: DE89 3704 0044 0532 0130 00\n" + ) + + pdf = FPDF() + pdf.add_page() + pdf.set_font("Helvetica", size=11) + pdf.multi_cell(0, 7, text=content) + + fd, path = tempfile.mkstemp(prefix=filename_prefix, suffix=".pdf") + os.close(fd) + pdf.output(path) + return path + + +# --------------------------------------------------------------------------- +# OpenAI +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("OPENAI_API_KEY"), + reason="Real OPENAI_API_KEY not available", +) +class TestOpenAIIntegration: + """Verify OpenAI API connectivity and metadata extraction with real credentials.""" + + def test_openai_api_key_is_valid(self, original_env: dict) -> None: + """Validate that the configured OpenAI API key can list models.""" + import openai + + api_key = original_env["OPENAI_API_KEY"] + base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1" + + oai = openai.OpenAI(api_key=api_key, base_url=base_url) + models = oai.models.list() + + assert hasattr(models, "data"), "Expected models response to have 'data' attribute" + assert len(models.data) > 0, "Expected at least one model to be available" + + def test_openai_test_endpoint_with_real_key(self, client, original_env: dict, monkeypatch) -> None: + """Test the /api/openai/test endpoint returns success with a real API key.""" + monkeypatch.setattr("app.config.settings.openai_api_key", original_env["OPENAI_API_KEY"]) + if original_env.get("OPENAI_BASE_URL"): + monkeypatch.setattr("app.config.settings.openai_base_url", original_env["OPENAI_BASE_URL"]) + + response = client.get("/api/openai/test") + assert response.status_code == 200 + + data = response.json() + assert data["status"] == "success", f"OpenAI test endpoint failed: {data.get('message')}" + assert data.get("models_available", 0) > 0 + + def test_openai_metadata_extraction(self, original_env: dict) -> None: + """End-to-end: send document text to OpenAI and receive structured metadata.""" + import openai + + api_key = original_env["OPENAI_API_KEY"] + base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1" + + # Use the same prompt structure as extract_metadata_with_gpt task + unique_id = uuid.uuid4().hex[:8] + sample_text = ( + f"Invoice #{unique_id}\n" + f"Date: 2024-06-15\n" + f"From: Acme Integration Testing GmbH\n" + f"To: DocuElevate QA Department\n" + f"Amount: EUR 1,234.56\n" + ) + + prompt = ( + "You are a specialized document analyzer. Analyze the given text and return a JSON object with:\n" + '- "document_type": precise classification (e.g., Invoice, Contract)\n' + '- "language": ISO 639-1 code\n' + '- "tags": list of up to 4 keywords\n' + '- "absender": sender name\n' + '- "empfaenger": recipient name\n\n' + f"Text:\n{sample_text}\n\n" + "Return only valid JSON." + ) + + oai = openai.OpenAI(api_key=api_key, base_url=base_url) + completion = oai.chat.completions.create( + model="gpt-4o-mini", + messages=[ + {"role": "system", "content": "You are an intelligent document classifier."}, + {"role": "user", "content": prompt}, + ], + temperature=0, + ) + + content = completion.choices[0].message.content + assert content, "OpenAI returned empty content" + + # Extract JSON from response (may be wrapped in markdown fences) + import re + + json_match = re.search(r"\{.*\}", content, re.DOTALL) + assert json_match, f"No JSON found in OpenAI response: {content[:200]}" + + metadata = json.loads(json_match.group()) + assert "document_type" in metadata, "Missing document_type in extracted metadata" + assert "language" in metadata, "Missing language in extracted metadata" + + +# --------------------------------------------------------------------------- +# Azure Document Intelligence +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("AZURE_AI_KEY", "AZURE_ENDPOINT"), + reason="Real AZURE_AI_KEY and AZURE_ENDPOINT not available", +) +class TestAzureDocumentIntelligenceIntegration: + """Verify Azure Document Intelligence connectivity and OCR with real credentials.""" + + def test_azure_admin_client_connects(self, original_env: dict) -> None: + """Validate that the Azure admin client can list operations.""" + from azure.ai.documentintelligence import DocumentIntelligenceAdministrationClient + from azure.core.credentials import AzureKeyCredential + + admin_client = DocumentIntelligenceAdministrationClient( + endpoint=original_env["AZURE_ENDPOINT"], + credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]), + ) + + operations = list(admin_client.list_operations()) + assert isinstance(operations, list) + + def test_azure_test_endpoint_with_real_credentials(self, client, original_env: dict, monkeypatch) -> None: + """Test the /api/azure/test endpoint returns success with real credentials.""" + monkeypatch.setattr("app.config.settings.azure_ai_key", original_env["AZURE_AI_KEY"]) + monkeypatch.setattr("app.config.settings.azure_endpoint", original_env["AZURE_ENDPOINT"]) + if original_env.get("AZURE_REGION"): + monkeypatch.setattr("app.config.settings.azure_region", original_env["AZURE_REGION"]) + + response = client.get("/api/azure/test") + assert response.status_code == 200 + + data = response.json() + assert data["status"] == "success", f"Azure test endpoint failed: {data.get('message')}" + + def test_azure_ocr_on_generated_pdf(self, original_env: dict) -> None: + """End-to-end: send a generated PDF to Azure and receive OCR text back.""" + from azure.ai.documentintelligence import DocumentIntelligenceClient + from azure.ai.documentintelligence.models import AnalyzeOutputOption + from azure.core.credentials import AzureKeyCredential + + pdf_path = generate_test_pdf() + try: + doc_client = DocumentIntelligenceClient( + endpoint=original_env["AZURE_ENDPOINT"], + credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]), + ) + + with open(pdf_path, "rb") as f: + poller = doc_client.begin_analyze_document( + "prebuilt-read", + body=f, + output=[AnalyzeOutputOption.PDF], + ) + result = poller.result() + + assert result.content, "Azure OCR returned no content" + assert len(result.content) > 10, f"OCR text too short: {result.content[:50]}" + + # Verify the generated text is recognisable + assert ( + "Acme" in result.content or "Invoice" in result.content + ), f"OCR text does not contain expected keywords: {result.content[:200]}" + + # Retrieve the searchable PDF output + operation_id = poller.details["operation_id"] + pdf_response = doc_client.get_analyze_result_pdf( + model_id=result.model_id, + result_id=operation_id, + ) + searchable_bytes = b"".join(pdf_response) + assert len(searchable_bytes) > 0, "Searchable PDF output is empty" + assert searchable_bytes[:5] == b"%PDF-", "Output is not a valid PDF" + finally: + os.unlink(pdf_path) + + +# --------------------------------------------------------------------------- +# AWS S3 – full upload/download/delete pipeline +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "S3_BUCKET_NAME"), + reason="Real AWS credentials and S3_BUCKET_NAME not available", +) +class TestS3Integration: + """Verify AWS S3 connectivity and upload/download pipeline with real credentials.""" + + def test_s3_bucket_accessible(self, original_env: dict) -> None: + """Validate that the S3 bucket exists and credentials are accepted.""" + import boto3 + + s3_client = boto3.client( + "s3", + region_name=original_env.get("AWS_REGION", "us-east-1"), + aws_access_key_id=original_env["AWS_ACCESS_KEY_ID"], + aws_secret_access_key=original_env["AWS_SECRET_ACCESS_KEY"], + ) + + response = s3_client.head_bucket(Bucket=original_env["S3_BUCKET_NAME"]) + assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 + + def test_s3_upload_download_delete(self, original_env: dict) -> None: + """End-to-end: upload a generated PDF to S3, download and verify, then delete.""" + import boto3 + + pdf_path = generate_test_pdf() + s3_key = None + try: + s3_client = boto3.client( + "s3", + region_name=original_env.get("AWS_REGION", "us-east-1"), + aws_access_key_id=original_env["AWS_ACCESS_KEY_ID"], + aws_secret_access_key=original_env["AWS_SECRET_ACCESS_KEY"], + ) + bucket = original_env["S3_BUCKET_NAME"] + prefix = original_env.get("S3_FOLDER_PREFIX", "") + if prefix and not prefix.endswith("/"): + prefix += "/" + s3_key = f"{prefix}{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf" + + # Upload + s3_client.upload_file(pdf_path, bucket, s3_key) + + # Download and verify + download_path = pdf_path + ".downloaded" + s3_client.download_file(bucket, s3_key, download_path) + + with open(pdf_path, "rb") as orig, open(download_path, "rb") as dl: + assert orig.read() == dl.read(), "Downloaded file does not match uploaded file" + + os.unlink(download_path) + finally: + os.unlink(pdf_path) + # Cleanup: delete the test object from S3 + if s3_key: + try: + s3_client.delete_object(Bucket=bucket, Key=s3_key) + except Exception as exc: + logger.warning(f"Failed to clean up S3 test object {s3_key}: {exc}") + + +# --------------------------------------------------------------------------- +# Dropbox – full upload/download/delete pipeline +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("DROPBOX_APP_KEY", "DROPBOX_APP_SECRET", "DROPBOX_REFRESH_TOKEN"), + reason="Real Dropbox credentials not available", +) +class TestDropboxIntegration: + """Verify Dropbox token validity and upload/download pipeline with real credentials.""" + + def test_dropbox_token_refresh_and_account_info(self, original_env: dict) -> None: + """Validate token refresh and account info retrieval in one go.""" + import dropbox as dbx_lib + + dbx = dbx_lib.Dropbox( + app_key=original_env["DROPBOX_APP_KEY"], + app_secret=original_env["DROPBOX_APP_SECRET"], + oauth2_refresh_token=original_env["DROPBOX_REFRESH_TOKEN"], + ) + account = dbx.users_get_current_account() + assert account.email, "Dropbox account missing email" + + def test_dropbox_upload_download_delete(self, original_env: dict) -> None: + """End-to-end: upload a generated PDF to Dropbox, download it, then delete it.""" + import dropbox as dbx_lib + + pdf_path = generate_test_pdf() + remote_path = f"/{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf" + dbx = None + try: + dbx = dbx_lib.Dropbox( + app_key=original_env["DROPBOX_APP_KEY"], + app_secret=original_env["DROPBOX_APP_SECRET"], + oauth2_refresh_token=original_env["DROPBOX_REFRESH_TOKEN"], + ) + + # Upload + with open(pdf_path, "rb") as f: + dbx.files_upload( + f.read(), + remote_path, + mode=dbx_lib.files.WriteMode.overwrite, + ) + + # Download and verify + _, response = dbx.files_download(remote_path) + downloaded = response.content + with open(pdf_path, "rb") as f: + assert f.read() == downloaded, "Downloaded Dropbox file does not match uploaded file" + finally: + os.unlink(pdf_path) + # Cleanup + if dbx: + try: + dbx.files_delete_v2(remote_path) + except Exception as exc: + logger.warning(f"Failed to clean up Dropbox test file {remote_path}: {exc}") + + +# --------------------------------------------------------------------------- +# OneDrive – full upload/download/delete pipeline +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("ONEDRIVE_CLIENT_ID", "ONEDRIVE_CLIENT_SECRET", "ONEDRIVE_REFRESH_TOKEN"), + reason="Real OneDrive credentials not available", +) +class TestOneDriveIntegration: + """Verify OneDrive token validity and upload/download pipeline with real credentials.""" + + @staticmethod + def _get_access_token(env: dict) -> str: + """Obtain a fresh OneDrive access token via MSAL.""" + import msal + + tenant = env.get("ONEDRIVE_TENANT_ID") or "common" + app = msal.ConfidentialClientApplication( + client_id=env["ONEDRIVE_CLIENT_ID"], + client_credential=env["ONEDRIVE_CLIENT_SECRET"], + authority=f"https://login.microsoftonline.com/{tenant}", + ) + result = app.acquire_token_by_refresh_token( + refresh_token=env["ONEDRIVE_REFRESH_TOKEN"], + scopes=["https://graph.microsoft.com/.default"], + ) + assert "access_token" in result, f"OneDrive token acquisition failed: {result.get('error_description')}" + return result["access_token"] + + def test_onedrive_token_refresh_and_user_info(self, original_env: dict) -> None: + """Validate token refresh and user info retrieval.""" + import requests + + token = self._get_access_token(original_env) + resp = requests.get( + "https://graph.microsoft.com/v1.0/me", + headers={"Authorization": f"Bearer {token}"}, + timeout=30, + ) + assert resp.status_code == 200, f"OneDrive user info failed: {resp.text}" + + def test_onedrive_upload_download_delete(self, original_env: dict) -> None: + """End-to-end: upload a generated PDF to OneDrive, download it, then delete it.""" + import requests + + pdf_path = generate_test_pdf() + filename = f"{_TEST_PREFIX}{uuid.uuid4().hex[:8]}.pdf" + folder = original_env.get("ONEDRIVE_FOLDER_PATH", "").strip("/") + item_id = None + token = None + try: + token = self._get_access_token(original_env) + headers = {"Authorization": f"Bearer {token}"} + + # Upload (simple upload for small files) + if folder: + upload_url = f"https://graph.microsoft.com/v1.0/me/drive/root:/{folder}/{filename}:/content" + else: + upload_url = f"https://graph.microsoft.com/v1.0/me/drive/root:/{filename}:/content" + + with open(pdf_path, "rb") as f: + upload_resp = requests.put( + upload_url, + headers={**headers, "Content-Type": "application/pdf"}, + data=f.read(), + timeout=60, + ) + assert upload_resp.status_code in (200, 201), f"OneDrive upload failed: {upload_resp.text}" + item_id = upload_resp.json().get("id") + + # Download and verify + download_url = f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}/content" + dl_resp = requests.get(download_url, headers=headers, timeout=60) + assert dl_resp.status_code == 200, f"OneDrive download failed: {dl_resp.status_code}" + + with open(pdf_path, "rb") as f: + assert f.read() == dl_resp.content, "Downloaded OneDrive file does not match uploaded file" + finally: + os.unlink(pdf_path) + # Cleanup + if item_id and token: + try: + requests.delete( + f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}", + headers={"Authorization": f"Bearer {token}"}, + timeout=30, + ) + except Exception as exc: + logger.warning(f"Failed to clean up OneDrive test file {filename}: {exc}") + + +# --------------------------------------------------------------------------- +# Authentik / OpenID Connect – discovery endpoint validation +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("AUTHENTIK_CONFIG_URL"), + reason="Real AUTHENTIK_CONFIG_URL not available", +) +class TestAuthentikIntegration: + """Verify Authentik / OpenID Connect discovery endpoint is reachable.""" + + def test_oidc_discovery_endpoint(self, original_env: dict) -> None: + """Validate that the OIDC discovery URL returns a valid JSON document.""" + import requests + + config_url = original_env["AUTHENTIK_CONFIG_URL"] + response = requests.get(config_url, timeout=30) + + assert response.status_code == 200, f"OIDC discovery failed: {response.status_code}" + data = response.json() + + assert "issuer" in data, "OIDC response missing 'issuer'" + assert "authorization_endpoint" in data, "OIDC response missing 'authorization_endpoint'" + assert "token_endpoint" in data, "OIDC response missing 'token_endpoint'" + + def test_authentik_client_credentials_present(self, original_env: dict) -> None: + """Validate that Authentik client credentials are configured alongside the config URL.""" + client_id = original_env.get("AUTHENTIK_CLIENT_ID") + client_secret = original_env.get("AUTHENTIK_CLIENT_SECRET") + + if client_id and client_secret: + assert len(client_id) > 0, "AUTHENTIK_CLIENT_ID should not be empty" + assert len(client_secret) > 0, "AUTHENTIK_CLIENT_SECRET should not be empty" + else: + pytest.skip("AUTHENTIK_CLIENT_ID and/or AUTHENTIK_CLIENT_SECRET not set") + + +# --------------------------------------------------------------------------- +# Full pipeline: Azure OCR → OpenAI metadata extraction (end-to-end) +# --------------------------------------------------------------------------- +@pytest.mark.requires_external +@pytest.mark.skipif( + not has_real_env("AZURE_AI_KEY", "AZURE_ENDPOINT", "OPENAI_API_KEY"), + reason="Real Azure + OpenAI credentials required for full pipeline test", +) +class TestFullOCRMetadataPipeline: + """End-to-end pipeline: generate PDF → Azure OCR → OpenAI metadata extraction. + + This replicates the core DocuElevate processing flow without requiring + Celery or Redis, by calling the service APIs directly. + """ + + def test_ocr_then_metadata_extraction(self, original_env: dict) -> None: + """Generate a PDF, OCR it with Azure, then extract metadata with OpenAI.""" + import re + + import openai + from azure.ai.documentintelligence import DocumentIntelligenceClient + from azure.ai.documentintelligence.models import AnalyzeOutputOption + from azure.core.credentials import AzureKeyCredential + + pdf_path = generate_test_pdf() + try: + # --- Step 1: Azure OCR --- + doc_client = DocumentIntelligenceClient( + endpoint=original_env["AZURE_ENDPOINT"], + credential=AzureKeyCredential(original_env["AZURE_AI_KEY"]), + ) + + with open(pdf_path, "rb") as f: + poller = doc_client.begin_analyze_document( + "prebuilt-read", + body=f, + output=[AnalyzeOutputOption.PDF], + ) + result = poller.result() + extracted_text = result.content + assert extracted_text and len(extracted_text) > 10, "OCR produced insufficient text" + + # --- Step 2: OpenAI metadata extraction --- + api_key = original_env["OPENAI_API_KEY"] + base_url = original_env.get("OPENAI_BASE_URL") or "https://api.openai.com/v1" + oai = openai.OpenAI(api_key=api_key, base_url=base_url) + + prompt = ( + "You are a specialized document analyzer. Analyze the following OCR-extracted text " + "and return a JSON object with these fields:\n" + '- "document_type": classification (e.g. Invoice, Contract, Letter)\n' + '- "language": ISO 639-1 code\n' + '- "absender": sender\n' + '- "empfaenger": recipient\n' + '- "tags": up to 4 keywords\n' + '- "confidence_score": 0-100\n\n' + f"OCR text:\n{extracted_text}\n\n" + "Return only valid JSON." + ) + + completion = oai.chat.completions.create( + model="gpt-4o-mini", + messages=[ + {"role": "system", "content": "You are an intelligent document classifier."}, + {"role": "user", "content": prompt}, + ], + temperature=0, + ) + + content = completion.choices[0].message.content + assert content, "OpenAI returned empty response" + + json_match = re.search(r"\{.*\}", content, re.DOTALL) + assert json_match, f"No JSON found in response: {content[:200]}" + + metadata = json.loads(json_match.group()) + + # Validate key metadata fields + assert "document_type" in metadata, "Missing document_type" + assert "language" in metadata, "Missing language" + assert "absender" in metadata, "Missing absender" + assert "empfaenger" in metadata, "Missing empfaenger" + + # The generated invoice should be classified reasonably + doc_type = metadata["document_type"].lower() + assert any( + kw in doc_type for kw in ("invoice", "rechnung", "bill") + ), f"Unexpected document_type: {metadata['document_type']}" + finally: + os.unlink(pdf_path) + + +# --------------------------------------------------------------------------- +# Configuration: verify that Settings correctly loads external env vars +# --------------------------------------------------------------------------- +@pytest.mark.unit +class TestExternalEnvVarConfiguration: + """Verify that the Settings class correctly reads external API environment variables. + + These tests do NOT call external APIs; they only validate that the configuration + layer correctly maps environment variables to Settings fields. + """ + + def test_settings_reads_openai_base_url(self) -> None: + """Test that OPENAI_BASE_URL is correctly loaded into Settings.""" + from app.config import Settings + + custom_url = "https://custom-openai.example.com/v1" + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + openai_base_url=custom_url, + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + ) + assert config.openai_base_url == custom_url + + def test_settings_reads_s3_configuration(self) -> None: + """Test that S3-related settings are correctly loaded.""" + from app.config import Settings + + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + aws_access_key_id="AKIAEXAMPLE", + aws_secret_access_key="secretkey", + s3_bucket_name="my-bucket", + s3_folder_prefix="uploads/", + ) + assert config.aws_access_key_id == "AKIAEXAMPLE" + assert config.aws_secret_access_key == "secretkey" + assert config.s3_bucket_name == "my-bucket" + assert config.s3_folder_prefix == "uploads/" + + def test_settings_reads_onedrive_configuration(self) -> None: + """Test that OneDrive-related settings are correctly loaded.""" + from app.config import Settings + + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + onedrive_client_id="client-123", + onedrive_client_secret="secret-456", + onedrive_tenant_id="tenant-789", + onedrive_refresh_token="refresh-abc", + onedrive_folder_path="Documents/Test", + ) + assert config.onedrive_client_id == "client-123" + assert config.onedrive_client_secret == "secret-456" + assert config.onedrive_tenant_id == "tenant-789" + assert config.onedrive_refresh_token == "refresh-abc" + assert config.onedrive_folder_path == "Documents/Test" + + def test_settings_reads_dropbox_configuration(self) -> None: + """Test that Dropbox-related settings are correctly loaded.""" + from app.config import Settings + + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + dropbox_app_key="dbx-key", + dropbox_app_secret="dbx-secret", + dropbox_refresh_token="dbx-refresh", + ) + assert config.dropbox_app_key == "dbx-key" + assert config.dropbox_app_secret == "dbx-secret" + assert config.dropbox_refresh_token == "dbx-refresh" + + def test_settings_reads_authentik_configuration(self) -> None: + """Test that Authentik/OIDC settings are correctly loaded.""" + from app.config import Settings + + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + authentik_client_id="auth-client", + authentik_client_secret="auth-secret", + authentik_config_url="https://auth.example.com/.well-known/openid-configuration", + ) + assert config.authentik_client_id == "auth-client" + assert config.authentik_client_secret == "auth-secret" + assert config.authentik_config_url == "https://auth.example.com/.well-known/openid-configuration" + + def test_settings_optional_services_default_to_none(self) -> None: + """Test that optional external service settings default to None when not provided.""" + from app.config import Settings + + config = Settings( + database_url="sqlite:///test.db", + redis_url="redis://localhost:6379", + openai_api_key="test", + azure_ai_key="test", + azure_region="test", + azure_endpoint="https://test.example.com", + gotenberg_url="http://localhost:3000", + workdir="/tmp", + auth_enabled=False, + ) + assert config.aws_access_key_id is None + assert config.aws_secret_access_key is None + assert config.s3_bucket_name is None + assert config.onedrive_client_id is None + assert config.onedrive_client_secret is None + assert config.onedrive_refresh_token is None + assert config.dropbox_app_key is None + assert config.dropbox_app_secret is None + assert config.dropbox_refresh_token is None + assert config.authentik_client_id is None + assert config.authentik_client_secret is None + assert config.authentik_config_url is None + + +# --------------------------------------------------------------------------- +# Test-PDF generator unit tests +# --------------------------------------------------------------------------- +@pytest.mark.unit +class TestPdfGenerator: + """Verify that the test PDF generator produces valid PDFs with extractable text.""" + + def test_generate_default_pdf(self) -> None: + """Test that generate_test_pdf creates a valid PDF with embedded text.""" + import PyPDF2 + + path = generate_test_pdf() + try: + assert os.path.exists(path) + assert os.path.getsize(path) > 100 + + with open(path, "rb") as f: + reader = PyPDF2.PdfReader(f) + assert len(reader.pages) >= 1 + text = reader.pages[0].extract_text() + assert "Invoice" in text + assert "Acme" in text + finally: + os.unlink(path) + + def test_generate_custom_content_pdf(self) -> None: + """Test that generate_test_pdf accepts custom content.""" + import PyPDF2 + + custom = "Custom test content for verification" + path = generate_test_pdf(content=custom) + try: + with open(path, "rb") as f: + reader = PyPDF2.PdfReader(f) + text = reader.pages[0].extract_text() + assert "Custom test content" in text + finally: + os.unlink(path) + + def test_generated_pdfs_are_unique(self) -> None: + """Test that consecutive calls produce different PDFs.""" + path1 = generate_test_pdf() + path2 = generate_test_pdf() + try: + with open(path1, "rb") as f1, open(path2, "rb") as f2: + assert f1.read() != f2.read(), "Two generated PDFs should differ" + finally: + os.unlink(path1) + os.unlink(path2)