diff --git a/README.md b/README.md index cd959f1..3685187 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ A Python client library for Microsoft Dataverse that provides a unified interfac - [Bulk operations](#bulk-operations) - [Query data](#query-data) - [Table management](#table-management) + - [Relationship management](#relationship-management) - [File operations](#file-operations) - [Next steps](#next-steps) - [Troubleshooting](#troubleshooting) @@ -36,6 +37,7 @@ A Python client library for Microsoft Dataverse that provides a unified interfac - **⚡ True Bulk Operations**: Automatically uses Dataverse's native `CreateMultiple`, `UpdateMultiple`, and `BulkDelete` Web API operations for maximum performance and transactional integrity - **📊 SQL Queries**: Execute read-only SQL queries via the Dataverse Web API `?sql=` parameter - **🏗️ Table Management**: Create, inspect, and delete custom tables and columns programmatically +- **🔗 Relationship Management**: Create one-to-many and many-to-many relationships between tables with full metadata control - **📎 File Operations**: Upload files to Dataverse file columns with automatic chunking for large files - **🔐 Azure Identity**: Built-in authentication using Azure Identity credential providers with comprehensive support - **🛡️ Error Handling**: Structured exception hierarchy with detailed error context and retry guidance @@ -259,9 +261,71 @@ client.delete_columns("new_Product", ["new_Category"]) client.delete_table("new_Product") ``` -> **Important**: All custom column names must include the customization prefix value (e.g., `"new_"`). +> **Important**: All custom column names must include the customization prefix value (e.g., `"new_"`). > This ensures explicit, predictable naming and aligns with Dataverse metadata requirements. +### Relationship management + +Create relationships between tables using the relationship API. For a complete working example, see [examples/advanced/relationships.py](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/advanced/relationships.py). + +```python +from PowerPlatform.Dataverse.models.metadata import ( + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + ManyToManyRelationshipMetadata, + Label, + LocalizedLabel, +) + +# Create a one-to-many relationship: Department (1) -> Employee (N) +# This adds a "Department" lookup field to the Employee table +lookup = LookupAttributeMetadata( + schema_name="new_DepartmentId", + display_name=Label(localized_labels=[LocalizedLabel(label="Department", language_code=1033)]), +) + +relationship = OneToManyRelationshipMetadata( + schema_name="new_Department_Employee", + referenced_entity="new_department", # Parent table (the "one" side) + referencing_entity="new_employee", # Child table (the "many" side) + referenced_attribute="new_departmentid", +) + +result = client.create_one_to_many_relationship(lookup, relationship) +print(f"Created lookup field: {result['lookup_schema_name']}") + +# Create a many-to-many relationship: Employee (N) <-> Project (N) +# Employees work on multiple projects; projects have multiple team members +m2m_relationship = ManyToManyRelationshipMetadata( + schema_name="new_employee_project", + entity1_logical_name="new_employee", + entity2_logical_name="new_project", +) + +result = client.create_many_to_many_relationship(m2m_relationship) +print(f"Created M:N relationship: {result['relationship_schema_name']}") + +# Query relationship metadata +rel = client.get_relationship("new_Department_Employee") +if rel: + print(f"Found: {rel['SchemaName']}") + +# Delete a relationship +client.delete_relationship(result['relationship_id']) +``` + +For simpler scenarios, use the convenience method: + +```python +# Quick way to create a lookup field with sensible defaults +result = client.create_lookup_field( + referencing_table="contact", # Child table gets the lookup field + lookup_field_name="new_AccountId", + referenced_table="account", # Parent table being referenced + display_name="Account", +) +``` + ### File operations ```python @@ -285,7 +349,8 @@ Explore our comprehensive examples in the [`examples/`](https://github.com/micro - **[Functional Testing](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/basic/functional_testing.py)** - Test core functionality in your environment **🚀 Advanced Usage:** -- **[Complete Walkthrough](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/advanced/walkthrough.py)** - Full feature demonstration with production patterns +- **[Complete Walkthrough](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/advanced/walkthrough.py)** - Full feature demonstration with production patterns +- **[Relationship Management](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/advanced/relationships.py)** - Create and manage table relationships - **[File Upload](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/advanced/file_upload.py)** - Upload files to Dataverse file columns 📖 See the [examples README](https://github.com/microsoft/PowerPlatform-DataverseClient-Python/blob/main/examples/README.md) for detailed guidance and learning progression. @@ -347,8 +412,7 @@ For optimal performance in production environments: ### Limitations - SQL queries are **read-only** and support a limited subset of SQL syntax -- Create Table supports a limited number of column types. Lookup columns are not yet supported. -- Creating relationships between tables is not yet supported. +- Create Table supports a limited number of column types (string, int, decimal, bool, datetime, picklist) - File uploads are limited by Dataverse file size restrictions (default 128MB per file) ## Contributing diff --git a/examples/advanced/relationships.py b/examples/advanced/relationships.py new file mode 100644 index 0000000..087c465 --- /dev/null +++ b/examples/advanced/relationships.py @@ -0,0 +1,387 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Relationship Management Example for Dataverse SDK. + +This example demonstrates: +- Creating one-to-many relationships using the core SDK API +- Creating lookup fields using the convenience method +- Creating many-to-many relationships +- Querying and deleting relationships +- Working with relationship metadata types + +Prerequisites: +- pip install PowerPlatform-Dataverse-Client +- pip install azure-identity +""" + +import sys +import time +from azure.identity import InteractiveBrowserCredential +from PowerPlatform.Dataverse.client import DataverseClient +from PowerPlatform.Dataverse.models.metadata import ( + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + ManyToManyRelationshipMetadata, + Label, + LocalizedLabel, + CascadeConfiguration, +) + + +# Simple logging helper +def log_call(description): + print(f"\n-> {description}") + + +def delete_relationship_if_exists(client, schema_name): + """Delete a relationship by schema name if it exists.""" + rel = client.get_relationship(schema_name) + if rel: + rel_id = rel.get("MetadataId") + if rel_id: + client.delete_relationship(rel_id) + print(f" (Cleaned up existing relationship: {schema_name})") + return True + return False + + +def cleanup_previous_run(client): + """Clean up any resources from a previous run to make the example idempotent.""" + print("\n-> Checking for resources from previous runs...") + + # Known relationship names created by this example + relationships = [ + "new_Department_Employee", + "contact_new_employee_new_ManagerId", + "new_employee_project", + ] + + # Known table names created by this example + tables = ["new_Employee", "new_Department", "new_Project"] + + # Delete relationships first (required before tables can be deleted) + for rel_name in relationships: + try: + delete_relationship_if_exists(client, rel_name) + except Exception as e: + print(f" [WARN] Could not delete relationship {rel_name}: {e}") + + # Delete tables + for table_name in tables: + try: + if client.get_table_info(table_name): + client.delete_table(table_name) + print(f" (Cleaned up existing table: {table_name})") + except Exception as e: + print(f" [WARN] Could not delete table {table_name}: {e}") + + +def backoff(op, *, delays=(0, 2, 5, 10, 20, 20)): + """Retry helper with exponential backoff.""" + last = None + total_delay = 0 + attempts = 0 + for d in delays: + if d: + time.sleep(d) + total_delay += d + attempts += 1 + try: + result = op() + if attempts > 1: + retry_count = attempts - 1 + print(f" * Backoff succeeded after {retry_count} retry(s); waited {total_delay}s total.") + return result + except Exception as ex: # noqa: BLE001 + last = ex + continue + if last: + if attempts: + retry_count = max(attempts - 1, 0) + print(f" [WARN] Backoff exhausted after {retry_count} retry(s); waited {total_delay}s total.") + raise last + + +def main(): + # Initialize relationship IDs to None for cleanup safety + rel_id_1 = None + rel_id_2 = None + rel_id_3 = None + + print("=" * 80) + print("Dataverse SDK - Relationship Management Example") + print("=" * 80) + + # ============================================================================ + # 1. SETUP & AUTHENTICATION + # ============================================================================ + print("\n" + "=" * 80) + print("1. Setup & Authentication") + print("=" * 80) + + base_url = input("Enter Dataverse org URL (e.g. https://yourorg.crm.dynamics.com): ").strip() + if not base_url: + print("No URL entered; exiting.") + sys.exit(1) + + base_url = base_url.rstrip("/") + + log_call("InteractiveBrowserCredential()") + credential = InteractiveBrowserCredential() + + log_call(f"DataverseClient(base_url='{base_url}', credential=...)") + client = DataverseClient(base_url=base_url, credential=credential) + print(f"[OK] Connected to: {base_url}") + + # ============================================================================ + # 2. CLEANUP PREVIOUS RUN (Idempotency) + # ============================================================================ + print("\n" + "=" * 80) + print("2. Cleanup Previous Run (Idempotency)") + print("=" * 80) + + cleanup_previous_run(client) + + # ============================================================================ + # 3. CREATE SAMPLE TABLES + # ============================================================================ + print("\n" + "=" * 80) + print("3. Create Sample Tables") + print("=" * 80) + + # Create a parent table (Department) + log_call("Creating 'new_Department' table") + + dept_table = backoff( + lambda: client.create_table( + "new_Department", + { + "new_DepartmentCode": "string", + "new_Budget": "decimal", + }, + ) + ) + print(f"[OK] Created table: {dept_table['table_schema_name']}") + + # Create a child table (Employee) + log_call("Creating 'new_Employee' table") + + emp_table = backoff( + lambda: client.create_table( + "new_Employee", + { + "new_EmployeeNumber": "string", + "new_Salary": "decimal", + }, + ) + ) + print(f"[OK] Created table: {emp_table['table_schema_name']}") + + # Create a project table for many-to-many example + log_call("Creating 'new_Project' table") + + proj_table = backoff( + lambda: client.create_table( + "new_Project", + { + "new_ProjectCode": "string", + "new_StartDate": "datetime", + }, + ) + ) + print(f"[OK] Created table: {proj_table['table_schema_name']}") + + # ============================================================================ + # 4. CREATE ONE-TO-MANY RELATIONSHIP (Core SDK API) + # ============================================================================ + print("\n" + "=" * 80) + print("4. Create One-to-Many Relationship (Core API)") + print("=" * 80) + + log_call("Creating lookup field on Employee referencing Department") + + # Define the lookup attribute metadata + lookup = LookupAttributeMetadata( + schema_name="new_DepartmentId", + display_name=Label(localized_labels=[LocalizedLabel(label="Department", language_code=1033)]), + required_level="None", + ) + + # Define the relationship metadata + relationship = OneToManyRelationshipMetadata( + schema_name="new_Department_Employee", + referenced_entity=dept_table["table_logical_name"], + referencing_entity=emp_table["table_logical_name"], + referenced_attribute=f"{dept_table['table_logical_name']}id", + cascade_configuration=CascadeConfiguration( + delete="RemoveLink", # When department is deleted, remove the link but keep employees + assign="NoCascade", + merge="NoCascade", + ), + ) + + # Create the relationship + result = backoff( + lambda: client.create_one_to_many_relationship( + lookup=lookup, + relationship=relationship, + ) + ) + + print(f"[OK] Created relationship: {result['relationship_schema_name']}") + print(f" Lookup field: {result['lookup_schema_name']}") + print(f" Relationship ID: {result['relationship_id']}") + + rel_id_1 = result["relationship_id"] + + # ============================================================================ + # 5. CREATE LOOKUP FIELD (Convenience Method) + # ============================================================================ + print("\n" + "=" * 80) + print("5. Create Lookup Field (Convenience Method)") + print("=" * 80) + + log_call("Creating lookup field on Employee referencing Contact as Manager") + + # Use the convenience method for simpler scenarios + # An Employee has a Manager (who is a Contact in the system) + result2 = backoff( + lambda: client.create_lookup_field( + referencing_table=emp_table["table_logical_name"], + lookup_field_name="new_ManagerId", + referenced_table="contact", + display_name="Manager", + description="The employee's direct manager", + required=False, + cascade_delete="RemoveLink", + ) + ) + + print(f"[OK] Created lookup using convenience method: {result2['lookup_schema_name']}") + print(f" Relationship: {result2['relationship_schema_name']}") + + rel_id_2 = result2["relationship_id"] + + # ============================================================================ + # 6. CREATE MANY-TO-MANY RELATIONSHIP + # ============================================================================ + print("\n" + "=" * 80) + print("6. Create Many-to-Many Relationship") + print("=" * 80) + + log_call("Creating M:N relationship between Employee and Project") + + # Define many-to-many relationship + m2m_relationship = ManyToManyRelationshipMetadata( + schema_name="new_employee_project", + entity1_logical_name=emp_table["table_logical_name"], + entity2_logical_name=proj_table["table_logical_name"], + ) + + result3 = backoff( + lambda: client.create_many_to_many_relationship( + relationship=m2m_relationship, + ) + ) + + print(f"[OK] Created M:N relationship: {result3['relationship_schema_name']}") + print(f" Relationship ID: {result3['relationship_id']}") + + rel_id_3 = result3["relationship_id"] + + # ============================================================================ + # 7. QUERY RELATIONSHIP METADATA + # ============================================================================ + print("\n" + "=" * 80) + print("7. Query Relationship Metadata") + print("=" * 80) + + log_call("Retrieving 1:N relationship by schema name") + + rel_metadata = client.get_relationship("new_Department_Employee") + if rel_metadata: + print(f"[OK] Found relationship: {rel_metadata.get('SchemaName')}") + print(f" Type: {rel_metadata.get('@odata.type')}") + print(f" Referenced Entity: {rel_metadata.get('ReferencedEntity')}") + print(f" Referencing Entity: {rel_metadata.get('ReferencingEntity')}") + else: + print(" Relationship not found") + + log_call("Retrieving M:N relationship by schema name") + + m2m_metadata = client.get_relationship("new_employee_project") + if m2m_metadata: + print(f"[OK] Found relationship: {m2m_metadata.get('SchemaName')}") + print(f" Type: {m2m_metadata.get('@odata.type')}") + print(f" Entity 1: {m2m_metadata.get('Entity1LogicalName')}") + print(f" Entity 2: {m2m_metadata.get('Entity2LogicalName')}") + else: + print(" Relationship not found") + + # ============================================================================ + # 8. CLEANUP + # ============================================================================ + print("\n" + "=" * 80) + print("8. Cleanup") + print("=" * 80) + + cleanup = input("\nDelete created relationships and tables? (y/n): ").strip().lower() + + if cleanup == "y": + # Delete relationships first (required before deleting tables) + log_call("Deleting relationships") + try: + if rel_id_1: + backoff(lambda: client.delete_relationship(rel_id_1)) + print(f" [OK] Deleted relationship: new_Department_Employee") + except Exception as e: + print(f" [WARN] Error deleting relationship 1: {e}") + + try: + if rel_id_2: + backoff(lambda: client.delete_relationship(rel_id_2)) + print(f" [OK] Deleted relationship: contact->employee (Manager)") + except Exception as e: + print(f" [WARN] Error deleting relationship 2: {e}") + + try: + if rel_id_3: + backoff(lambda: client.delete_relationship(rel_id_3)) + print(f" [OK] Deleted relationship: new_employee_project") + except Exception as e: + print(f" [WARN] Error deleting relationship 3: {e}") + + # Delete tables + log_call("Deleting tables") + for table_name in ["new_Employee", "new_Department", "new_Project"]: + try: + backoff(lambda name=table_name: client.delete_table(name)) + print(f" [OK] Deleted table: {table_name}") + except Exception as e: + print(f" [WARN] Error deleting {table_name}: {e}") + + print("\n[OK] Cleanup complete") + else: + print("\nSkipping cleanup. Remember to manually delete:") + print(" - Relationships: new_Department_Employee, contact->employee (Manager), new_employee_project") + print(" - Tables: new_Employee, new_Department, new_Project") + + print("\n" + "=" * 80) + print("Example Complete!") + print("=" * 80) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\n\nExample interrupted by user.") + sys.exit(1) + except Exception as e: + print(f"\n\nError: {e}") + import traceback + + traceback.print_exc() + sys.exit(1) diff --git a/src/PowerPlatform/Dataverse/client.py b/src/PowerPlatform/Dataverse/client.py index 84bd5d4..448ee1c 100644 --- a/src/PowerPlatform/Dataverse/client.py +++ b/src/PowerPlatform/Dataverse/client.py @@ -11,6 +11,14 @@ from .core._auth import _AuthManager from .core.config import DataverseConfig from .data._odata import _ODataClient +from .models.metadata import ( + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + ManyToManyRelationshipMetadata, + Label, + LocalizedLabel, + CascadeConfiguration, +) class DataverseClient: @@ -436,7 +444,7 @@ def create_table( :param columns: Dictionary mapping column names (with customization prefix value) to their types. All custom column names must include the customization prefix value (e.g. ``"new_Title"``). Supported types: - - Primitive types: ``"string"`` (alias: ``"text"``), ``"int"`` (alias: ``"integer"``), ``"decimal"`` (alias: ``"money"``), ``"float"`` (alias: ``"double"``), ``"datetime"`` (alias: ``"date"``), ``"bool"`` (alias: ``"boolean"``) + - Primitive types: ``"string"``, ``"int"``, ``"decimal"``, ``"float"``, ``"datetime"``, ``"bool"`` - Enum subclass (IntEnum preferred): Creates a local option set. Optional multilingual labels can be provided via ``__labels__`` class attribute, defined inside the Enum subclass:: @@ -546,7 +554,7 @@ def create_columns( :param table_schema_name: Schema name of the table (e.g. ``"new_MyTestTable"``). :type table_schema_name: :class:`str` :param columns: Mapping of column schema names (with customization prefix value) to supported types. All custom column names must include the customization prefix value** (e.g. ``"new_Notes"``). Primitive types include - ``"string"`` (alias: ``"text"``), ``"int"`` (alias: ``"integer"``), ``"decimal"`` (alias: ``"money"``), ``"float"`` (alias: ``"double"``), ``"datetime"`` (alias: ``"date"``), and ``"bool"`` (alias: ``"boolean"``). Enum subclasses (IntEnum preferred) + ``string``, ``int``, ``decimal``, ``float``, ``datetime``, and ``bool``. Enum subclasses (IntEnum preferred) generate a local option set and can specify localized labels via ``__labels__``. :type columns: :class:`dict` mapping :class:`str` to :class:`typing.Any` :returns: Schema names for the columns that were created. @@ -698,5 +706,249 @@ def flush_cache(self, kind) -> int: with self._scoped_odata() as od: return od._flush_cache(kind) + # Relationship operations + def create_one_to_many_relationship( + self, + lookup: LookupAttributeMetadata, + relationship: OneToManyRelationshipMetadata, + *, + solution: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Create a one-to-many relationship between tables. + + This operation creates both the relationship and the lookup attribute + on the referencing table. + + :param lookup: Metadata defining the lookup attribute. + :type lookup: ~PowerPlatform.Dataverse.models.metadata.LookupAttributeMetadata + :param relationship: Metadata defining the relationship. + :type relationship: ~PowerPlatform.Dataverse.models.metadata.OneToManyRelationshipMetadata + :param solution: Optional solution unique name to add relationship to. + :type solution: :class:`str` or None + + :return: Dictionary with relationship_id, lookup_schema_name, and related metadata. + :rtype: :class:`dict` + + :raises ~PowerPlatform.Dataverse.core.errors.HttpError: If the Web API request fails. + + Example: + Create a one-to-many relationship: Department (1) -> Employee (N):: + + from PowerPlatform.Dataverse.models.metadata import ( + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + Label, + LocalizedLabel, + CascadeConfiguration, + ) + + # Define the lookup attribute (added to Employee table) + lookup = LookupAttributeMetadata( + schema_name="new_DepartmentId", + display_name=Label( + localized_labels=[ + LocalizedLabel(label="Department", language_code=1033) + ] + ), + ) + + # Define the relationship + relationship = OneToManyRelationshipMetadata( + schema_name="new_Department_Employee", + referenced_entity="new_department", # Parent table (the "one" side) + referencing_entity="new_employee", # Child table (the "many" side) + referenced_attribute="new_departmentid", + cascade_configuration=CascadeConfiguration( + delete="RemoveLink", # When department deleted, unlink employees + ), + ) + + result = client.create_one_to_many_relationship(lookup, relationship) + print(f"Created lookup field: {result['lookup_schema_name']}") + """ + with self._scoped_odata() as od: + return od._create_one_to_many_relationship( + lookup, + relationship, + solution, + ) + + def create_many_to_many_relationship( + self, + relationship: ManyToManyRelationshipMetadata, + *, + solution: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Create a many-to-many relationship between tables. + + This operation creates a many-to-many relationship and an intersect table + to manage the relationship. + + :param relationship: Metadata defining the many-to-many relationship. + :type relationship: ~PowerPlatform.Dataverse.models.metadata.ManyToManyRelationshipMetadata + :param solution: Optional solution unique name to add relationship to. + :type solution: :class:`str` or None + + :return: Dictionary with relationship_id, relationship_schema_name, and entity names. + :rtype: :class:`dict` + + :raises ~PowerPlatform.Dataverse.core.errors.HttpError: If the Web API request fails. + + Example: + Create a many-to-many relationship: Employee (N) <-> Project (N):: + + from PowerPlatform.Dataverse.models.metadata import ( + ManyToManyRelationshipMetadata, + ) + + # Employees work on multiple projects; projects have multiple team members + relationship = ManyToManyRelationshipMetadata( + schema_name="new_employee_project", + entity1_logical_name="new_employee", + entity2_logical_name="new_project", + ) + + result = client.create_many_to_many_relationship(relationship) + print(f"Created M:N relationship: {result['relationship_schema_name']}") + """ + with self._scoped_odata() as od: + return od._create_many_to_many_relationship( + relationship, + solution, + ) + + def delete_relationship(self, relationship_id: str) -> None: + """ + Delete a relationship by its metadata ID. + + :param relationship_id: The GUID of the relationship metadata. + :type relationship_id: :class:`str` + + :raises ~PowerPlatform.Dataverse.core.errors.HttpError: If the Web API request fails. + + .. warning:: + Deleting a relationship also removes the associated lookup attribute + for one-to-many relationships. This operation is irreversible. + + Example: + Delete a relationship:: + + client.delete_relationship("12345678-1234-1234-1234-123456789abc") + """ + with self._scoped_odata() as od: + od._delete_relationship(relationship_id) + + def get_relationship(self, schema_name: str) -> Optional[Dict[str, Any]]: + """ + Retrieve relationship metadata by schema name. + + :param schema_name: The schema name of the relationship. + :type schema_name: :class:`str` + + :return: Relationship metadata dictionary, or None if not found. + :rtype: :class:`dict` or None + + :raises ~PowerPlatform.Dataverse.core.errors.HttpError: If the Web API request fails. + + Example: + Get relationship metadata:: + + rel = client.get_relationship("new_Department_Employee") + if rel: + print(f"Found relationship: {rel['SchemaName']}") + """ + with self._scoped_odata() as od: + return od._get_relationship(schema_name) + + def create_lookup_field( + self, + referencing_table: str, + lookup_field_name: str, + referenced_table: str, + *, + display_name: Optional[str] = None, + description: Optional[str] = None, + required: bool = False, + cascade_delete: str = "RemoveLink", + solution: Optional[str] = None, + language_code: int = 1033, + ) -> Dict[str, Any]: + """ + Create a simple lookup field relationship. + + This is a convenience method that wraps :meth:`create_one_to_many_relationship` + for the common case of adding a lookup field to an existing table. + + :param referencing_table: Logical name of the table that will have the lookup field (child table). + :type referencing_table: :class:`str` + :param lookup_field_name: Schema name for the lookup field (e.g., ``"new_AccountId"``). + :type lookup_field_name: :class:`str` + :param referenced_table: Logical name of the table being referenced (parent table). + :type referenced_table: :class:`str` + :param display_name: Display name for the lookup field. Defaults to the referenced table name. + :type display_name: :class:`str` or None + :param description: Optional description for the lookup field. + :type description: :class:`str` or None + :param required: Whether the lookup is required. Defaults to ``False``. + :type required: :class:`bool` + :param cascade_delete: Delete behavior (``"RemoveLink"``, ``"Cascade"``, ``"Restrict"``). + Defaults to ``"RemoveLink"``. + :type cascade_delete: :class:`str` + :param solution: Optional solution unique name to add the relationship to. + :type solution: :class:`str` or None + :param language_code: Language code for labels. Defaults to 1033 (English). + :type language_code: :class:`int` + + :return: Dictionary with ``relationship_id``, ``lookup_schema_name``, and related metadata. + :rtype: :class:`dict` + + :raises ~PowerPlatform.Dataverse.core.errors.HttpError: If the Web API request fails. + + Example: + Create a simple lookup field:: + + result = client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + display_name="Account", + required=True, + cascade_delete="RemoveLink" + ) + + print(f"Created lookup: {result['lookup_schema_name']}") + """ + # Build the label + localized_labels = [LocalizedLabel(label=display_name or referenced_table, language_code=language_code)] + + # Build the lookup attribute + lookup = LookupAttributeMetadata( + schema_name=lookup_field_name, + display_name=Label(localized_labels=localized_labels), + required_level="ApplicationRequired" if required else "None", + ) + + # Add description if provided + if description: + lookup.description = Label( + localized_labels=[LocalizedLabel(label=description, language_code=language_code)] + ) + + # Generate a relationship name + relationship_name = f"{referenced_table}_{referencing_table}_{lookup_field_name}" + + # Build the relationship metadata + relationship = OneToManyRelationshipMetadata( + schema_name=relationship_name, + referenced_entity=referenced_table, + referencing_entity=referencing_table, + referenced_attribute=f"{referenced_table}id", + cascade_configuration=CascadeConfiguration(delete=cascade_delete), + ) + + return self.create_one_to_many_relationship(lookup, relationship, solution=solution) + __all__ = ["DataverseClient"] diff --git a/src/PowerPlatform/Dataverse/common/__init__.py b/src/PowerPlatform/Dataverse/common/__init__.py new file mode 100644 index 0000000..049d80d --- /dev/null +++ b/src/PowerPlatform/Dataverse/common/__init__.py @@ -0,0 +1,10 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Common utilities and constants for the Dataverse SDK. + +This module contains shared constants and utilities used across the SDK. +""" + +__all__ = [] diff --git a/src/PowerPlatform/Dataverse/common/constants.py b/src/PowerPlatform/Dataverse/common/constants.py new file mode 100644 index 0000000..e725f86 --- /dev/null +++ b/src/PowerPlatform/Dataverse/common/constants.py @@ -0,0 +1,22 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Constants for Dataverse Web API metadata types. + +These constants define the OData type identifiers used in Web API payloads +for metadata operations. +""" + +# OData type identifiers for metadata entities +ODATA_TYPE_LOCALIZED_LABEL = "Microsoft.Dynamics.CRM.LocalizedLabel" +ODATA_TYPE_LABEL = "Microsoft.Dynamics.CRM.Label" +ODATA_TYPE_LOOKUP_ATTRIBUTE = "Microsoft.Dynamics.CRM.LookupAttributeMetadata" +ODATA_TYPE_ONE_TO_MANY_RELATIONSHIP = "Microsoft.Dynamics.CRM.OneToManyRelationshipMetadata" +ODATA_TYPE_MANY_TO_MANY_RELATIONSHIP = "Microsoft.Dynamics.CRM.ManyToManyRelationshipMetadata" + +# Cascade behavior values for relationship operations +CASCADE_BEHAVIOR_CASCADE = "Cascade" +CASCADE_BEHAVIOR_NO_CASCADE = "NoCascade" +CASCADE_BEHAVIOR_REMOVE_LINK = "RemoveLink" +CASCADE_BEHAVIOR_RESTRICT = "Restrict" diff --git a/src/PowerPlatform/Dataverse/data/_odata.py b/src/PowerPlatform/Dataverse/data/_odata.py index 7c5fc6c..024285c 100644 --- a/src/PowerPlatform/Dataverse/data/_odata.py +++ b/src/PowerPlatform/Dataverse/data/_odata.py @@ -19,7 +19,8 @@ from contextvars import ContextVar from ..core._http import _HttpClient -from ._upload import _ODataFileUpload +from ._upload import _FileUploadMixin +from ._relationships import _RelationshipOperationsMixin from ..core.errors import * from ..core._error_codes import ( _http_subcode, @@ -76,7 +77,7 @@ def build( ) -class _ODataClient(_ODataFileUpload): +class _ODataClient(_FileUploadMixin, _RelationshipOperationsMixin): """Dataverse Web API client: CRUD, SQL-over-API, and table metadata helpers.""" @staticmethod diff --git a/src/PowerPlatform/Dataverse/data/_relationships.py b/src/PowerPlatform/Dataverse/data/_relationships.py new file mode 100644 index 0000000..57e31d5 --- /dev/null +++ b/src/PowerPlatform/Dataverse/data/_relationships.py @@ -0,0 +1,158 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Relationship metadata operations for Dataverse Web API. + +This module provides mixin functionality for relationship CRUD operations. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, Optional + + +class _RelationshipOperationsMixin: + """ + Mixin providing relationship metadata operations. + + This mixin is designed to be used with _ODataClient and depends on: + - self.api: The API base URL + - self._headers(): Method to get auth headers + - self._request(): Method to make HTTP requests + """ + + def _create_one_to_many_relationship( + self, + lookup, + relationship, + solution: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Create a one-to-many relationship with lookup attribute. + + Posts to /RelationshipDefinitions with OneToManyRelationshipMetadata. + + :param lookup: Lookup attribute metadata (LookupAttributeMetadata instance). + :type lookup: ~PowerPlatform.Dataverse.models.metadata.LookupAttributeMetadata + :param relationship: Relationship metadata (OneToManyRelationshipMetadata instance). + :type relationship: ~PowerPlatform.Dataverse.models.metadata.OneToManyRelationshipMetadata + :param solution: Optional solution unique name to add the relationship to. + :type solution: ``str`` | ``None`` + + :return: Dictionary with relationship_id, attribute_id, and schema names. + :rtype: ``dict[str, Any]`` + + :raises HttpError: If the Web API request fails. + """ + url = f"{self.api}/RelationshipDefinitions" + + # Build the payload by combining relationship and lookup metadata + payload = relationship.to_dict() + payload["Lookup"] = lookup.to_dict() + + headers = self._headers().copy() + if solution: + headers["MSCRM.SolutionUniqueName"] = solution + + r = self._request("post", url, headers=headers, json=payload) + + # Extract IDs from response headers + relationship_id = self._extract_id_from_header(r.headers.get("OData-EntityId")) + + return { + "relationship_id": relationship_id, + "relationship_schema_name": relationship.schema_name, + "lookup_schema_name": lookup.schema_name, + "referenced_entity": relationship.referenced_entity, + "referencing_entity": relationship.referencing_entity, + } + + def _create_many_to_many_relationship( + self, + relationship, + solution: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Create a many-to-many relationship. + + Posts to /RelationshipDefinitions with ManyToManyRelationshipMetadata. + + :param relationship: Relationship metadata (ManyToManyRelationshipMetadata instance). + :type relationship: ~PowerPlatform.Dataverse.models.metadata.ManyToManyRelationshipMetadata + :param solution: Optional solution unique name to add the relationship to. + :type solution: ``str`` | ``None`` + + :return: Dictionary with relationship_id and schema name. + :rtype: ``dict[str, Any]`` + + :raises HttpError: If the Web API request fails. + """ + url = f"{self.api}/RelationshipDefinitions" + + payload = relationship.to_dict() + + headers = self._headers().copy() + if solution: + headers["MSCRM.SolutionUniqueName"] = solution + + r = self._request("post", url, headers=headers, json=payload) + + # Extract ID from response header + relationship_id = self._extract_id_from_header(r.headers.get("OData-EntityId")) + + return { + "relationship_id": relationship_id, + "relationship_schema_name": relationship.schema_name, + "entity1_logical_name": relationship.entity1_logical_name, + "entity2_logical_name": relationship.entity2_logical_name, + } + + def _delete_relationship(self, relationship_id: str) -> None: + """ + Delete a relationship by its metadata ID. + + :param relationship_id: The GUID of the relationship metadata. + :type relationship_id: ``str`` + + :raises HttpError: If the Web API request fails. + """ + url = f"{self.api}/RelationshipDefinitions({relationship_id})" + headers = self._headers().copy() + headers["If-Match"] = "*" + self._request("delete", url, headers=headers) + + def _get_relationship(self, schema_name: str) -> Optional[Dict[str, Any]]: + """ + Retrieve relationship metadata by schema name. + + :param schema_name: The schema name of the relationship. + :type schema_name: ``str`` + + :return: Relationship metadata dictionary, or None if not found. + :rtype: ``dict[str, Any]`` | ``None`` + + :raises HttpError: If the Web API request fails. + """ + url = f"{self.api}/RelationshipDefinitions" + params = {"$filter": f"SchemaName eq '{self._escape_odata_quotes(schema_name)}'"} + r = self._request("get", url, headers=self._headers(), params=params) + data = r.json() + results = data.get("value", []) + return results[0] if results else None + + def _extract_id_from_header(self, header_value: Optional[str]) -> Optional[str]: + """ + Extract a GUID from an OData-EntityId header value. + + :param header_value: The header value containing a URL with GUID. + :type header_value: ``str`` | ``None`` + + :return: Extracted GUID or None if not found. + :rtype: ``str`` | ``None`` + """ + if not header_value: + return None + match = re.search(r"\(([0-9a-fA-F-]+)\)", header_value) + return match.group(1) if match else None diff --git a/src/PowerPlatform/Dataverse/data/_upload.py b/src/PowerPlatform/Dataverse/data/_upload.py index d82efb5..80c951f 100644 --- a/src/PowerPlatform/Dataverse/data/_upload.py +++ b/src/PowerPlatform/Dataverse/data/_upload.py @@ -8,7 +8,7 @@ from typing import Optional -class _ODataFileUpload: +class _FileUploadMixin: """File upload capabilities (small + chunk) with auto selection.""" def _upload_file( diff --git a/src/PowerPlatform/Dataverse/models/__init__.py b/src/PowerPlatform/Dataverse/models/__init__.py index 396cd4e..d37d084 100644 --- a/src/PowerPlatform/Dataverse/models/__init__.py +++ b/src/PowerPlatform/Dataverse/models/__init__.py @@ -1,9 +1,6 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. -""" -Data models and type definitions for the Dataverse SDK. Currently a placeholder. -""" +"""Data models for Dataverse metadata types.""" -# Will be populated with models as they are created __all__ = [] diff --git a/src/PowerPlatform/Dataverse/models/metadata.py b/src/PowerPlatform/Dataverse/models/metadata.py new file mode 100644 index 0000000..7696c6f --- /dev/null +++ b/src/PowerPlatform/Dataverse/models/metadata.py @@ -0,0 +1,396 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Metadata entity types for Microsoft Dataverse. + +These classes represent the metadata entity types used in the Dataverse Web API +for defining and managing table definitions, attributes, and relationships. + +See: https://learn.microsoft.com/en-us/power-apps/developer/data-platform/webapi/reference/metadataentitytypes +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional +from dataclasses import dataclass, field + +from ..common.constants import ( + ODATA_TYPE_LOCALIZED_LABEL, + ODATA_TYPE_LABEL, + ODATA_TYPE_LOOKUP_ATTRIBUTE, + ODATA_TYPE_ONE_TO_MANY_RELATIONSHIP, + ODATA_TYPE_MANY_TO_MANY_RELATIONSHIP, + CASCADE_BEHAVIOR_CASCADE, + CASCADE_BEHAVIOR_NO_CASCADE, + CASCADE_BEHAVIOR_REMOVE_LINK, + CASCADE_BEHAVIOR_RESTRICT, +) + + +@dataclass +class LocalizedLabel: + """ + Represents a localized label with a language code. + + :param label: The text of the label. + :type label: str + :param language_code: The language code (LCID), e.g., 1033 for English. + :type language_code: int + :param additional_properties: Optional dict of additional properties to include + in the Web API payload. These are merged last and can override default values. + :type additional_properties: Optional[Dict[str, Any]] + """ + + label: str + language_code: int + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> label = LocalizedLabel(label="Account", language_code=1033) + >>> label.to_dict() + { + '@odata.type': 'Microsoft.Dynamics.CRM.LocalizedLabel', + 'Label': 'Account', + 'LanguageCode': 1033 + } + """ + result = { + "@odata.type": ODATA_TYPE_LOCALIZED_LABEL, + "Label": self.label, + "LanguageCode": self.language_code, + } + if self.additional_properties: + result.update(self.additional_properties) + return result + + +@dataclass +class Label: + """ + Represents a label that can have multiple localized versions. + + :param localized_labels: List of LocalizedLabel instances. + :type localized_labels: List[LocalizedLabel] + :param user_localized_label: Optional user-specific localized label. + :type user_localized_label: Optional[LocalizedLabel] + :param additional_properties: Optional dict of additional properties to include + in the Web API payload. These are merged last and can override default values. + :type additional_properties: Optional[Dict[str, Any]] + """ + + localized_labels: List[LocalizedLabel] + user_localized_label: Optional[LocalizedLabel] = None + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> label = Label(localized_labels=[LocalizedLabel("Account", 1033)]) + >>> label.to_dict() + { + '@odata.type': 'Microsoft.Dynamics.CRM.Label', + 'LocalizedLabels': [ + {'@odata.type': '...', 'Label': 'Account', 'LanguageCode': 1033} + ], + 'UserLocalizedLabel': {'@odata.type': '...', 'Label': 'Account', ...} + } + """ + result = { + "@odata.type": ODATA_TYPE_LABEL, + "LocalizedLabels": [ll.to_dict() for ll in self.localized_labels], + } + # Use explicit user_localized_label, or default to first localized label + if self.user_localized_label: + result["UserLocalizedLabel"] = self.user_localized_label.to_dict() + elif self.localized_labels: + result["UserLocalizedLabel"] = self.localized_labels[0].to_dict() + if self.additional_properties: + result.update(self.additional_properties) + return result + + +@dataclass +class CascadeConfiguration: + """ + Defines cascade behavior for relationship operations. + + :param assign: Cascade behavior for assign operations. + :type assign: str + :param delete: Cascade behavior for delete operations. + :type delete: str + :param merge: Cascade behavior for merge operations. + :type merge: str + :param reparent: Cascade behavior for reparent operations. + :type reparent: str + :param share: Cascade behavior for share operations. + :type share: str + :param unshare: Cascade behavior for unshare operations. + :type unshare: str + :param additional_properties: Optional dict of additional properties to include + in the Web API payload (e.g., "Archive", "RollupView"). These are merged + last and can override default values. + :type additional_properties: Optional[Dict[str, Any]] + + Valid values for each parameter: + - "Cascade": Perform the operation on all related records + - "NoCascade": Do not perform the operation on related records + - "RemoveLink": Remove the relationship link but keep the records + - "Restrict": Prevent the operation if related records exist + """ + + assign: str = CASCADE_BEHAVIOR_NO_CASCADE + delete: str = CASCADE_BEHAVIOR_REMOVE_LINK + merge: str = CASCADE_BEHAVIOR_NO_CASCADE + reparent: str = CASCADE_BEHAVIOR_NO_CASCADE + share: str = CASCADE_BEHAVIOR_NO_CASCADE + unshare: str = CASCADE_BEHAVIOR_NO_CASCADE + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> config = CascadeConfiguration(delete="Cascade", assign="NoCascade") + >>> config.to_dict() + { + 'Assign': 'NoCascade', + 'Delete': 'Cascade', + 'Merge': 'NoCascade', + 'Reparent': 'NoCascade', + 'Share': 'NoCascade', + 'Unshare': 'NoCascade' + } + """ + result = { + "Assign": self.assign, + "Delete": self.delete, + "Merge": self.merge, + "Reparent": self.reparent, + "Share": self.share, + "Unshare": self.unshare, + } + if self.additional_properties: + result.update(self.additional_properties) + return result + + +@dataclass +class LookupAttributeMetadata: + """ + Metadata for a lookup attribute. + + :param schema_name: Schema name for the attribute (e.g., "new_AccountId"). + :type schema_name: str + :param display_name: Display name for the attribute. + :type display_name: Label + :param description: Optional description of the attribute. + :type description: Optional[Label] + :param required_level: Requirement level for the attribute. + :type required_level: str + :param additional_properties: Optional dict of additional properties to include + in the Web API payload. Useful for setting properties like "Targets" (to + specify which entity types the lookup can reference), "LogicalName", + "IsSecured", "IsValidForAdvancedFind", etc. These are merged last and + can override default values. + :type additional_properties: Optional[Dict[str, Any]] + + Valid required_level values: + - "None": The attribute is optional + - "Recommended": The attribute is recommended + - "ApplicationRequired": The attribute is required + """ + + schema_name: str + display_name: Label + description: Optional[Label] = None + required_level: str = "None" + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> lookup = LookupAttributeMetadata( + ... schema_name="new_AccountId", + ... display_name=Label([LocalizedLabel("Account", 1033)]) + ... ) + >>> lookup.to_dict() + { + '@odata.type': 'Microsoft.Dynamics.CRM.LookupAttributeMetadata', + 'SchemaName': 'new_AccountId', + 'AttributeType': 'Lookup', + 'AttributeTypeName': {'Value': 'LookupType'}, + 'DisplayName': {...}, + 'RequiredLevel': {'Value': 'None', 'CanBeChanged': True, ...} + } + """ + result = { + "@odata.type": ODATA_TYPE_LOOKUP_ATTRIBUTE, + "SchemaName": self.schema_name, + "AttributeType": "Lookup", + "AttributeTypeName": {"Value": "LookupType"}, + "DisplayName": self.display_name.to_dict(), + "RequiredLevel": { + "Value": self.required_level, + "CanBeChanged": True, + "ManagedPropertyLogicalName": "canmodifyrequirementlevelsettings", + }, + } + if self.description: + result["Description"] = self.description.to_dict() + if self.additional_properties: + result.update(self.additional_properties) + return result + + +@dataclass +class OneToManyRelationshipMetadata: + """ + Metadata for a one-to-many entity relationship. + + :param schema_name: Schema name for the relationship (e.g., "new_Account_Orders"). + :type schema_name: str + :param referenced_entity: Logical name of the referenced (parent) entity. + :type referenced_entity: str + :param referencing_entity: Logical name of the referencing (child) entity. + :type referencing_entity: str + :param referenced_attribute: Attribute on the referenced entity (typically the primary key). + :type referenced_attribute: str + :param cascade_configuration: Cascade behavior configuration. + :type cascade_configuration: CascadeConfiguration + :param referencing_attribute: Optional name for the referencing attribute (usually auto-generated). + :type referencing_attribute: Optional[str] + :param additional_properties: Optional dict of additional properties to include + in the Web API payload. Useful for setting inherited properties like + "IsValidForAdvancedFind", "IsCustomizable", "SecurityTypes", etc. + These are merged last and can override default values. + :type additional_properties: Optional[Dict[str, Any]] + """ + + schema_name: str + referenced_entity: str + referencing_entity: str + referenced_attribute: str + cascade_configuration: CascadeConfiguration = field(default_factory=CascadeConfiguration) + referencing_attribute: Optional[str] = None + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> rel = OneToManyRelationshipMetadata( + ... schema_name="new_account_orders", + ... referenced_entity="account", + ... referencing_entity="new_order", + ... referenced_attribute="accountid" + ... ) + >>> rel.to_dict() + { + '@odata.type': 'Microsoft.Dynamics.CRM.OneToManyRelationshipMetadata', + 'SchemaName': 'new_account_orders', + 'ReferencedEntity': 'account', + 'ReferencingEntity': 'new_order', + 'ReferencedAttribute': 'accountid', + 'CascadeConfiguration': {...} + } + """ + result = { + "@odata.type": ODATA_TYPE_ONE_TO_MANY_RELATIONSHIP, + "SchemaName": self.schema_name, + "ReferencedEntity": self.referenced_entity, + "ReferencingEntity": self.referencing_entity, + "ReferencedAttribute": self.referenced_attribute, + "CascadeConfiguration": self.cascade_configuration.to_dict(), + } + if self.referencing_attribute: + result["ReferencingAttribute"] = self.referencing_attribute + if self.additional_properties: + result.update(self.additional_properties) + return result + + +@dataclass +class ManyToManyRelationshipMetadata: + """ + Metadata for a many-to-many entity relationship. + + :param schema_name: Schema name for the relationship. + :type schema_name: str + :param entity1_logical_name: Logical name of the first entity. + :type entity1_logical_name: str + :param entity2_logical_name: Logical name of the second entity. + :type entity2_logical_name: str + :param intersect_entity_name: Name for the intersect table (defaults to schema_name if not provided). + :type intersect_entity_name: Optional[str] + :param additional_properties: Optional dict of additional properties to include + in the Web API payload. Useful for setting inherited properties like + "IsValidForAdvancedFind", "IsCustomizable", "SecurityTypes", or direct + properties like "Entity1NavigationPropertyName". These are merged last + and can override default values. + :type additional_properties: Optional[Dict[str, Any]] + """ + + schema_name: str + entity1_logical_name: str + entity2_logical_name: str + intersect_entity_name: Optional[str] = None + additional_properties: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + """ + Convert to Web API JSON format. + + Example:: + + >>> rel = ManyToManyRelationshipMetadata( + ... schema_name="new_account_contact", + ... entity1_logical_name="account", + ... entity2_logical_name="contact" + ... ) + >>> rel.to_dict() + { + '@odata.type': 'Microsoft.Dynamics.CRM.ManyToManyRelationshipMetadata', + 'SchemaName': 'new_account_contact', + 'Entity1LogicalName': 'account', + 'Entity2LogicalName': 'contact', + 'IntersectEntityName': 'new_account_contact' + } + """ + # IntersectEntityName is required - use provided value or default to schema_name + intersect_name = self.intersect_entity_name or self.schema_name + result = { + "@odata.type": ODATA_TYPE_MANY_TO_MANY_RELATIONSHIP, + "SchemaName": self.schema_name, + "Entity1LogicalName": self.entity1_logical_name, + "Entity2LogicalName": self.entity2_logical_name, + "IntersectEntityName": intersect_name, + } + if self.additional_properties: + result.update(self.additional_properties) + return result + + +__all__ = [ + "LocalizedLabel", + "Label", + "CascadeConfiguration", + "LookupAttributeMetadata", + "OneToManyRelationshipMetadata", + "ManyToManyRelationshipMetadata", +] diff --git a/tests/unit/data/test_relationships.py b/tests/unit/data/test_relationships.py new file mode 100644 index 0000000..581c0a4 --- /dev/null +++ b/tests/unit/data/test_relationships.py @@ -0,0 +1,294 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for relationship metadata operations.""" + +import unittest +from unittest.mock import MagicMock, Mock + +from PowerPlatform.Dataverse.data._relationships import _RelationshipOperationsMixin +from PowerPlatform.Dataverse.models.metadata import ( + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + ManyToManyRelationshipMetadata, + Label, + LocalizedLabel, +) + + +class TestExtractIdFromHeader(unittest.TestCase): + """Tests for _extract_id_from_header method.""" + + def setUp(self): + """Create a minimal mixin instance for testing.""" + self.mixin = _RelationshipOperationsMixin() + + def test_extract_id_from_standard_header(self): + """Test extracting GUID from standard OData-EntityId header.""" + header = "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)" + result = self.mixin._extract_id_from_header(header) + self.assertEqual(result, "12345678-1234-1234-1234-123456789abc") + + def test_extract_id_from_header_uppercase_guid(self): + """Test extracting uppercase GUID.""" + header = "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(ABCDEF12-3456-7890-ABCD-EF1234567890)" + result = self.mixin._extract_id_from_header(header) + self.assertEqual(result, "ABCDEF12-3456-7890-ABCD-EF1234567890") + + def test_extract_id_from_none_header(self): + """Test that None header returns None.""" + result = self.mixin._extract_id_from_header(None) + self.assertIsNone(result) + + def test_extract_id_from_empty_header(self): + """Test that empty header returns None.""" + result = self.mixin._extract_id_from_header("") + self.assertIsNone(result) + + def test_extract_id_from_header_without_guid(self): + """Test that header without GUID returns None.""" + header = "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions" + result = self.mixin._extract_id_from_header(header) + self.assertIsNone(result) + + +class MockODataClient(_RelationshipOperationsMixin): + """Mock client that inherits from mixin for integration testing.""" + + def __init__(self, api_base: str): + self.api = api_base + self._mock_request = MagicMock() + self._mock_headers = {"Authorization": "Bearer test-token"} + + def _headers(self): + return self._mock_headers.copy() + + def _request(self, method, url, **kwargs): + return self._mock_request(method, url, **kwargs) + + def _escape_odata_quotes(self, value: str) -> str: + """Escape single quotes for OData filter values.""" + return value.replace("'", "''") + + +class TestCreateOneToManyRelationship(unittest.TestCase): + """Tests for _create_one_to_many_relationship method.""" + + def setUp(self): + """Set up test fixtures.""" + self.client = MockODataClient("https://example.crm.dynamics.com/api/data/v9.2") + + # Create test metadata objects + self.lookup = LookupAttributeMetadata( + schema_name="new_AccountId", + display_name=Label(localized_labels=[LocalizedLabel(label="Account", language_code=1033)]), + ) + self.relationship = OneToManyRelationshipMetadata( + schema_name="new_account_orders", + referenced_entity="account", + referencing_entity="new_order", + referenced_attribute="accountid", + ) + + def test_create_relationship_url(self): + """Test that correct URL is used.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)" + } + self.client._mock_request.return_value = mock_response + + self.client._create_one_to_many_relationship(self.lookup, self.relationship) + + # Verify URL + call_args = self.client._mock_request.call_args + self.assertEqual(call_args[0][0], "post") + self.assertEqual(call_args[0][1], "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions") + + def test_create_relationship_payload_includes_lookup(self): + """Test that payload includes both relationship and lookup metadata.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)" + } + self.client._mock_request.return_value = mock_response + + self.client._create_one_to_many_relationship(self.lookup, self.relationship) + + # Verify payload + call_args = self.client._mock_request.call_args + payload = call_args[1]["json"] + self.assertIn("@odata.type", payload) + self.assertEqual(payload["@odata.type"], "Microsoft.Dynamics.CRM.OneToManyRelationshipMetadata") + self.assertIn("Lookup", payload) + self.assertEqual(payload["Lookup"]["SchemaName"], "new_AccountId") + + def test_create_relationship_with_solution(self): + """Test that solution header is added when specified.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)" + } + self.client._mock_request.return_value = mock_response + + self.client._create_one_to_many_relationship(self.lookup, self.relationship, solution="MySolution") + + # Verify solution header + call_args = self.client._mock_request.call_args + headers = call_args[1]["headers"] + self.assertEqual(headers["MSCRM.SolutionUniqueName"], "MySolution") + + def test_create_relationship_returns_result(self): + """Test that result dictionary is correctly populated.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)" + } + self.client._mock_request.return_value = mock_response + + result = self.client._create_one_to_many_relationship(self.lookup, self.relationship) + + self.assertEqual(result["relationship_id"], "12345678-1234-1234-1234-123456789abc") + self.assertEqual(result["relationship_schema_name"], "new_account_orders") + self.assertEqual(result["lookup_schema_name"], "new_AccountId") + self.assertEqual(result["referenced_entity"], "account") + self.assertEqual(result["referencing_entity"], "new_order") + + +class TestCreateManyToManyRelationship(unittest.TestCase): + """Tests for _create_many_to_many_relationship method.""" + + def setUp(self): + """Set up test fixtures.""" + self.client = MockODataClient("https://example.crm.dynamics.com/api/data/v9.2") + + self.relationship = ManyToManyRelationshipMetadata( + schema_name="new_account_contact", + entity1_logical_name="account", + entity2_logical_name="contact", + ) + + def test_create_m2m_relationship_url(self): + """Test that correct URL is used.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(abcd1234-abcd-1234-abcd-1234abcd5678)" + } + self.client._mock_request.return_value = mock_response + + self.client._create_many_to_many_relationship(self.relationship) + + call_args = self.client._mock_request.call_args + self.assertEqual(call_args[0][0], "post") + self.assertEqual(call_args[0][1], "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions") + + def test_create_m2m_relationship_returns_result(self): + """Test that result dictionary is correctly populated.""" + mock_response = Mock() + mock_response.headers = { + "OData-EntityId": "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(abcd1234-abcd-1234-abcd-1234abcd5678)" + } + self.client._mock_request.return_value = mock_response + + result = self.client._create_many_to_many_relationship(self.relationship) + + self.assertEqual(result["relationship_id"], "abcd1234-abcd-1234-abcd-1234abcd5678") + self.assertEqual(result["relationship_schema_name"], "new_account_contact") + self.assertEqual(result["entity1_logical_name"], "account") + self.assertEqual(result["entity2_logical_name"], "contact") + + +class TestDeleteRelationship(unittest.TestCase): + """Tests for _delete_relationship method.""" + + def setUp(self): + """Set up test fixtures.""" + self.client = MockODataClient("https://example.crm.dynamics.com/api/data/v9.2") + + def test_delete_relationship_url(self): + """Test that correct URL is constructed.""" + mock_response = Mock() + self.client._mock_request.return_value = mock_response + + self.client._delete_relationship("12345678-1234-1234-1234-123456789abc") + + call_args = self.client._mock_request.call_args + self.assertEqual(call_args[0][0], "delete") + self.assertEqual( + call_args[0][1], + "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions(12345678-1234-1234-1234-123456789abc)", + ) + + def test_delete_relationship_has_if_match_header(self): + """Test that If-Match header is set.""" + mock_response = Mock() + self.client._mock_request.return_value = mock_response + + self.client._delete_relationship("12345678-1234-1234-1234-123456789abc") + + call_args = self.client._mock_request.call_args + headers = call_args[1]["headers"] + self.assertEqual(headers["If-Match"], "*") + + +class TestGetRelationship(unittest.TestCase): + """Tests for _get_relationship method.""" + + def setUp(self): + """Set up test fixtures.""" + self.client = MockODataClient("https://example.crm.dynamics.com/api/data/v9.2") + + def test_get_relationship_url_and_filter(self): + """Test that correct URL and filter are used.""" + mock_response = Mock() + mock_response.json.return_value = {"value": []} + self.client._mock_request.return_value = mock_response + + self.client._get_relationship("new_account_orders") + + call_args = self.client._mock_request.call_args + self.assertEqual(call_args[0][0], "get") + self.assertEqual(call_args[0][1], "https://example.crm.dynamics.com/api/data/v9.2/RelationshipDefinitions") + self.assertEqual(call_args[1]["params"]["$filter"], "SchemaName eq 'new_account_orders'") + + def test_get_relationship_escapes_quotes(self): + """Test that single quotes in schema name are escaped.""" + mock_response = Mock() + mock_response.json.return_value = {"value": []} + self.client._mock_request.return_value = mock_response + + # Schema names shouldn't have quotes, but test the escaping anyway + self.client._get_relationship("schema'name") + + call_args = self.client._mock_request.call_args + self.assertEqual(call_args[1]["params"]["$filter"], "SchemaName eq 'schema''name'") + + def test_get_relationship_returns_first_result(self): + """Test that first result is returned when found.""" + mock_response = Mock() + mock_response.json.return_value = { + "value": [ + {"SchemaName": "new_account_orders", "MetadataId": "12345"}, + {"SchemaName": "other", "MetadataId": "67890"}, + ] + } + self.client._mock_request.return_value = mock_response + + result = self.client._get_relationship("new_account_orders") + + self.assertEqual(result["SchemaName"], "new_account_orders") + self.assertEqual(result["MetadataId"], "12345") + + def test_get_relationship_returns_none_when_not_found(self): + """Test that None is returned when not found.""" + mock_response = Mock() + mock_response.json.return_value = {"value": []} + self.client._mock_request.return_value = mock_response + + result = self.client._get_relationship("nonexistent") + + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/models/__init__.py b/tests/unit/models/__init__.py new file mode 100644 index 0000000..4d4828e --- /dev/null +++ b/tests/unit/models/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Unit tests for metadata models.""" diff --git a/tests/unit/models/test_metadata.py b/tests/unit/models/test_metadata.py new file mode 100644 index 0000000..691b02e --- /dev/null +++ b/tests/unit/models/test_metadata.py @@ -0,0 +1,307 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for metadata entity types.""" + +from PowerPlatform.Dataverse.models.metadata import ( + LocalizedLabel, + Label, + CascadeConfiguration, + LookupAttributeMetadata, + OneToManyRelationshipMetadata, + ManyToManyRelationshipMetadata, +) + + +class TestLocalizedLabel: + """Tests for LocalizedLabel.""" + + def test_to_dict_basic(self): + """Test basic serialization.""" + label = LocalizedLabel(label="Test", language_code=1033) + result = label.to_dict() + + assert result["@odata.type"] == "Microsoft.Dynamics.CRM.LocalizedLabel" + assert result["Label"] == "Test" + assert result["LanguageCode"] == 1033 + + def test_to_dict_with_additional_properties(self): + """Test that additional_properties are merged.""" + label = LocalizedLabel( + label="Test", + language_code=1033, + additional_properties={"IsManaged": True, "MetadataId": "abc-123"}, + ) + result = label.to_dict() + + assert result["Label"] == "Test" + assert result["IsManaged"] is True + assert result["MetadataId"] == "abc-123" + + def test_additional_properties_can_override(self): + """Test that additional_properties can override default values.""" + label = LocalizedLabel( + label="Original", + language_code=1033, + additional_properties={"Label": "Overridden"}, + ) + result = label.to_dict() + + assert result["Label"] == "Overridden" + + +class TestLabel: + """Tests for Label.""" + + def test_to_dict_basic(self): + """Test basic serialization with auto UserLocalizedLabel.""" + label = Label(localized_labels=[LocalizedLabel(label="Test", language_code=1033)]) + result = label.to_dict() + + assert result["@odata.type"] == "Microsoft.Dynamics.CRM.Label" + assert len(result["LocalizedLabels"]) == 1 + assert result["LocalizedLabels"][0]["Label"] == "Test" + # UserLocalizedLabel should default to first localized label + assert result["UserLocalizedLabel"]["Label"] == "Test" + + def test_to_dict_with_explicit_user_label(self): + """Test that explicit user_localized_label is used.""" + label = Label( + localized_labels=[ + LocalizedLabel(label="English", language_code=1033), + LocalizedLabel(label="French", language_code=1036), + ], + user_localized_label=LocalizedLabel(label="French", language_code=1036), + ) + result = label.to_dict() + + assert result["UserLocalizedLabel"]["Label"] == "French" + assert result["UserLocalizedLabel"]["LanguageCode"] == 1036 + + def test_to_dict_with_additional_properties(self): + """Test that additional_properties are merged.""" + label = Label( + localized_labels=[LocalizedLabel(label="Test", language_code=1033)], + additional_properties={"CustomProperty": "value"}, + ) + result = label.to_dict() + + assert result["CustomProperty"] == "value" + + +class TestCascadeConfiguration: + """Tests for CascadeConfiguration.""" + + def test_to_dict_defaults(self): + """Test default values.""" + cascade = CascadeConfiguration() + result = cascade.to_dict() + + assert result["Assign"] == "NoCascade" + assert result["Delete"] == "RemoveLink" + assert result["Merge"] == "NoCascade" + assert result["Reparent"] == "NoCascade" + assert result["Share"] == "NoCascade" + assert result["Unshare"] == "NoCascade" + + def test_to_dict_custom_values(self): + """Test custom cascade values.""" + cascade = CascadeConfiguration( + assign="Cascade", + delete="Restrict", + ) + result = cascade.to_dict() + + assert result["Assign"] == "Cascade" + assert result["Delete"] == "Restrict" + + def test_to_dict_with_additional_properties(self): + """Test additional properties like Archive and RollupView.""" + cascade = CascadeConfiguration( + additional_properties={ + "Archive": "NoCascade", + "RollupView": "NoCascade", + } + ) + result = cascade.to_dict() + + assert result["Archive"] == "NoCascade" + assert result["RollupView"] == "NoCascade" + + +class TestLookupAttributeMetadata: + """Tests for LookupAttributeMetadata.""" + + def test_to_dict_basic(self): + """Test basic serialization.""" + lookup = LookupAttributeMetadata( + schema_name="new_AccountId", + display_name=Label(localized_labels=[LocalizedLabel(label="Account", language_code=1033)]), + ) + result = lookup.to_dict() + + assert result["@odata.type"] == "Microsoft.Dynamics.CRM.LookupAttributeMetadata" + assert result["SchemaName"] == "new_AccountId" + assert result["AttributeType"] == "Lookup" + assert result["AttributeTypeName"]["Value"] == "LookupType" + assert result["RequiredLevel"]["Value"] == "None" + + def test_to_dict_required(self): + """Test required level.""" + lookup = LookupAttributeMetadata( + schema_name="new_AccountId", + display_name=Label(localized_labels=[LocalizedLabel(label="Account", language_code=1033)]), + required_level="ApplicationRequired", + ) + result = lookup.to_dict() + + assert result["RequiredLevel"]["Value"] == "ApplicationRequired" + + def test_to_dict_with_description(self): + """Test with description.""" + lookup = LookupAttributeMetadata( + schema_name="new_AccountId", + display_name=Label(localized_labels=[LocalizedLabel(label="Account", language_code=1033)]), + description=Label(localized_labels=[LocalizedLabel(label="The related account", language_code=1033)]), + ) + result = lookup.to_dict() + + assert "Description" in result + assert result["Description"]["LocalizedLabels"][0]["Label"] == "The related account" + + def test_to_dict_with_additional_properties(self): + """Test additional properties like Targets and IsSecured.""" + lookup = LookupAttributeMetadata( + schema_name="new_ParentId", + display_name=Label(localized_labels=[LocalizedLabel(label="Parent", language_code=1033)]), + additional_properties={ + "Targets": ["account", "contact"], + "IsSecured": True, + "IsValidForAdvancedFind": True, + }, + ) + result = lookup.to_dict() + + assert result["Targets"] == ["account", "contact"] + assert result["IsSecured"] is True + assert result["IsValidForAdvancedFind"] is True + + +class TestOneToManyRelationshipMetadata: + """Tests for OneToManyRelationshipMetadata.""" + + def test_to_dict_basic(self): + """Test basic serialization.""" + rel = OneToManyRelationshipMetadata( + schema_name="new_account_orders", + referenced_entity="account", + referencing_entity="new_order", + referenced_attribute="accountid", + ) + result = rel.to_dict() + + assert result["@odata.type"] == "Microsoft.Dynamics.CRM.OneToManyRelationshipMetadata" + assert result["SchemaName"] == "new_account_orders" + assert result["ReferencedEntity"] == "account" + assert result["ReferencingEntity"] == "new_order" + assert result["ReferencedAttribute"] == "accountid" + assert "CascadeConfiguration" in result + + def test_to_dict_with_custom_cascade(self): + """Test with custom cascade configuration.""" + rel = OneToManyRelationshipMetadata( + schema_name="new_account_orders", + referenced_entity="account", + referencing_entity="new_order", + referenced_attribute="accountid", + cascade_configuration=CascadeConfiguration( + delete="Cascade", + assign="Cascade", + ), + ) + result = rel.to_dict() + + assert result["CascadeConfiguration"]["Delete"] == "Cascade" + assert result["CascadeConfiguration"]["Assign"] == "Cascade" + + def test_to_dict_with_referencing_attribute(self): + """Test with explicit referencing attribute.""" + rel = OneToManyRelationshipMetadata( + schema_name="new_account_orders", + referenced_entity="account", + referencing_entity="new_order", + referenced_attribute="accountid", + referencing_attribute="new_accountid", + ) + result = rel.to_dict() + + assert result["ReferencingAttribute"] == "new_accountid" + + def test_to_dict_with_additional_properties(self): + """Test additional properties like IsCustomizable.""" + rel = OneToManyRelationshipMetadata( + schema_name="new_account_orders", + referenced_entity="account", + referencing_entity="new_order", + referenced_attribute="accountid", + additional_properties={ + "IsCustomizable": {"Value": True, "CanBeChanged": True}, + "IsValidForAdvancedFind": True, + "SecurityTypes": "None", + }, + ) + result = rel.to_dict() + + assert result["IsCustomizable"]["Value"] is True + assert result["IsValidForAdvancedFind"] is True + assert result["SecurityTypes"] == "None" + + +class TestManyToManyRelationshipMetadata: + """Tests for ManyToManyRelationshipMetadata.""" + + def test_to_dict_basic(self): + """Test basic serialization with auto intersect name.""" + rel = ManyToManyRelationshipMetadata( + schema_name="new_account_contact", + entity1_logical_name="account", + entity2_logical_name="contact", + ) + result = rel.to_dict() + + assert result["@odata.type"] == "Microsoft.Dynamics.CRM.ManyToManyRelationshipMetadata" + assert result["SchemaName"] == "new_account_contact" + assert result["Entity1LogicalName"] == "account" + assert result["Entity2LogicalName"] == "contact" + # IntersectEntityName should default to schema_name + assert result["IntersectEntityName"] == "new_account_contact" + + def test_to_dict_with_explicit_intersect_name(self): + """Test with explicit intersect entity name.""" + rel = ManyToManyRelationshipMetadata( + schema_name="new_account_contact", + entity1_logical_name="account", + entity2_logical_name="contact", + intersect_entity_name="new_account_contact_assoc", + ) + result = rel.to_dict() + + assert result["IntersectEntityName"] == "new_account_contact_assoc" + + def test_to_dict_with_additional_properties(self): + """Test additional properties like navigation property names.""" + rel = ManyToManyRelationshipMetadata( + schema_name="new_account_contact", + entity1_logical_name="account", + entity2_logical_name="contact", + additional_properties={ + "Entity1NavigationPropertyName": "new_contacts", + "Entity2NavigationPropertyName": "new_accounts", + "IsCustomizable": {"Value": True, "CanBeChanged": True}, + }, + ) + result = rel.to_dict() + + assert result["Entity1NavigationPropertyName"] == "new_contacts" + assert result["Entity2NavigationPropertyName"] == "new_accounts" + assert result["IsCustomizable"]["Value"] is True diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index e765ba0..216df50 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -127,3 +127,219 @@ def test_get_multiple(self): page_size=None, ) self.assertEqual(results, [expected_batch]) + + +class TestCreateLookupField(unittest.TestCase): + """Tests for DataverseClient.create_lookup_field convenience method.""" + + def setUp(self): + """Set up test fixtures.""" + self.mock_credential = MagicMock(spec=TokenCredential) + self.base_url = "https://example.crm.dynamics.com" + self.client = DataverseClient(self.base_url, self.mock_credential) + + # Mock create_one_to_many_relationship since create_lookup_field calls it + self.client.create_one_to_many_relationship = MagicMock( + return_value={ + "relationship_id": "12345678-1234-1234-1234-123456789abc", + "relationship_schema_name": "account_new_order_new_AccountId", + "lookup_schema_name": "new_AccountId", + "referenced_entity": "account", + "referencing_entity": "new_order", + } + ) + + def test_basic_lookup_field_creation(self): + """Test basic lookup field creation with minimal parameters.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + ) + + # Verify create_one_to_many_relationship was called + self.client.create_one_to_many_relationship.assert_called_once() + + # Get the call arguments + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + relationship = call_args[0][1] + solution = call_args.kwargs.get("solution") + + # Verify lookup metadata + self.assertEqual(lookup.schema_name, "new_AccountId") + self.assertEqual(lookup.required_level, "None") + + # Verify relationship metadata + self.assertEqual(relationship.referenced_entity, "account") + self.assertEqual(relationship.referencing_entity, "new_order") + self.assertEqual(relationship.referenced_attribute, "accountid") + + # Verify no solution (keyword-only, defaults to None) + self.assertIsNone(solution) + + def test_lookup_with_display_name(self): + """Test that display_name is correctly set.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + display_name="Parent Account", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + # Verify display name is in the label + label_dict = lookup.display_name.to_dict() + self.assertEqual(label_dict["LocalizedLabels"][0]["Label"], "Parent Account") + + def test_lookup_with_default_display_name(self): + """Test that display_name defaults to referenced table name.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + # Verify display name defaults to referenced table + label_dict = lookup.display_name.to_dict() + self.assertEqual(label_dict["LocalizedLabels"][0]["Label"], "account") + + def test_lookup_with_description(self): + """Test that description is correctly set.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + description="The customer account for this order", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + # Verify description is set + self.assertIsNotNone(lookup.description) + desc_dict = lookup.description.to_dict() + self.assertEqual(desc_dict["LocalizedLabels"][0]["Label"], "The customer account for this order") + + def test_lookup_required_true(self): + """Test that required=True sets ApplicationRequired level.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + required=True, + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + self.assertEqual(lookup.required_level, "ApplicationRequired") + + def test_lookup_required_false(self): + """Test that required=False sets None level.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + required=False, + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + self.assertEqual(lookup.required_level, "None") + + def test_cascade_delete_configuration(self): + """Test that cascade_delete is correctly passed to relationship.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + cascade_delete="Cascade", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + relationship = call_args[0][1] + + cascade_dict = relationship.cascade_configuration.to_dict() + self.assertEqual(cascade_dict["Delete"], "Cascade") + + def test_solution_passed(self): + """Test that solution is passed through.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + solution="MySolution", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + solution = call_args.kwargs.get("solution") + + self.assertEqual(solution, "MySolution") + + def test_custom_language_code(self): + """Test that custom language_code is used for labels.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + display_name="Compte", + language_code=1036, # French + ) + + call_args = self.client.create_one_to_many_relationship.call_args + lookup = call_args[0][0] + + label_dict = lookup.display_name.to_dict() + self.assertEqual(label_dict["LocalizedLabels"][0]["LanguageCode"], 1036) + self.assertEqual(label_dict["LocalizedLabels"][0]["Label"], "Compte") + + def test_generated_relationship_name(self): + """Test that relationship name is auto-generated correctly.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + relationship = call_args[0][1] + + # Should be: {referenced}_{referencing}_{lookup_field} + self.assertEqual(relationship.schema_name, "account_new_order_new_AccountId") + + def test_referenced_attribute_auto_generated(self): + """Test that referenced_attribute defaults to {table}id.""" + self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + ) + + call_args = self.client.create_one_to_many_relationship.call_args + relationship = call_args[0][1] + + self.assertEqual(relationship.referenced_attribute, "accountid") + + def test_returns_result(self): + """Test that the method returns the result from create_one_to_many_relationship.""" + expected_result = { + "relationship_id": "test-guid", + "relationship_schema_name": "test_schema", + "lookup_schema_name": "test_lookup", + } + self.client.create_one_to_many_relationship.return_value = expected_result + + result = self.client.create_lookup_field( + referencing_table="new_order", + lookup_field_name="new_AccountId", + referenced_table="account", + ) + + self.assertEqual(result, expected_result)