import uuid from unittest import TestCase from metadata.generated.schema.api.data.createTable import CreateTableRequest from metadata.generated.schema.entity.data.table import ( Column, ColumnName, DataType, Table, TableConstraint, ) from metadata.generated.schema.type.basic import ( EntityExtension, EntityName, FullyQualifiedEntityName, Markdown, ) from metadata.generated.schema.type.entityReference import EntityReference class CustomPydanticValidationTest(TestCase): create_request = CreateTableRequest( name=EntityName("Sales::>Territory"), displayName="SalesTerritory", description=Markdown(root="Sales territory lookup table."), tableType="Regular", columns=[ Column( name=ColumnName(root="Sales::Last>Year"), displayName="SalesLastYear", dataType="NUMBER", arrayDataType=None, dataLength=None, precision=None, scale=None, dataTypeDisplay="NUMBER", description=Markdown(root="Sales total of previous year."), fullyQualifiedName=None, tags=None, constraint="NOT_NULL", ordinalPosition=7, jsonSchema=None, children=None, profile=None, customMetrics=None, ), Column( name=ColumnName(root="Bonus"), displayName="Bonus", dataType="NUMBER", arrayDataType=None, dataLength=None, precision=None, scale=None, dataTypeDisplay="NUMBER", description=Markdown(root="Bonus due if quota is met."), fullyQualifiedName=None, tags=None, constraint="NOT_NULL", ordinalPosition=4, jsonSchema=None, children=None, profile=None, customMetrics=None, ), Column( name=ColumnName(root="ModifiedDate"), displayName="ModifiedDate", dataType="DATETIME", arrayDataType=None, dataLength=None, precision=None, scale=None, dataTypeDisplay="DATETIME", description=Markdown(root="Date and time the record was last updated."), fullyQualifiedName=None, tags=None, constraint="NOT_NULL", ordinalPosition=9, jsonSchema=None, children=None, profile=None, customMetrics=None, ), ], dataModel=None, tableConstraints=[ TableConstraint(constraintType="PRIMARY_KEY", columns=["Sales::Last>Year"]) ], tablePartition=None, tableProfilerConfig=None, owners=None, databaseSchema=FullyQualifiedEntityName( root='New Gyro 360.New Gyro 360."AdventureWorks2017.HumanResources"' ), tags=None, schemaDefinition=None, retentionPeriod=None, extension=EntityExtension( root={ "DataQuality": '

Last evaluation: 07/24/2023
Interval: 30 days
Next run: 08/23/2023, 10:44:20
Measurement unit: percent [%]


MetricTargetLatest result

Completeness

90%
100%

Integrity

90%
100%

Timeliness

90%
25%

Uniqueness

90%
60%

Validity

90%
100%

Overall score of the table is: 77%


' } ), sourceUrl=None, domain=None, dataProducts=None, fileFormat=None, lifeCycle=None, sourceHash=None, ) def test_replace_separator(self): assert ( self.create_request.name.root == "Sales__reserved__colon____reserved__arrow__Territory" ) assert ( self.create_request.columns[0].name.root == "Sales__reserved__colon__Last__reserved__arrow__Year" ) assert ( self.create_request.tableConstraints[0].columns[0] == "Sales__reserved__colon__Last__reserved__arrow__Year" ) def test_revert_separator(self): fetch_response_revert_separator = Table( id=uuid.uuid4(), name="test__reserved__colon__table", databaseSchema=EntityReference(id=uuid.uuid4(), type="databaseSchema"), fullyQualifiedName="test-service-table.test-db.test-schema.test", columns=[Column(name="id", dataType=DataType.BIGINT)], ) fetch_response_revert_separator_2 = Table( id=uuid.uuid4(), name="test__reserved__colon__table__reserved__arrow__", databaseSchema=EntityReference(id=uuid.uuid4(), type="databaseSchema"), fullyQualifiedName="test-service-table.test-db.test-schema.test", columns=[Column(name="id", dataType=DataType.BIGINT)], ) assert fetch_response_revert_separator.name.root == "test::table" assert fetch_response_revert_separator_2.name.root == "test::table>"