diff --git a/e2e/test_entities.py b/e2e/test_entities.py new file mode 100644 index 0000000000..2ba9945052 --- /dev/null +++ b/e2e/test_entities.py @@ -0,0 +1,429 @@ +"""E2E tests for the entities API. + +These tests verify basic entity operations work correctly when running +against a fully deployed NMP platform. This includes: +- Entity CRUD operations (create, retrieve, update, delete) +- Entity creation within and without projects +- Listing with sorting and filtering + +Note: Many other e2e tests implicitly test the entities API since most +services are built on top of it. These tests provide a direct indicator +for deeper problems in the entities service itself. +""" + +import json +import time +import uuid + +import pytest +from nemo_platform import APIStatusError, NeMoPlatform + +ENTITY_TYPE = "e2e-test-entity" + + +def _unique_name(prefix: str = "entity") -> str: + """Generate a unique entity name.""" + return f"{prefix}-{uuid.uuid4().hex[:8]}" + + +def test_cluster_info_endpoint_returns_json_with_platform_version_and_revision(sdk: NeMoPlatform): + """Test GET /cluster-info returns JSON with platform_version and revision keys. + + Verifies the platform cluster-info endpoint returns a json-encoded response + and includes platform_version and revision fields (values are not validated). + """ + response = sdk._client.get("/cluster-info") + assert response.status_code == 200 + data = response.json() + assert isinstance(data, dict), "Response body should be JSON object" + assert "platform_version" in data, "Response should include a 'platform_version' key" + assert "revision" in data, "Response should include a 'revision' key" + + +def test_entity_crud_lifecycle(sdk: NeMoPlatform, workspace: str): + """Test basic entity create, retrieve, update, delete operations. + + This test verifies the complete entity lifecycle: + 1. Create an entity with specific data + 2. Retrieve it by name and verify contents + 3. Update the entity data + 4. Delete the entity + 5. Verify it no longer exists + """ + entity_name = _unique_name() + initial_data = {"key": "initial-value", "nested": {"field": 123}} + + # Create entity + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=entity_name, + data=initial_data, + ) + assert entity.name == entity_name + assert entity.workspace == workspace + assert entity.entity_type == ENTITY_TYPE + assert entity.data["key"] == "initial-value" + assert entity.data["nested"]["field"] == 123 # ty: ignore[not-subscriptable] + + try: + # Retrieve by name + retrieved = sdk.entities.get_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved.name == entity_name + assert retrieved.id == entity.id + assert retrieved.data == initial_data + + # Update entity + updated_data = {"key": "updated-value", "nested": {"field": 456}, "new_field": True} + updated = sdk.entities.update_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + data=updated_data, + ) + assert updated.name == entity_name + assert updated.data["key"] == "updated-value" + assert updated.data["nested"]["field"] == 456 # ty: ignore[not-subscriptable] + assert updated.data["new_field"] is True + + # Verify update persisted + retrieved_after_update = sdk.entities.get_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved_after_update.data == updated_data + + finally: + # Delete entity + sdk.entities.delete_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + + # Verify entity no longer exists + with pytest.raises(APIStatusError) as exc_info: + sdk.entities.get_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert exc_info.value.status_code == 404 + + +def test_entity_with_project(sdk: NeMoPlatform, workspace: str): + """Test entity creation within a project. + + Verifies that entities can be associated with projects and that + the project association is correctly persisted and retrievable. + """ + project_name = _unique_name("project") + entity_name = _unique_name() + + # Create project first + project = sdk.projects.create( + workspace=workspace, + name=project_name, + description="E2E test project", + ) + assert project.name == project_name + + try: + # Create entity within project + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=entity_name, + data={"project_data": "value"}, + project=project_name, + ) + assert entity.name == entity_name + assert entity.project == project_name + + # Retrieve and verify project association + retrieved = sdk.entities.get_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved.project == project_name + + # Delete entity + sdk.entities.delete_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + + finally: + # Clean up project + sdk.projects.delete(name=project_name, workspace=workspace) + + +def test_entity_without_project(sdk: NeMoPlatform, workspace: str): + """Test entity creation without a project association. + + Verifies that entities can exist at the workspace level without + being associated with any project. + """ + entity_name = _unique_name() + + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=entity_name, + data={"standalone": True}, + ) + + try: + assert entity.name == entity_name + assert entity.project is None + + retrieved = sdk.entities.get_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved.project is None + + finally: + sdk.entities.delete_entity_by_name( + name=entity_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + + +def test_entity_list_and_sorting(sdk: NeMoPlatform, workspace: str): + """Test listing entities with sorting. + + Creates multiple entities and verifies: + 1. All entities are returned in list + 2. Sorting by created_at works (ascending and descending) + 3. Sorting by name works + """ + entity_names = [_unique_name(f"sort-{i:02d}") for i in range(5)] + created_entities = [] + + try: + # Create entities in order + for name in entity_names: + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=name, + data={"order": name}, + ) + time.sleep(1) + created_entities.append(entity) + + # List all entities of this type + response = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + listed_names = {e.name for e in response.data} + for name in entity_names: + assert name in listed_names + + # Test descending sort by created_at (default, newest first) + response_desc = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + sort="-created_at", + ) + desc_names = [e.name for e in response_desc.data if e.name in entity_names] + assert desc_names == list(reversed(entity_names)) + + # Test ascending sort by created_at (oldest first) + response_asc = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + sort="created_at", + ) + asc_names = [e.name for e in response_asc.data if e.name in entity_names] + assert asc_names == entity_names + + # Test sort by name + response_by_name = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + sort="name", + ) + name_sorted = [e.name for e in response_by_name.data if e.name in entity_names] + assert name_sorted == sorted(entity_names) + + finally: + # Clean up all created entities + for name in entity_names: + try: + sdk.entities.delete_entity_by_name( + name=name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + except Exception: + pass + + +def test_entity_search_filter(sdk: NeMoPlatform, workspace: str): + """Test filtering entities with search queries. + + Verifies that the search parameter correctly filters entities + based on field values. + """ + prefix = _unique_name("filter") + entity_alpha = f"{prefix}-alpha" + entity_beta = f"{prefix}-beta" + + try: + # Create two entities with different data + sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=entity_alpha, + data={"category": "alpha", "value": 100}, + ) + sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=entity_beta, + data={"category": "beta", "value": 200}, + ) + + # Filter by exact name match + filter_query = json.dumps({"name": {"$eq": entity_alpha}}) + response = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + filter=filter_query, + ) + assert len(response.data) == 1 + assert response.data[0].name == entity_alpha + + # Filter by name pattern (like) + filter_query = json.dumps({"name": {"$like": f"{prefix}%"}}) + response = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + filter=filter_query, + ) + found_names = {e.name for e in response.data} + assert entity_alpha in found_names + assert entity_beta in found_names + + # Filter by data field + filter_query = json.dumps({"data.category": {"$eq": "beta"}}) + response = sdk.entities.list( + entity_type=ENTITY_TYPE, + workspace=workspace, + filter=filter_query, + ) + assert len(response.data) == 1 + assert response.data[0].name == entity_beta + + finally: + for name in [entity_alpha, entity_beta]: + try: + sdk.entities.delete_entity_by_name( + name=name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + except Exception: + pass + + +def test_entity_rename(sdk: NeMoPlatform, workspace: str): + """Test renaming an entity via update. + + Verifies that entities can be renamed and the old name + no longer works after rename. + """ + old_name = _unique_name("old") + new_name = _unique_name("new") + + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + name=old_name, + data={"test": "rename"}, + ) + + try: + # Rename entity + renamed = sdk.entities.update_entity_by_name( + name=old_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + data=entity.data, + new_name=new_name, + ) + assert renamed.name == new_name + assert renamed.id == entity.id + + # Verify old name no longer works + with pytest.raises(APIStatusError) as exc_info: + sdk.entities.get_entity_by_name( + name=old_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert exc_info.value.status_code == 404 + + # Verify new name works + retrieved = sdk.entities.get_entity_by_name( + name=new_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved.name == new_name + + finally: + # Clean up with new name + try: + sdk.entities.delete_entity_by_name( + name=new_name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + except Exception: + pass + + +def test_entity_auto_generated_name(sdk: NeMoPlatform, workspace: str): + """Test that entities can be created without specifying a name. + + When no name is provided, the API should auto-generate a unique name. + """ + entity = sdk.entities.create( + entity_type=ENTITY_TYPE, + workspace=workspace, + data={"auto_name": True}, + ) + + try: + assert entity.name is not None + assert len(entity.name) > 0 + # Auto-generated names typically follow a pattern like "e2e-test-entity-xxxxx" + assert ENTITY_TYPE.replace("_", "-").replace("-", "") in entity.name.replace("-", "") or entity.name + + # Verify we can retrieve by the generated name + retrieved = sdk.entities.get_entity_by_name( + name=entity.name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) + assert retrieved.id == entity.id + + finally: + sdk.entities.delete_entity_by_name( + name=entity.name, + entity_type=ENTITY_TYPE, + workspace=workspace, + ) diff --git a/e2e/test_files.py b/e2e/test_files.py new file mode 100644 index 0000000000..121097d710 --- /dev/null +++ b/e2e/test_files.py @@ -0,0 +1,209 @@ +"""E2E tests for the files service. + +These tests verify basic file upload and download operations +work correctly when running against a fully deployed NMP platform. +""" + +import tempfile +import uuid +from collections.abc import Iterator +from pathlib import Path + +import pytest +from nemo_platform import NeMoPlatform +from nemo_platform.types.files import Fileset + + +@pytest.fixture +def fileset(sdk: NeMoPlatform, workspace: str) -> Iterator[Fileset]: + """Create a unique fileset for each test with automatic cleanup.""" + fileset_name = f"e2e-fileset-{uuid.uuid4().hex[:8]}" + fileset = sdk.files.filesets.create(workspace=workspace, name=fileset_name) + yield fileset + try: + sdk.files.filesets.delete(fileset_name, workspace=workspace) + except Exception: + pass # Ignore cleanup errors + + +def test_file_upload_and_download(sdk: NeMoPlatform, workspace: str, fileset: Fileset): + """Test uploading and downloading a file. + + This test verifies the files system works end-to-end: + 1. Upload a file with test content + 2. Download the file and verify content matches + """ + test_content = b"Hello from e2e test! This is test file content." + + with tempfile.TemporaryDirectory() as tmpdir: + # Create local file to upload + local_file = Path(tmpdir, "test.txt") + local_file.write_bytes(test_content) + + # Upload file using high-level API + sdk.files.upload( + fileset=fileset.name, + workspace=workspace, + local_path=str(local_file), + remote_path="test.txt", + ) + + # Verify file was uploaded + files = sdk.files.list(fileset=fileset.name, workspace=workspace) + assert len(files.data) == 1 + assert files.data[0].path == "test.txt" + assert files.data[0].size == len(test_content) + + # Download file and verify content + download_path = Path(tmpdir, "downloaded.txt") + sdk.files.download( + fileset=fileset.name, + workspace=workspace, + remote_path="test.txt", + local_path=str(download_path), + ) + assert download_path.read_bytes() == test_content + + +def test_file_list_cache_status_for_default_storage(sdk: NeMoPlatform, workspace: str, fileset: Fileset): + """Test cache status reporting for files stored in the default backend.""" + test_content = b"cache status coverage" + + sdk.files.upload_content( + fileset=fileset.name, + workspace=workspace, + remote_path="cache-status.txt", + content=test_content, + ) + + files_without_cache_check = sdk.files.list(fileset=fileset.name, workspace=workspace) + assert len(files_without_cache_check.data) == 1 + assert files_without_cache_check.data[0].cache_status == "not_cacheable" + + files_with_cache_check = sdk.files.list( + fileset=fileset.name, + workspace=workspace, + include_cache_status=True, + ) + assert len(files_with_cache_check.data) == 1 + assert files_with_cache_check.data[0].cache_status == "not_cacheable" + + +def test_file_upload_nested_path(sdk: NeMoPlatform, workspace: str, fileset: Fileset): + """Test uploading a file with a nested path. + + Verifies that files can be uploaded to nested directories + within a fileset. + """ + test_content = b"Nested file content" + test_path = "folder/subfolder/nested.txt" + + with tempfile.TemporaryDirectory() as tmpdir: + # Create local file to upload + local_file = Path(tmpdir, "nested.txt") + local_file.write_bytes(test_content) + + # Upload file to nested path + sdk.files.upload( + fileset=fileset.name, + workspace=workspace, + local_path=str(local_file), + remote_path=test_path, + ) + + # List files and verify the nested file appears + files = sdk.files.list(fileset=fileset.name, workspace=workspace) + file_paths = {f.path for f in files.data} + assert test_path in file_paths + + # Download and verify + download_path = Path(tmpdir, "downloaded.txt") + sdk.files.download( + fileset=fileset.name, + workspace=workspace, + remote_path=test_path, + local_path=str(download_path), + ) + assert download_path.read_bytes() == test_content + + +def test_file_delete(sdk: NeMoPlatform, workspace: str, fileset: Fileset): + """Test deleting a file from a fileset. + + Verifies that files can be deleted and are no longer + accessible after deletion. + """ + test_content = b"File to be deleted" + test_path = "delete-me.txt" + + with tempfile.TemporaryDirectory() as tmpdir: + # Create local file to upload + local_file = Path(tmpdir, "delete-me.txt") + local_file.write_bytes(test_content) + + # Upload file + sdk.files.upload( + fileset=fileset.name, + workspace=workspace, + local_path=str(local_file), + remote_path=test_path, + ) + + # Verify file exists by listing + files = sdk.files.list(fileset=fileset.name, workspace=workspace) + assert any(f.path == test_path for f in files.data) + + # Delete file + sdk.files.delete( + fileset=fileset.name, + workspace=workspace, + remote_path=test_path, + ) + + # Verify file is gone + files = sdk.files.list(fileset=fileset.name, workspace=workspace) + assert not any(f.path == test_path for f in files.data) + + +def test_directory_upload_and_download(sdk: NeMoPlatform, workspace: str, fileset: Fileset): + """Test uploading and downloading a directory. + + Verifies that entire directories can be uploaded and downloaded + with their structure preserved. + """ + with tempfile.TemporaryDirectory() as tmpdir: + # Create directory structure to upload + upload_dir = Path(tmpdir, "upload") + upload_dir.mkdir() + (upload_dir / "file1.txt").write_text("content1") + (upload_dir / "file2.txt").write_text("content2") + subdir = upload_dir / "subdir" + subdir.mkdir() + (subdir / "file3.txt").write_text("content3") + + # Upload entire directory + sdk.files.upload( + fileset=fileset.name, + workspace=workspace, + local_path=f"{upload_dir}/", + remote_path="", + ) + + # Verify all files were uploaded + files = sdk.files.list(fileset=fileset.name, workspace=workspace) + paths = {f.path for f in files.data} + assert paths == {"file1.txt", "file2.txt", "subdir/file3.txt"} + + # Download entire fileset + download_dir = Path(tmpdir, "download") + download_dir.mkdir() + sdk.files.download( + fileset=fileset.name, + workspace=workspace, + local_path=f"{download_dir}/", + ) + + # Verify downloaded content matches + assert (download_dir / "file1.txt").read_text() == "content1" + assert (download_dir / "file2.txt").read_text() == "content2" + assert (download_dir / "subdir" / "file3.txt").read_text() == "content3" diff --git a/e2e/test_secrets.py b/e2e/test_secrets.py new file mode 100644 index 0000000000..7ec8f14909 --- /dev/null +++ b/e2e/test_secrets.py @@ -0,0 +1,185 @@ +"""E2E tests for the secrets service. + +These tests verify basic secret creation and listing operations +work correctly when running against a fully deployed NMP platform. +""" + +import uuid + +from nemo_platform import NeMoPlatform + + +def test_secret_create_and_list(sdk: NeMoPlatform, workspace: str): + """Test creating a secret and listing it in the workspace. + + This test verifies the secrets system works end-to-end: + 1. Create a secret with a test value + 2. Verify the secret appears in the list of workspace secrets + 3. Verify the secret can be retrieved + """ + secret_name = f"e2e-secret-{uuid.uuid4().hex[:8]}" + secret_value = "e2e-test-secret-value" + + # Create a secret + secret = sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + assert secret.name == secret_name + assert secret.workspace == workspace + + # List secrets and verify the new secret appears + list_response = sdk.secrets.list(workspace=workspace) + secret_names = [s.name for s in list_response.data] + assert secret_name in secret_names + + # Retrieve the secret to verify it was created correctly + retrieved_secret = sdk.secrets.retrieve(secret_name, workspace=workspace) + assert retrieved_secret.name == secret_name + assert retrieved_secret.workspace == workspace + + +def test_secret_create_duplicate_fails(sdk: NeMoPlatform, workspace: str): + """Test that creating a secret with a duplicate name fails. + + This test verifies that the secrets system enforces unique + secret names within a workspace. + """ + secret_name = f"e2e-duplicate-secret-{uuid.uuid4().hex[:8]}" + secret_value = "e2e-duplicate-test-secret-value" + + # Create the initial secret + sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + + # Attempt to create a duplicate secret and expect failure + try: + sdk.secrets.create( + workspace=workspace, + name=secret_name, + value="some-other-value", + ) + assert False, "Expected an exception when creating a duplicate secret" + except Exception as e: + # Verify that the exception indicates a duplicate resource + assert "already exists" in str(e) or "duplicate" in str(e) + + +def test_secret_create_and_delete(sdk: NeMoPlatform, workspace: str): + """Test creating and deleting a secret. + + This test verifies that a secret can be created and then deleted, + and that it no longer appears in the list of secrets after deletion. + """ + secret_name = f"e2e-delete-secret-{uuid.uuid4().hex[:8]}" + secret_value = "e2e-delete-test-secret-value" + + # Create a secret + sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + + # Verify the secret appears in the list + list_response = sdk.secrets.list(workspace=workspace) + secret_names = [s.name for s in list_response.data] + assert secret_name in secret_names + + # Delete the secret + sdk.secrets.delete( + workspace=workspace, + name=secret_name, + ) + + # Verify the secret no longer appears in the list + list_response = sdk.secrets.list(workspace=workspace) + secret_names = [s.name for s in list_response.data] + assert secret_name not in secret_names + + +def test_secret_data_not_in_create_response(sdk: NeMoPlatform, workspace: str): + """Test that secret data is not exposed in the create response. + + This test verifies that when creating a secret, the response does not + contain the secret value - only metadata like name and workspace. + """ + secret_name = f"e2e-no-data-create-{uuid.uuid4().hex[:8]}" + secret_value = "this-should-not-appear-in-response" + + secret = sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + + # Verify name and workspace are present + assert secret.name == secret_name + assert secret.workspace == workspace + + # Verify the secret value is not exposed in the response object + # The SDK object should not have a 'data' attribute with the secret value + secret_dict = secret.model_dump() + assert "data" not in secret_dict or secret_dict.get("data") is None + assert "_data" not in secret_dict + + +def test_secret_data_not_in_retrieve_response(sdk: NeMoPlatform, workspace: str): + """Test that secret data is not exposed in the retrieve response. + + This test verifies that when retrieving a secret by name, the response + does not contain the secret value - only metadata. + """ + secret_name = f"e2e-no-data-retrieve-{uuid.uuid4().hex[:8]}" + secret_value = "this-should-not-appear-in-retrieve" + + sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + + # Retrieve the secret + retrieved = sdk.secrets.retrieve(secret_name, workspace=workspace) + + # Verify name and workspace are present + assert retrieved.name == secret_name + assert retrieved.workspace == workspace + + # Verify the secret value is not exposed + secret_dict = retrieved.model_dump() + assert "data" not in secret_dict or secret_dict.get("data") is None + assert "_data" not in secret_dict + + +def test_secret_data_not_in_list_response(sdk: NeMoPlatform, workspace: str): + """Test that secret data is not exposed in the list response. + + This test verifies that when listing secrets, none of the secrets + in the response contain their actual values. + """ + secret_name = f"e2e-no-data-list-{uuid.uuid4().hex[:8]}" + secret_value = "this-should-not-appear-in-list" + + sdk.secrets.create( + workspace=workspace, + name=secret_name, + value=secret_value, + ) + + # List secrets + list_response = sdk.secrets.list(workspace=workspace) + + # Find our secret in the list + our_secret = next((s for s in list_response.data if s.name == secret_name), None) + assert our_secret is not None, "Created secret should appear in list" + + # Verify no secrets in the list expose their values + for secret in list_response.data: + secret_dict = secret.model_dump() + assert "data" not in secret_dict or secret_dict.get("data") is None + assert "_data" not in secret_dict diff --git a/e2e/test_studio.py b/e2e/test_studio.py new file mode 100644 index 0000000000..f8a7576e0d --- /dev/null +++ b/e2e/test_studio.py @@ -0,0 +1,135 @@ +"""E2E tests for the Studio UI service. + +These tests verify that the Studio static files are correctly served +and that the production build is properly configured. + +Note: These tests require the Studio UI to be built and available. +All tests self-skip when Studio static files are not mounted. +""" + +import re + +from nemo_platform import NeMoPlatform +from nmp.testing.pytest_outcomes import pytest_skip + + +def _studio_available(sdk: NeMoPlatform) -> bool: + """Check if the Studio UI is available (static files are mounted).""" + response = sdk._client.get("/studio/") + return response.status_code == 200 + + +def test_studio_index_html(sdk: NeMoPlatform): + """Test that /studio/ serves the index.html correctly. + + This verifies: + 1. The Studio service is mounted and serving static files from static_files_path + 2. The response is HTML with correct content-type + 3. Assets are prefixed with /studio/ (correct Vite base URL) + 4. No unreplaced STUDIO_UI_ markers remain in the HTML + """ + if not _studio_available(sdk): + pytest_skip("Studio UI not available (static files not mounted)") + + response = sdk._client.get("/studio/") + + assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}" + assert "text/html" in response.headers.get("content-type", ""), ( + f"Expected text/html content-type, got: {response.headers.get('content-type')}" + ) + + html = response.text + + # Basic HTML structure checks + assert "