diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b045bcc5..b9fb4bfba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `hana` server type for `datacontract test`, supporting SAP HANA Cloud and SAP Datasphere (Open SQL schemas). Install with `pip install datacontract-cli[hana]`. (`hdbcli` is under SAP's proprietary license and is therefore not part of `[all]`.) + ## [1.0.4] - 2026-06-22 ### Added diff --git a/README.md b/README.md index 0f5e112f3..f91b71984 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# Data Contract CLI +# Data Contract CLI

@@ -267,6 +267,7 @@ A list of available extras: | RDF | `pip install datacontract-cli[rdf]` | | Amazon Redshift | `pip install datacontract-cli[redshift]` | | S3 Integration | `pip install datacontract-cli[s3]` | +| SAP HANA / Datasphere | `pip install datacontract-cli[hana]` | | Snowflake Integration | `pip install datacontract-cli[snowflake]` | | Microsoft SQL Server | `pip install datacontract-cli[sqlserver]` | | Trino | `pip install datacontract-cli[trino]` | diff --git a/datacontract/engines/data_contract_test.py b/datacontract/engines/data_contract_test.py index fa3fbb610..86d7e70c9 100644 --- a/datacontract/engines/data_contract_test.py +++ b/datacontract/engines/data_contract_test.py @@ -64,6 +64,12 @@ def execute_data_contract_test( if server.type == "api": server = process_api_response(run, server) + if server.type == "hana": + from datacontract.engines.hana.check_hana_execute import check_hana_execute + + check_hana_execute(run, data_contract, server, schema_name=schema_name, check_categories=check_categories) + return + specs = create_checks(data_contract, server, schema_name=schema_name) if check_categories is not None: specs = [s for s in specs if s.category in check_categories] diff --git a/datacontract/engines/hana/__init__.py b/datacontract/engines/hana/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/datacontract/engines/hana/__init__.py @@ -0,0 +1 @@ + diff --git a/datacontract/engines/hana/check_hana_execute.py b/datacontract/engines/hana/check_hana_execute.py new file mode 100644 index 000000000..b95780bbe --- /dev/null +++ b/datacontract/engines/hana/check_hana_execute.py @@ -0,0 +1,59 @@ +from open_data_contract_standard.model import OpenDataContractStandard, Server + +from datacontract.engines.hana.hana_connection import get_connection +from datacontract.engines.hana.hana_quality_check import run_quality_checks, run_sla_checks +from datacontract.engines.hana.hana_schema_check import run_schema_checks +from datacontract.model.exceptions import DataContractException +from datacontract.model.run import ResultEnum, Run + + +def check_hana_execute( + run: Run, + data_contract: OpenDataContractStandard, + server: Server, + schema_name: str = "all", + check_categories: set[str] | None = None, +): + connection = None + try: + run.log_info("Running engine hana") + server_schema = server.schema_ + if not server_schema: + raise DataContractException( + type="hana-connection", + name="Missing SAP HANA schema", + result=ResultEnum.failed, + reason="Server schema is required for SAP HANA Cloud.", + engine="hana", + ) + connection = get_connection(server) + checks_before = len(run.checks) + + for schema_object in data_contract.schema_ or []: + if schema_name != "all" and schema_object.name != schema_name: + continue + if check_categories is None or "schema" in check_categories: + run.checks.extend(run_schema_checks(connection, server_schema, schema_object)) + if check_categories is None or "quality" in check_categories: + run.checks.extend(run_quality_checks(connection, server_schema, schema_object)) + + if check_categories is None or "servicelevel" in check_categories: + run.checks.extend(run_sla_checks(connection, server_schema, data_contract, schema_filter=schema_name)) + + if check_categories is not None and len(run.checks) == checks_before: + run.log_warn(f"No checks found for categories: {', '.join(sorted(check_categories))}") + + except DataContractException: + raise + except Exception as e: + raise DataContractException( + type="hana", + name="SAP HANA Cloud test execution error", + result=ResultEnum.error, + reason=str(e), + engine="hana", + original_exception=e, + ) + finally: + if connection is not None: + connection.close() diff --git a/datacontract/engines/hana/hana_connection.py b/datacontract/engines/hana/hana_connection.py new file mode 100644 index 000000000..d15dcbcda --- /dev/null +++ b/datacontract/engines/hana/hana_connection.py @@ -0,0 +1,66 @@ +import os +from contextlib import contextmanager +from typing import Any + +from open_data_contract_standard.model import Server + +from datacontract.model.exceptions import DataContractException, require_env +from datacontract.model.run import ResultEnum + + +def _env_bool(name: str, default: str = "true") -> bool: + return os.getenv(name, default).lower() in ("true", "1", "yes") + + +def import_hdbcli(): + try: + from hdbcli import dbapi + + return dbapi + except ImportError as e: + raise DataContractException( + type="hana-connection", + result=ResultEnum.failed, + name="hdbcli is missing", + reason="Install the extra datacontract-cli[hana] to use SAP HANA Cloud.", + engine="hana", + original_exception=e, + ) + + +def get_connection(server: Server) -> Any: + dbapi = import_hdbcli() + username = require_env("DATACONTRACT_HANA_USERNAME", server_type="hana") + password = require_env("DATACONTRACT_HANA_PASSWORD", server_type="hana") + encrypt = _env_bool("DATACONTRACT_HANA_ENCRYPT") + ssl_validate = _env_bool("DATACONTRACT_HANA_SSL_VALIDATE_CERTIFICATE") + ssl_hostname = os.getenv("DATACONTRACT_HANA_SSL_HOSTNAME_IN_CERTIFICATE", "*") + + try: + return dbapi.connect( + address=server.host, + port=server.port or 443, + user=username, + password=password, + encrypt=encrypt, + sslValidateCertificate=ssl_validate, + sslHostnameInCertificate=ssl_hostname, + ) + except Exception as e: + raise DataContractException( + type="hana-connection", + result=ResultEnum.failed, + name="SAP HANA Cloud connection error", + reason=str(e), + engine="hana", + original_exception=e, + ) + + +@contextmanager +def hana_connection(server: Server): + connection = get_connection(server) + try: + yield connection + finally: + connection.close() diff --git a/datacontract/engines/hana/hana_quality_check.py b/datacontract/engines/hana/hana_quality_check.py new file mode 100644 index 000000000..faa5e12c3 --- /dev/null +++ b/datacontract/engines/hana/hana_quality_check.py @@ -0,0 +1,476 @@ +import re +import uuid +from typing import Any + +from open_data_contract_standard.model import DataQuality, OpenDataContractStandard, SchemaObject + +from datacontract.engines.checks.create_checks import _retention_value_to_seconds +from datacontract.export.sodacl_check_builder import to_sodacl_threshold +from datacontract.engines.hana.hana_schema_check import qualified_table_name, quote_identifier +from datacontract.model.run import Check, ResultEnum + + +def run_quality_checks(connection, schema_name: str, schema_object: SchemaObject) -> list[Check]: + table_name = schema_object.physicalName or schema_object.name + checks: list[Check] = [] + + for prop in schema_object.properties or []: + field_name = prop.physicalName or prop.name + for index, quality in enumerate(prop.quality or []): + check = _quality_check(connection, schema_name, table_name, field_name, quality, index) + if check is not None: + checks.append(check) + + for index, quality in enumerate(schema_object.quality or []): + check = _quality_check(connection, schema_name, table_name, None, quality, index) + if check is not None: + checks.append(check) + + return checks + + +def run_sla_checks( + connection, schema_name: str, data_contract: OpenDataContractStandard, schema_filter: str = "all" +) -> list[Check]: + checks: list[Check] = [] + for sla in data_contract.slaProperties or []: + if sla.property not in ("freshness", "retention"): + continue + resolved = _resolve_sla_element(data_contract, sla.element) + if resolved is None: + continue + schema_object, field_name = resolved + if schema_filter != "all" and schema_object.name != schema_filter: + continue + table_name = schema_object.physicalName or schema_object.name + if sla.property == "freshness": + check = _freshness_check(connection, schema_name, table_name, field_name, sla) + else: + check = _retention_check(connection, schema_name, table_name, field_name, sla) + if check is not None: + checks.append(check) + return checks + + +def prepare_hana_query(query: str, schema_name: str, model_name: str, field_name: str | None = None) -> str | None: + if query is None or query == "": + return None + + qualified_model_name = qualified_table_name(schema_name, model_name) + quoted_schema_name = quote_identifier(schema_name) + query = re.sub(r'["\']?\$?\{model}["\']?', qualified_model_name, query) + query = re.sub(r'["\']?\$?\{table}["\']?', qualified_model_name, query) + query = re.sub(r'["\']?\$?\{object}["\']?', qualified_model_name, query) + query = re.sub(r'["\']?\$?\{schema}["\']?', quoted_schema_name, query) + + if field_name is not None: + quoted_field_name = quote_identifier(field_name) + query = re.sub(r'["\']?\$?\{field}["\']?', quoted_field_name, query) + query = re.sub(r'["\']?\$?\{column}["\']?', quoted_field_name, query) + query = re.sub(r'["\']?\$?\{property}["\']?', quoted_field_name, query) + + return query + + +def _quality_check( + connection, schema_name: str, table_name: str, field_name: str | None, quality: DataQuality, index: int +) -> Check | None: + if quality.type == "custom" and quality.engine == "soda": + return _check( + check_type="quality_custom_soda", + key=_quality_key(table_name, field_name, f"quality_custom_soda_{index}"), + name=quality.description or "Custom SodaCL Check", + model=table_name, + field=field_name, + implementation=quality.implementation, + result=ResultEnum.warning, + reason="SodaCL quality checks are not supported for HANA. Use type: sql instead.", + ) + + if quality.type == "sql": + return _sql_quality_check(connection, schema_name, table_name, field_name, quality, index) + + if quality.metric == "rowCount": + sql = f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)}" + return _metric_quality_check( + connection, + quality, + check_type="row_count", + key=f"{table_name}__row_count", + name="Check row count", + model=table_name, + field=None, + sql=sql, + ) + + if quality.metric == "duplicateValues": + if field_name is not None: + sql = ( + f"SELECT COUNT(*) - COUNT(DISTINCT {quote_identifier(field_name)}) " + f"FROM {qualified_table_name(schema_name, table_name)}" + ) + return _metric_quality_check( + connection, + quality, + check_type="field_duplicate_values", + key=f"{table_name}__{field_name}__field_duplicate_values", + name=f"Check duplicate values for field {field_name}", + model=table_name, + field=field_name, + sql=sql, + ) + + properties = _arguments(quality).get("properties") + if not properties: + return _warning_check( + check_type="model_duplicate_values", + key=f"{table_name}__model_duplicate_values", + name="Check duplicate values for model", + model=table_name, + field=None, + reason="duplicateValues requires arguments.properties at model level.", + ) + quoted_properties = ", ".join(quote_identifier(prop) for prop in properties) + sql = ( + f"SELECT COUNT(*) - COUNT(DISTINCT {quoted_properties}) " + f"FROM {qualified_table_name(schema_name, table_name)}" + ) + return _metric_quality_check( + connection, + quality, + check_type="model_duplicate_values", + key=f"{table_name}__model_duplicate_values", + name="Check duplicate values for model", + model=table_name, + field=None, + sql=sql, + ) + + if quality.metric == "nullValues" and field_name is not None: + sql = ( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} IS NULL" + ) + return _metric_quality_check( + connection, + quality, + check_type="field_null_values", + key=f"{table_name}__{field_name}__field_null_values", + name=f"Check null values for field {field_name}", + model=table_name, + field=field_name, + sql=sql, + ) + + if quality.metric == "invalidValues" and field_name is not None: + valid_values = _arguments(quality).get("validValues") + if not valid_values: + return _warning_check( + check_type="field_invalid_values", + key=f"{table_name}__{field_name}__field_invalid_values", + name=f"Check invalid values for field {field_name}", + model=table_name, + field=field_name, + reason="invalidValues requires arguments.validValues at field level.", + ) + placeholders = ", ".join("?" for _ in valid_values) + sql = ( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} NOT IN ({placeholders})" + ) + return _metric_quality_check( + connection, + quality, + check_type="field_invalid_values", + key=f"{table_name}__{field_name}__field_invalid_values", + name=f"Check invalid values for field {field_name}", + model=table_name, + field=field_name, + sql=sql, + params=valid_values, + ) + + if quality.metric == "missingValues" and field_name is not None: + missing_values = [value for value in (_arguments(quality).get("missingValues") or []) if value is not None] + if missing_values: + placeholders = ", ".join("?" for _ in missing_values) + missing_predicate = f" OR {quote_identifier(field_name)} IN ({placeholders})" + else: + missing_predicate = "" + sql = ( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} IS NULL{missing_predicate}" + ) + return _metric_quality_check( + connection, + quality, + check_type="field_missing_values", + key=f"{table_name}__{field_name}__field_missing_values", + name=f"Check missing values for field {field_name}", + model=table_name, + field=field_name, + sql=sql, + params=missing_values, + ) + + return None + + +def _sql_quality_check( + connection, schema_name: str, table_name: str, field_name: str | None, quality: DataQuality, index: int +) -> Check | None: + query = prepare_hana_query(quality.query, schema_name, table_name, field_name) + if query is None: + return None + check_type = "field_quality_sql" if field_name is not None else "model_quality_sql" + return _metric_quality_check( + connection, + quality, + check_type=check_type, + key=_quality_key(table_name, field_name, f"quality_sql_{index}"), + name=quality.description or "Quality Check", + model=table_name, + field=field_name, + sql=query, + ) + + +def _metric_quality_check( + connection, + quality: DataQuality, + *, + check_type: str, + key: str, + name: str, + model: str, + field: str | None, + sql: str, + params: list[Any] | None = None, +) -> Check | None: + threshold = to_sodacl_threshold(quality) + if threshold is None: + return None + + try: + value = _fetch_scalar(connection, sql, params or []) + except Exception as e: + return _check( + check_type=check_type, + key=key, + name=name, + model=model, + field=field, + implementation=sql, + result=ResultEnum.error, + reason=str(e), + ) + + passed = _evaluate_threshold(value, quality) + return _check( + check_type=check_type, + key=key, + name=name, + model=model, + field=field, + implementation=sql, + result=ResultEnum.passed if passed else ResultEnum.failed, + reason=None if passed else f"Actual value {value} does not satisfy threshold {threshold}.", + diagnostics={"value": value, "threshold": threshold}, + ) + + +def _freshness_check(connection, schema_name: str, table_name: str, field_name: str, sla) -> Check | None: + threshold = _freshness_value_to_seconds(sla.value, sla.unit) + if threshold is None: + return None + sql = ( + f"SELECT SECONDS_BETWEEN(MAX({quote_identifier(field_name)}), CURRENT_TIMESTAMP) " + f"FROM {qualified_table_name(schema_name, table_name)}" + ) + return _servicelevel_check( + connection, + check_type="servicelevel_freshness", + key="servicelevel_freshness", + name=f"Freshness of {table_name}.{field_name} < {threshold}s", + model=table_name, + field=field_name, + sql=sql, + threshold=threshold, + ) + + +def _retention_check(connection, schema_name: str, table_name: str, field_name: str, sla) -> Check | None: + threshold = _retention_value_to_seconds(sla.value, sla.unit) + if threshold is None: + return None + sql = ( + f"SELECT SECONDS_BETWEEN(MIN({quote_identifier(field_name)}), CURRENT_TIMESTAMP) " + f"FROM {qualified_table_name(schema_name, table_name)}" + ) + return _servicelevel_check( + connection, + check_type="servicelevel_retention", + key="servicelevel_retention", + name=f"Retention of {table_name}.{field_name} < {threshold}s", + model=table_name, + field=field_name, + sql=sql, + threshold=threshold, + ) + + +def _servicelevel_check( + connection, + *, + check_type: str, + key: str, + name: str, + model: str, + field: str, + sql: str, + threshold: int, +) -> Check: + try: + value = _fetch_scalar(connection, sql, []) + except Exception as e: + return _check( + check_type=check_type, + key=key, + name=name, + model=model, + field=field, + implementation=sql, + result=ResultEnum.error, + reason=str(e), + category="servicelevel", + ) + passed = value is not None and value < threshold + return _check( + check_type=check_type, + key=key, + name=name, + model=model, + field=field, + implementation=sql, + result=ResultEnum.passed if passed else ResultEnum.failed, + reason=None if passed else f"Actual value {value} seconds is not less than {threshold} seconds.", + diagnostics={"value": value, "threshold": threshold}, + category="servicelevel", + ) + + +def _evaluate_threshold(value: Any, quality: DataQuality) -> bool: + if quality.mustBe is not None: + return value == quality.mustBe + if quality.mustNotBe is not None: + return value != quality.mustNotBe + if quality.mustBeGreaterThan is not None: + return value > quality.mustBeGreaterThan + if quality.mustBeGreaterOrEqualTo is not None: + return value >= quality.mustBeGreaterOrEqualTo + if quality.mustBeLessThan is not None: + return value < quality.mustBeLessThan + if quality.mustBeLessOrEqualTo is not None: + return value <= quality.mustBeLessOrEqualTo + if quality.mustBeBetween is not None: + return quality.mustBeBetween[0] <= value <= quality.mustBeBetween[1] + if quality.mustNotBeBetween is not None: + return value < quality.mustNotBeBetween[0] or value > quality.mustNotBeBetween[1] + return False + + +def _freshness_value_to_seconds(value, unit: str | None) -> int | None: + if value is None: + return None + numeric_value = int(value) + unit_lower = unit.lower() if unit else "d" + if unit_lower in ("d", "day", "days"): + return numeric_value * 24 * 60 * 60 + if unit_lower in ("h", "hr", "hour", "hours"): + return numeric_value * 60 * 60 + if unit_lower in ("m", "min", "minute", "minutes"): + return numeric_value * 60 + if unit_lower in ("s", "sec", "second", "seconds"): + return numeric_value + return None + + +def _resolve_sla_element(data_contract: OpenDataContractStandard, element: str | None): + if element is None or element.count(".") != 1: + return None + model_name, field_name = element.split(".") + for schema_object in data_contract.schema_ or []: + if schema_object.name != model_name: + continue + for prop in schema_object.properties or []: + if prop.name == field_name: + return schema_object, prop.physicalName or prop.name + return None + + +def _quality_key(table_name: str, field_name: str | None, suffix: str) -> str: + if field_name is None: + return f"{table_name}__{suffix}" + return f"{table_name}__{field_name}__{suffix}" + + +def _arguments(quality: DataQuality) -> dict: + return quality.arguments or {} + + +def _warning_check(*, check_type: str, key: str, name: str, model: str, field: str | None, reason: str) -> Check: + return _check( + check_type=check_type, + key=key, + name=name, + model=model, + field=field, + implementation=None, + result=ResultEnum.warning, + reason=reason, + ) + + +def _check( + *, + check_type: str, + key: str, + name: str, + model: str, + field: str | None, + implementation: str | None, + result: ResultEnum, + reason: str | None, + diagnostics: dict | None = None, + category: str = "quality", +) -> Check: + return Check( + id=str(uuid.uuid4()), + key=key, + category=category, + type=check_type, + name=name, + model=model, + field=field, + engine="hana", + language="sql", + implementation=implementation, + result=result, + reason=reason, + diagnostics=diagnostics, + ) + + +def _fetch_scalar(connection, sql: str, params: list[Any] | None = None): + cursor = connection.cursor() + try: + cursor.execute(sql, params or []) + row = cursor.fetchone() + if row is None: + return None + if isinstance(row, dict): + return next(iter(row.values())) + return row[0] + finally: + if hasattr(cursor, "close"): + cursor.close() diff --git a/datacontract/engines/hana/hana_schema_check.py b/datacontract/engines/hana/hana_schema_check.py new file mode 100644 index 000000000..ec6c2f201 --- /dev/null +++ b/datacontract/engines/hana/hana_schema_check.py @@ -0,0 +1,464 @@ +import uuid +from dataclasses import dataclass +from typing import Any + +from open_data_contract_standard.model import SchemaObject, SchemaProperty + +from datacontract.engines.hana.hana_type_mapping import types_match +from datacontract.model.run import Check, ResultEnum + + +@dataclass +class ColumnMetadata: + name: str + data_type_name: str + length: int | None + scale: int | None + is_nullable: str | None + position: int | None + source: str + + +def quote_identifier(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def qualified_table_name(schema: str, table: str) -> str: + return f"{quote_identifier(schema)}.{quote_identifier(table)}" + + +def run_schema_checks(connection, schema_name: str, schema_object: SchemaObject) -> list[Check]: + table_name = _schema_name(schema_object) + checks: list[Check] = [] + column_metadata = _get_column_metadata(connection, schema_name, table_name) + + model_exists_check = _result_check( + check_type="model_exists", + key=f"{table_name}__model_exists", + name=f"Check that model {table_name} exists", + model=table_name, + field=None, + implementation="Catalog lookup in SYS.TABLE_COLUMNS and SYS.VIEW_COLUMNS", + result=ResultEnum.passed if column_metadata else ResultEnum.failed, + reason=None if column_metadata else f"Model {schema_name}.{table_name} does not exist.", + ) + checks.append(model_exists_check) + if not column_metadata: + return checks + + columns_by_name = {column.name: column for column in column_metadata} + source = column_metadata[0].source + + for prop in schema_object.properties or []: + field_name = _property_name(prop) + column = columns_by_name.get(field_name) + field_present_check = _result_check( + check_type="field_is_present", + key=f"{table_name}__{field_name}__field_is_present", + name=f"Check that field {field_name} is present", + model=table_name, + field=field_name, + implementation="Catalog lookup in SYS.TABLE_COLUMNS and SYS.VIEW_COLUMNS", + result=ResultEnum.passed if column else ResultEnum.failed, + reason=None if column else f"Field {field_name} is missing in {schema_name}.{table_name}.", + ) + checks.append(field_present_check) + if column is None: + continue + + expected_type = prop.physicalType or prop.logicalType + if expected_type: + checks.append(_field_type_check(table_name, field_name, expected_type, column)) + + if prop.required: + checks.append(_field_required_check(table_name, field_name, column)) + + if prop.unique: + checks.append( + _zero_violations_check( + connection, + check_type="field_unique", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) - COUNT(DISTINCT {quote_identifier(field_name)}) " + f"FROM {qualified_table_name(schema_name, table_name)}" + ), + params=None, + name=f"Check that unique field {field_name} has no duplicate values", + failure_reason="Found {value} duplicate values.", + ) + ) + + if prop.primaryKey: + checks.append(_primary_key_check(connection, schema_name, table_name, field_name, source)) + + checks.extend(_logical_type_option_checks(connection, schema_name, table_name, field_name, prop)) + + return checks + + +def _schema_name(schema_object: SchemaObject) -> str: + return schema_object.physicalName or schema_object.name + + +def _property_name(prop: SchemaProperty) -> str: + return prop.physicalName or prop.name + + +def _logical_type_option_checks( + connection, schema_name: str, table_name: str, field_name: str, prop: SchemaProperty +) -> list[Check]: + checks: list[Check] = [] + options = prop.logicalTypeOptions or {} + + min_length = options.get("minLength") + if min_length is not None: + checks.append( + _zero_violations_check( + connection, + check_type="field_min_length", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE LENGTH({quote_identifier(field_name)}) < ?" + ), + params=[min_length], + name=f"Check that field {field_name} has a min length of {min_length}", + failure_reason=f"Found {{value}} values shorter than {min_length}.", + ) + ) + + max_length = options.get("maxLength") + if max_length is not None: + checks.append( + _zero_violations_check( + connection, + check_type="field_max_length", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE LENGTH({quote_identifier(field_name)}) > ?" + ), + params=[max_length], + name=f"Check that field {field_name} has a max length of {max_length}", + failure_reason=f"Found {{value}} values longer than {max_length}.", + ) + ) + + minimum = options.get("minimum") + if minimum is not None: + checks.append(_minimum_check(connection, schema_name, table_name, field_name, minimum)) + + maximum = options.get("maximum") + if maximum is not None: + checks.append(_maximum_check(connection, schema_name, table_name, field_name, maximum)) + + exclusive_minimum = options.get("exclusiveMinimum") + if exclusive_minimum is not None: + checks.append(_minimum_check(connection, schema_name, table_name, field_name, exclusive_minimum)) + checks.append(_not_equal_check(connection, schema_name, table_name, field_name, exclusive_minimum)) + + exclusive_maximum = options.get("exclusiveMaximum") + if exclusive_maximum is not None: + checks.append(_maximum_check(connection, schema_name, table_name, field_name, exclusive_maximum)) + checks.append(_not_equal_check(connection, schema_name, table_name, field_name, exclusive_maximum)) + + enum_values = options.get("enum") + if enum_values is not None and len(enum_values) > 0: + placeholders = ", ".join("?" for _ in enum_values) + checks.append( + _zero_violations_check( + connection, + check_type="field_enum", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} NOT IN ({placeholders})" + ), + params=enum_values, + name=f"Check that field {field_name} only contains enum values {enum_values}", + failure_reason="Found {value} values outside the enum.", + ) + ) + + pattern = options.get("pattern") + if pattern is not None: + checks.append( + _zero_violations_check( + connection, + check_type="field_regex", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} NOT LIKE_REGEXPR ?" + ), + params=[pattern], + name=f"Check that field {field_name} matches regex {pattern}", + failure_reason="Found {value} values that do not match the regex.", + ) + ) + + return checks + + +def _minimum_check(connection, schema_name: str, table_name: str, field_name: str, minimum: Any) -> Check: + return _zero_violations_check( + connection, + check_type="field_minimum", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} < ?" + ), + params=[minimum], + name=f"Check that field {field_name} has a minimum of {minimum}", + failure_reason=f"Found {{value}} values below {minimum}.", + ) + + +def _maximum_check(connection, schema_name: str, table_name: str, field_name: str, maximum: Any) -> Check: + return _zero_violations_check( + connection, + check_type="field_maximum", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} > ?" + ), + params=[maximum], + name=f"Check that field {field_name} has a maximum of {maximum}", + failure_reason=f"Found {{value}} values above {maximum}.", + ) + + +def _not_equal_check(connection, schema_name: str, table_name: str, field_name: str, value: Any) -> Check: + return _zero_violations_check( + connection, + check_type="field_not_equal", + table_schema=schema_name, + table_name=table_name, + field_name=field_name, + sql=( + f"SELECT COUNT(*) FROM {qualified_table_name(schema_name, table_name)} " + f"WHERE {quote_identifier(field_name)} = ?" + ), + params=[value], + name=f"Check that field {field_name} is not equal to {value}", + failure_reason=f"Found {{value}} values equal to {value}.", + ) + + +def _field_type_check(table_name: str, field_name: str, expected_type: str, column: ColumnMetadata) -> Check: + actual_type = column.data_type_name + result = ResultEnum.passed if types_match(actual_type, expected_type) else ResultEnum.failed + return _result_check( + check_type="field_type", + key=f"{table_name}__{field_name}__field_type", + name=f"Check that field {field_name} has type {expected_type}", + model=table_name, + field=field_name, + implementation="Catalog DATA_TYPE_NAME comparison", + result=result, + reason=None if result == ResultEnum.passed else f"Expected {expected_type}, got {actual_type}.", + diagnostics={"expected": expected_type, "actual": actual_type}, + ) + + +def _field_required_check(table_name: str, field_name: str, column: ColumnMetadata) -> Check: + nullable = str(column.is_nullable).upper() + result = ResultEnum.passed if nullable == "FALSE" else ResultEnum.failed + return _result_check( + check_type="field_required", + key=f"{table_name}__{field_name}__field_required", + name=f"Check that field {field_name} is not nullable", + model=table_name, + field=field_name, + implementation="Catalog IS_NULLABLE comparison", + result=result, + reason=None if result == ResultEnum.passed else f"Field {field_name} is nullable in the catalog.", + diagnostics={"isNullable": column.is_nullable}, + ) + + +def _primary_key_check(connection, schema_name: str, table_name: str, field_name: str, source: str) -> Check: + sql = """ +SELECT COLUMN_NAME +FROM SYS.CONSTRAINTS +WHERE SCHEMA_NAME = ? AND TABLE_NAME = ? AND IS_PRIMARY_KEY = 'TRUE' +""" + if source == "view": + return _result_check( + check_type="field_primary_key", + key=f"{table_name}__{field_name}__field_primary_key", + name=f"Check that field {field_name} is a primary key", + model=table_name, + field=field_name, + implementation=sql.strip(), + result=ResultEnum.warning, + reason="Primary key metadata not available for views.", + ) + + rows = _fetch_all(connection, sql, [schema_name, table_name]) + primary_key_columns = {_row_value(row, 0, "COLUMN_NAME") for row in rows} + result = ResultEnum.passed if field_name in primary_key_columns else ResultEnum.failed + return _result_check( + check_type="field_primary_key", + key=f"{table_name}__{field_name}__field_primary_key", + name=f"Check that field {field_name} is a primary key", + model=table_name, + field=field_name, + implementation=sql.strip(), + result=result, + reason=None if result == ResultEnum.passed else f"Field {field_name} is not part of the primary key.", + diagnostics={"primaryKeyColumns": sorted(primary_key_columns)}, + ) + + +def _zero_violations_check( + connection, + *, + check_type: str, + table_schema: str, + table_name: str, + field_name: str, + sql: str, + params: list[Any] | None, + name: str, + failure_reason: str, +) -> Check: + try: + value = _fetch_scalar(connection, sql, params) + except Exception as e: + return _result_check( + check_type=check_type, + key=f"{table_name}__{field_name}__{check_type}", + name=name, + model=table_name, + field=field_name, + implementation=sql, + result=ResultEnum.error, + reason=str(e), + ) + + result = ResultEnum.passed if value == 0 else ResultEnum.failed + return _result_check( + check_type=check_type, + key=f"{table_name}__{field_name}__{check_type}", + name=name, + model=table_name, + field=field_name, + implementation=sql, + result=result, + reason=None if result == ResultEnum.passed else failure_reason.format(value=value), + diagnostics={"value": value, "schema": table_schema}, + ) + + +def _get_column_metadata(connection, schema_name: str, table_name: str) -> list[ColumnMetadata]: + table_sql = """ +SELECT COLUMN_NAME, DATA_TYPE_NAME, LENGTH, SCALE, IS_NULLABLE, POSITION +FROM SYS.TABLE_COLUMNS +WHERE SCHEMA_NAME = ? AND TABLE_NAME = ? +ORDER BY POSITION +""" + table_rows = _fetch_all(connection, table_sql, [schema_name, table_name]) + if table_rows: + return [_column_metadata_from_row(row, "table") for row in table_rows] + + view_sql = """ +SELECT COLUMN_NAME, DATA_TYPE_NAME, LENGTH, SCALE, IS_NULLABLE, POSITION +FROM SYS.VIEW_COLUMNS +WHERE SCHEMA_NAME = ? AND VIEW_NAME = ? +ORDER BY POSITION +""" + view_rows = _fetch_all(connection, view_sql, [schema_name, table_name]) + return [_column_metadata_from_row(row, "view") for row in view_rows] + + +def _column_metadata_from_row(row, source: str) -> ColumnMetadata: + return ColumnMetadata( + name=_row_value(row, 0, "COLUMN_NAME"), + data_type_name=_row_value(row, 1, "DATA_TYPE_NAME"), + length=_row_value(row, 2, "LENGTH"), + scale=_row_value(row, 3, "SCALE"), + is_nullable=_row_value(row, 4, "IS_NULLABLE"), + position=_row_value(row, 5, "POSITION"), + source=source, + ) + + +def _fetch_scalar(connection, sql: str, params: list[Any] | None = None): + cursor = connection.cursor() + try: + cursor.execute(sql, params or []) + row = cursor.fetchone() + if row is None: + return None + return _row_value(row, 0, None) + finally: + _close_cursor(cursor) + + +def _fetch_all(connection, sql: str, params: list[Any] | None = None): + cursor = connection.cursor() + try: + cursor.execute(sql, params or []) + return cursor.fetchall() + finally: + _close_cursor(cursor) + + +def _row_value(row, index: int, key: str | None): + if isinstance(row, dict): + if key is None: + return next(iter(row.values())) + return row[key] + return row[index] + + +def _close_cursor(cursor) -> None: + if hasattr(cursor, "close"): + cursor.close() + + +def _result_check( + *, + check_type: str, + key: str, + name: str, + model: str, + field: str | None, + implementation: str, + result: ResultEnum, + reason: str | None, + diagnostics: dict | None = None, +) -> Check: + return Check( + id=str(uuid.uuid4()), + key=key, + category="schema", + type=check_type, + name=name, + model=model, + field=field, + engine="hana", + language="sql", + implementation=implementation, + result=result, + reason=reason, + diagnostics=diagnostics, + ) diff --git a/datacontract/engines/hana/hana_type_mapping.py b/datacontract/engines/hana/hana_type_mapping.py new file mode 100644 index 000000000..d9b5ab465 --- /dev/null +++ b/datacontract/engines/hana/hana_type_mapping.py @@ -0,0 +1,205 @@ +from typing import Any + +from open_data_contract_standard.model import SchemaProperty + +_HANA_LOGICAL_TYPE_MAPPING = { + "string": {"NVARCHAR", "VARCHAR", "NCLOB", "CLOB", "SHORTTEXT"}, + "text": {"NVARCHAR", "VARCHAR", "NCLOB", "CLOB", "SHORTTEXT"}, + "varchar": {"NVARCHAR", "VARCHAR", "NCLOB", "CLOB", "SHORTTEXT"}, + "integer": {"INTEGER", "INT", "BIGINT", "SMALLINT", "TINYINT"}, + "int": {"INTEGER", "INT", "BIGINT", "SMALLINT", "TINYINT"}, + "long": {"BIGINT"}, + "bigint": {"BIGINT"}, + "float": {"DOUBLE", "FLOAT", "REAL"}, + "double": {"DOUBLE", "FLOAT", "REAL"}, + "decimal": {"DECIMAL", "NUMERIC"}, + "numeric": {"DECIMAL", "NUMERIC"}, + "number": {"DECIMAL", "NUMERIC"}, + "boolean": {"BOOLEAN"}, + "date": {"DATE"}, + "time": {"TIME"}, + "timestamp": {"TIMESTAMP", "SECONDDATE"}, + "timestamp_tz": {"TIMESTAMP", "SECONDDATE"}, + "timestamp_ntz": {"SECONDDATE", "TIMESTAMP"}, + "binary": {"VARBINARY", "BLOB"}, + "bytes": {"VARBINARY", "BLOB"}, +} + +_HANA_TYPES = {hana_type for types in _HANA_LOGICAL_TYPE_MAPPING.values() for hana_type in types} +_HANA_TYPES.update({"SMALLINT", "TINYINT", "REAL", "NCLOB", "CLOB", "SHORTTEXT"}) + + +def _get_base_type(type_name: str | None) -> str | None: + if type_name is None: + return None + for separator in ("(", "<"): + if separator in type_name: + return type_name[: type_name.index(separator)].strip().upper() + return type_name.strip().upper() + + +def _get_base_type_raw(type_name: str | None) -> str | None: + if type_name is None: + return None + for separator in ("(", "<"): + if separator in type_name: + return type_name[: type_name.index(separator)].strip() + return type_name.strip() + + +def types_match(hana_catalog_type: str, expected_type: str) -> bool: + catalog_base_type = _get_base_type(hana_catalog_type) + expected_base_type = _get_base_type(expected_type) + expected_base_type_raw = _get_base_type_raw(expected_type) + if catalog_base_type is None or expected_base_type is None: + return False + + if expected_base_type in _HANA_TYPES and expected_base_type_raw != expected_base_type_raw.lower(): + return catalog_base_type == expected_base_type + + accepted_hana_types = _HANA_LOGICAL_TYPE_MAPPING.get(expected_base_type_raw.lower()) + if accepted_hana_types is None: + return catalog_base_type == expected_base_type + + return catalog_base_type in accepted_hana_types + + +def _custom_properties_dict(field: Any) -> dict[str, Any]: + if isinstance(field, SchemaProperty): + if field.customProperties is None: + return {} + return {custom_property.property: custom_property.value for custom_property in field.customProperties} + config = getattr(field, "config", None) + if config is None: + return {} + return config + + +def _get_type(field: Any) -> str | None: + if isinstance(field, SchemaProperty): + return field.physicalType or field.logicalType + return getattr(field, "type", None) + + +def _get_base_logical_type(field: Any) -> str | None: + field_type = _get_type(field) + if field_type is None: + return None + for separator in ("(", "<"): + if separator in field_type: + return field_type[: field_type.index(separator)].strip().lower() + return field_type.lower() + + +def _get_params(field: Any) -> str | None: + field_type = _get_type(field) + if field_type and "(" in field_type and field_type.endswith(")"): + return field_type[field_type.index("(") + 1 : -1] + return None + + +def _get_precision(field: Any) -> int | None: + if ( + isinstance(field, SchemaProperty) + and field.logicalTypeOptions + and field.logicalTypeOptions.get("precision") is not None + ): + return int(field.logicalTypeOptions.get("precision")) + precision = getattr(field, "precision", None) + if precision is not None: + return int(precision) + config_value = _custom_properties_dict(field).get("precision") + if config_value: + return int(config_value) + params = _get_params(field) + if params: + try: + return int(params.split(",")[0].strip()) + except ValueError: + return None + return None + + +def _get_scale(field: Any) -> int | None: + if ( + isinstance(field, SchemaProperty) + and field.logicalTypeOptions + and field.logicalTypeOptions.get("scale") is not None + ): + return int(field.logicalTypeOptions.get("scale")) + scale = getattr(field, "scale", None) + if scale is not None: + return int(scale) + config_value = _custom_properties_dict(field).get("scale") + if config_value: + return int(config_value) + params = _get_params(field) + if params and "," in params: + try: + return int(params.split(",")[1].strip()) + except ValueError: + return None + return None + + +def _attach_params_if_present(base_type: str, field: Any, *, default: str | None = None) -> str: + params = _get_params(field) + if params: + return f"{base_type}({params})" + if default: + return f"{base_type}({default})" + return base_type + + +def _decimal_type(field: Any) -> str: + precision = _get_precision(field) + scale = _get_scale(field) + if precision is not None and scale is not None: + return f"DECIMAL({precision},{scale})" + if precision is not None: + return f"DECIMAL({precision})" + return _attach_params_if_present("DECIMAL", field) + + +def convert_type_to_hana(field: Any) -> str | None: + hana_type = _custom_properties_dict(field).get("hanaType") + if hana_type: + return hana_type + + base_type = _get_base_logical_type(field) + if base_type is None: + return None + + if base_type in ["string", "varchar", "text", "nvarchar"]: + return _attach_params_if_present("NVARCHAR", field, default="5000") + if base_type in ["integer", "int"]: + return "INTEGER" + if base_type in ["long", "bigint"]: + return "BIGINT" + if base_type == "smallint": + return "SMALLINT" + if base_type == "tinyint": + return "TINYINT" + if base_type == "float": + return "REAL" + if base_type == "double": + return "DOUBLE" + if base_type in ["decimal", "numeric", "number"]: + return _decimal_type(field) + if base_type == "boolean": + return "BOOLEAN" + if base_type == "date": + return "DATE" + if base_type == "time": + return "TIME" + if base_type in ["timestamp", "timestamp_tz"]: + return "TIMESTAMP" + if base_type == "timestamp_ntz": + return "SECONDDATE" + if base_type in ["binary", "bytes", "varbinary"]: + return _attach_params_if_present("VARBINARY", field) + if base_type in ["object", "record", "struct", "array"]: + return "NCLOB" + if _get_params(field): + return _get_type(field) + return None diff --git a/datacontract/export/sodacl_check_builder.py b/datacontract/export/sodacl_check_builder.py index c31b78dc0..a77f9df75 100644 --- a/datacontract/export/sodacl_check_builder.py +++ b/datacontract/export/sodacl_check_builder.py @@ -60,7 +60,7 @@ def _quote_field_name(field_name: str, quoting_config: QuotingConfig) -> str: _BACKTICK_DIALECTS = {"databricks", "bigquery", "mysql", "impala", "dataframe", "kafka"} -_ANSI_QUOTING_DIALECTS = {"postgres", "redshift", "sqlserver", "snowflake", "azure", "s3", "gcs", "local"} +_ANSI_QUOTING_DIALECTS = {"postgres", "redshift", "sqlserver", "snowflake", "azure", "s3", "gcs", "local", "hana"} _BARE_IDENTIFIER_STRICT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _BARE_IDENTIFIER_PERMISSIVE = re.compile(r"^[A-Za-z_][A-Za-z0-9_$]*$") @@ -84,7 +84,7 @@ def _has_unsupported_databricks_type(prop) -> bool: return False -_PERMISSIVE_BARE_DIALECTS = {"postgres", "redshift", "snowflake", "oracle", "sqlserver"} +_PERMISSIVE_BARE_DIALECTS = {"postgres", "redshift", "snowflake", "oracle", "sqlserver", "hana"} def _quote_identifier_if_needed(identifier: str, server: Optional[Server]) -> str: diff --git a/datacontract/export/sql_type_converter.py b/datacontract/export/sql_type_converter.py index ed28435b9..926561782 100644 --- a/datacontract/export/sql_type_converter.py +++ b/datacontract/export/sql_type_converter.py @@ -209,6 +209,8 @@ def _convert_base_to_sql_type(field: Union[SchemaProperty, FieldLike], server_ty return convert_type_to_oracle(field) elif server_type == "clickhouse": return convert_type_to_clickhouse(field) + elif server_type == "hana": + return convert_type_to_hana(field) # Fallback: return the raw type string (preserves original behavior for unknown/None server_type) return _get_type(field) @@ -979,3 +981,9 @@ def convert_type_to_clickhouse(field: Union[SchemaProperty, FieldLike]) -> None if _get_params(field): return _get_type(field) return _warn_cannot_map_type(field, "clickhouse") + + +def convert_type_to_hana(field: Union[SchemaProperty, FieldLike]) -> None | str: + from datacontract.engines.hana.hana_type_mapping import convert_type_to_hana as _convert_type_to_hana + + return _convert_type_to_hana(field) diff --git a/datacontract/schemas/datacontract-1.2.1.schema.json b/datacontract/schemas/datacontract-1.2.1.schema.json index b888b1e9c..d42270c73 100644 --- a/datacontract/schemas/datacontract-1.2.1.schema.json +++ b/datacontract/schemas/datacontract-1.2.1.schema.json @@ -1148,6 +1148,7 @@ "databricks", "dataframe", "glue", + "hana", "kafka", "kinesis", "local", diff --git a/datacontract/schemas/odcs-3.0.1.schema.json b/datacontract/schemas/odcs-3.0.1.schema.json index e501aad32..21e8f1633 100644 --- a/datacontract/schemas/odcs-3.0.1.schema.json +++ b/datacontract/schemas/odcs-3.0.1.schema.json @@ -180,6 +180,7 @@ "glue", "cloudsql", "db2", + "hana", "informix", "kafka", "kinesis", diff --git a/datacontract/schemas/odcs-3.0.2.schema.json b/datacontract/schemas/odcs-3.0.2.schema.json index cb9bfb9d8..fbd5622c3 100644 --- a/datacontract/schemas/odcs-3.0.2.schema.json +++ b/datacontract/schemas/odcs-3.0.2.schema.json @@ -147,7 +147,7 @@ "description": "Type of the server.", "enum": [ "api", "athena", "azure", "bigquery", "clickhouse", "databricks", "denodo", "dremio", - "duckdb", "glue", "cloudsql", "db2", "informix", "kafka", "kinesis", "local", + "duckdb", "glue", "cloudsql", "db2", "hana", "informix", "kafka", "kinesis", "local", "mysql", "oracle", "postgresql", "postgres", "presto", "pubsub", "redshift", "s3", "sftp", "snowflake", "sqlserver", "synapse", "trino", "vertica", "custom" ] diff --git a/datacontract/schemas/odcs-3.1.0.schema.json b/datacontract/schemas/odcs-3.1.0.schema.json index 79da72bf9..3289405f8 100644 --- a/datacontract/schemas/odcs-3.1.0.schema.json +++ b/datacontract/schemas/odcs-3.1.0.schema.json @@ -177,7 +177,7 @@ "description": "Type of the server.", "enum": [ "api", "athena", "azure", "bigquery", "clickhouse", "databricks", "denodo", "dremio", - "duckdb", "glue", "cloudsql", "db2", "hive", "impala", "informix", "kafka", "kinesis", "local", + "duckdb", "glue", "cloudsql", "db2", "hana", "hive", "impala", "informix", "kafka", "kinesis", "local", "mysql", "oracle", "postgresql", "postgres", "presto", "pubsub", "redshift", "s3", "sftp", "snowflake", "sqlserver", "synapse", "trino", "vertica", "zen", "custom" ] @@ -489,6 +489,19 @@ "$ref": "#/$defs/ServerSource/OracleServer" } }, + { + "if": { + "properties": { + "type": { + "const": "hana" + } + }, + "required": ["type"] + }, + "then": { + "$ref": "#/$defs/ServerSource/HanaServer" + } + }, { "if": { "properties": { @@ -1255,6 +1268,30 @@ "serviceName" ] }, + "HanaServer": { + "type": "object", + "title": "HanaServer", + "properties": { + "host": { + "type": "string", + "description": "The host to the SAP HANA Cloud server." + }, + "port": { + "type": "integer", + "description": "The port to the SAP HANA Cloud server. Default 443 for HANA Cloud.", + "default": 443 + }, + "schema": { + "type": "string", + "description": "The name of the schema. For SAP Datasphere, use the Open SQL schema name." + } + }, + "required": [ + "host", + "port", + "schema" + ] + }, "PostgresServer": { "type": "object", "title": "PostgresServer", diff --git a/pyproject.toml b/pyproject.toml index 3a0337617..978d5a97a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -139,6 +139,10 @@ athena = [ "ibis-framework[athena]>=10.0.0,<13.0.0" ] +hana = [ + "hdbcli>=2.20.0,<3.0.0", +] + trino = [ "ibis-framework[trino]>=10.0.0,<13.0.0" ] @@ -177,6 +181,11 @@ protobuf = [ "proto-schema-parser>=1.0,<3.0", ] +# NOTE: `hana` is intentionally NOT part of `all`. It depends on `hdbcli`, the +# SAP HANA Client, which is distributed under SAP's proprietary license (SAP +# Developer License), not an OSI-approved open-source license. Bundling it into +# the `all` meta-extra would pull a proprietary package into a default-ish +# install, so HANA support must be opted into explicitly via `datacontract-cli[hana]`. all = [ "datacontract-cli[kafka,bigquery,csv,excel,snowflake,postgres,redshift,mysql,databricks,sqlserver,s3,gcs,azure,athena,trino,impala,dbml,duckdb,iceberg,parquet,rdf,api,protobuf,oracle]" ] diff --git a/tests/fixtures/hana/datacontract_hana_all_types.yaml b/tests/fixtures/hana/datacontract_hana_all_types.yaml new file mode 100644 index 000000000..eca9cacc3 --- /dev/null +++ b/tests/fixtures/hana/datacontract_hana_all_types.yaml @@ -0,0 +1,42 @@ +apiVersion: v3.1.0 +kind: DataContract +id: hana-all-types +name: HANA all supported types contract +version: 1.0.0 +status: draft +schema: +- name: ALL_TYPES + properties: + - name: STRING_COL + logicalType: string + - name: INTEGER_COL + logicalType: integer + - name: BIGINT_COL + logicalType: integer + physicalType: BIGINT + - name: FLOAT_COL + logicalType: number + physicalType: REAL + - name: DOUBLE_COL + logicalType: number + physicalType: DOUBLE + - name: DECIMAL_COL + logicalType: number + physicalType: DECIMAL + - name: BOOLEAN_COL + logicalType: boolean + - name: DATE_COL + logicalType: date + - name: TIME_COL + logicalType: time + - name: TIMESTAMP_COL + logicalType: timestamp + - name: BINARY_COL + logicalType: string + physicalType: VARBINARY +servers: +- server: production + type: hana + host: ${DATACONTRACT_HANA_HOST} + port: 443 + schema: ${DATACONTRACT_HANA_SCHEMA} diff --git a/tests/fixtures/hana/datacontract_hana_basic.yaml b/tests/fixtures/hana/datacontract_hana_basic.yaml new file mode 100644 index 000000000..9d75022c4 --- /dev/null +++ b/tests/fixtures/hana/datacontract_hana_basic.yaml @@ -0,0 +1,26 @@ +apiVersion: v3.1.0 +kind: DataContract +id: hana-basic +name: HANA basic contract +version: 1.0.0 +status: draft +schema: +- name: ORDERS + physicalName: ORDERS + properties: + - name: ORDER_ID + logicalType: string + physicalType: NVARCHAR + required: true + primaryKey: true + - name: AMOUNT + logicalType: number + physicalType: DECIMAL(15,2) + - name: UPDATED_AT + logicalType: timestamp +servers: +- server: production + type: hana + host: ${DATACONTRACT_HANA_HOST} + port: 443 + schema: ${DATACONTRACT_HANA_SCHEMA} diff --git a/tests/fixtures/hana/datacontract_hana_fail_schema.yaml b/tests/fixtures/hana/datacontract_hana_fail_schema.yaml new file mode 100644 index 000000000..21568dc41 --- /dev/null +++ b/tests/fixtures/hana/datacontract_hana_fail_schema.yaml @@ -0,0 +1,17 @@ +apiVersion: v3.1.0 +kind: DataContract +id: hana-fail-schema +name: HANA failing schema contract +version: 1.0.0 +status: draft +schema: +- name: ORDERS + properties: + - name: ORDER_ID + logicalType: integer +servers: +- server: production + type: hana + host: ${DATACONTRACT_HANA_HOST} + port: 443 + schema: ${DATACONTRACT_HANA_SCHEMA} diff --git a/tests/fixtures/hana/datacontract_hana_quality.yaml b/tests/fixtures/hana/datacontract_hana_quality.yaml new file mode 100644 index 000000000..b8303f735 --- /dev/null +++ b/tests/fixtures/hana/datacontract_hana_quality.yaml @@ -0,0 +1,35 @@ +apiVersion: v3.1.0 +kind: DataContract +id: hana-quality +name: HANA quality contract +version: 1.0.0 +status: draft +schema: +- name: ORDERS + quality: + - type: sql + description: ORDERS has rows + query: SELECT COUNT(*) FROM {model} + mustBeGreaterThan: 0 + properties: + - name: ORDER_ID + logicalType: string + quality: + - metric: duplicateValues + mustBe: 0 + - name: STATUS + logicalType: string + quality: + - metric: invalidValues + arguments: + validValues: + - OPEN + - PAID + - CANCELLED + mustBe: 0 +servers: +- server: production + type: hana + host: ${DATACONTRACT_HANA_HOST} + port: 443 + schema: ${DATACONTRACT_HANA_SCHEMA} diff --git a/tests/fixtures/hana/datacontract_hana_sla.yaml b/tests/fixtures/hana/datacontract_hana_sla.yaml new file mode 100644 index 000000000..8deb6a31e --- /dev/null +++ b/tests/fixtures/hana/datacontract_hana_sla.yaml @@ -0,0 +1,27 @@ +apiVersion: v3.1.0 +kind: DataContract +id: hana-sla +name: HANA SLA contract +version: 1.0.0 +status: draft +schema: +- name: ORDERS + properties: + - name: CREATED_AT + logicalType: timestamp + - name: UPDATED_AT + logicalType: timestamp +slaProperties: +- property: freshness + element: ORDERS.UPDATED_AT + value: 24 + unit: h +- property: retention + element: ORDERS.CREATED_AT + value: P1Y +servers: +- server: production + type: hana + host: ${DATACONTRACT_HANA_HOST} + port: 443 + schema: ${DATACONTRACT_HANA_SCHEMA} diff --git a/tests/test_hana_connection.py b/tests/test_hana_connection.py new file mode 100644 index 000000000..68249bbfd --- /dev/null +++ b/tests/test_hana_connection.py @@ -0,0 +1,99 @@ +import sys +import types + +import pytest +from open_data_contract_standard.model import Server + +from datacontract.engines.hana.hana_connection import get_connection +from datacontract.model.exceptions import DataContractException + + +class FakeDbApi: + def __init__(self): + self.kwargs = None + + def connect(self, **kwargs): + self.kwargs = kwargs + return object() + + +def install_hdbcli(monkeypatch): + dbapi = FakeDbApi() + hdbcli = types.ModuleType("hdbcli") + hdbcli.dbapi = dbapi + monkeypatch.setitem(sys.modules, "hdbcli", hdbcli) + return dbapi + + +def hana_server(**kwargs): + defaults = {"type": "hana", "host": "hana.example.com", "schema": "SALES"} + defaults.update(kwargs) + return Server(**defaults) + + +def test_connection_missing_username(monkeypatch): + install_hdbcli(monkeypatch) + monkeypatch.delenv("DATACONTRACT_HANA_USERNAME", raising=False) + monkeypatch.setenv("DATACONTRACT_HANA_PASSWORD", "secret") + + with pytest.raises(DataContractException) as exc_info: + get_connection(hana_server()) + + assert exc_info.value.type == "hana-connection" + + +def test_connection_missing_password(monkeypatch): + install_hdbcli(monkeypatch) + monkeypatch.setenv("DATACONTRACT_HANA_USERNAME", "user") + monkeypatch.delenv("DATACONTRACT_HANA_PASSWORD", raising=False) + + with pytest.raises(DataContractException) as exc_info: + get_connection(hana_server()) + + assert exc_info.value.type == "hana-connection" + + +def test_connection_default_port_443(monkeypatch): + dbapi = install_hdbcli(monkeypatch) + monkeypatch.setenv("DATACONTRACT_HANA_USERNAME", "user") + monkeypatch.setenv("DATACONTRACT_HANA_PASSWORD", "secret") + + get_connection(hana_server()) + + assert dbapi.kwargs["port"] == 443 + + +def test_connection_encrypt_default_true(monkeypatch): + dbapi = install_hdbcli(monkeypatch) + monkeypatch.setenv("DATACONTRACT_HANA_USERNAME", "user") + monkeypatch.setenv("DATACONTRACT_HANA_PASSWORD", "secret") + monkeypatch.delenv("DATACONTRACT_HANA_ENCRYPT", raising=False) + + get_connection(hana_server()) + + assert dbapi.kwargs["encrypt"] is True + + +def test_connection_encrypt_env_override(monkeypatch): + dbapi = install_hdbcli(monkeypatch) + monkeypatch.setenv("DATACONTRACT_HANA_USERNAME", "user") + monkeypatch.setenv("DATACONTRACT_HANA_PASSWORD", "secret") + monkeypatch.setenv("DATACONTRACT_HANA_ENCRYPT", "false") + + get_connection(hana_server()) + + assert dbapi.kwargs["encrypt"] is False + + +def test_connection_ssl_validate_certificate(monkeypatch): + dbapi = install_hdbcli(monkeypatch) + monkeypatch.setenv("DATACONTRACT_HANA_USERNAME", "user") + monkeypatch.setenv("DATACONTRACT_HANA_PASSWORD", "secret") + monkeypatch.setenv("DATACONTRACT_HANA_SSL_VALIDATE_CERTIFICATE", "0") + monkeypatch.setenv("DATACONTRACT_HANA_SSL_HOSTNAME_IN_CERTIFICATE", "*.example.com") + + get_connection(hana_server(port=30015)) + + assert dbapi.kwargs["sslValidateCertificate"] is False + assert dbapi.kwargs["sslHostnameInCertificate"] == "*.example.com" + assert dbapi.kwargs["port"] == 30015 diff --git a/tests/test_hana_quality_check.py b/tests/test_hana_quality_check.py new file mode 100644 index 000000000..7e9ba2adf --- /dev/null +++ b/tests/test_hana_quality_check.py @@ -0,0 +1,220 @@ +from open_data_contract_standard.model import ( + DataQuality, + OpenDataContractStandard, + SchemaObject, + SchemaProperty, + ServiceLevelAgreementProperty, +) + +from datacontract.engines.hana.hana_quality_check import ( + prepare_hana_query, + run_quality_checks, + run_sla_checks, +) +from datacontract.model.run import ResultEnum + + +class FakeConnection: + def __init__(self, response=(0,)): + self.response = response + self.executed = [] + + def cursor(self): + return FakeCursor(self) + + +class FakeCursor: + def __init__(self, connection): + self.connection = connection + + def execute(self, sql, params=None): + self.connection.executed.append((sql, params or [])) + + def fetchone(self): + return self.connection.response + + def close(self): + pass + + +def check_by_type(checks, check_type): + return next(check for check in checks if check.type == check_type) + + +def test_sql_quality_pass(): + connection = FakeConnection((5,)) + schema = SchemaObject( + name="ORDERS", + quality=[DataQuality(type="sql", query="SELECT COUNT(*) FROM {model}", mustBe=5)], + ) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_quality_sql").result == ResultEnum.passed + assert connection.executed[0][0] == 'SELECT COUNT(*) FROM "SALES"."ORDERS"' + + +def test_sql_quality_fail(): + connection = FakeConnection((4,)) + schema = SchemaObject( + name="ORDERS", + quality=[DataQuality(type="sql", query="SELECT COUNT(*) FROM {model}", mustBe=5)], + ) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_quality_sql").result == ResultEnum.failed + + +def test_sql_quality_threshold_operators(): + cases = [ + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustNotBe=4), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeGreaterThan=4), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeGreaterOrEqualTo=5), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeLessThan=6), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeLessOrEqualTo=5), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeBetween=[1, 5]), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustNotBeBetween=[6, 9]), 5, ResultEnum.passed), + (DataQuality(type="sql", query="SELECT 5 FROM {model}", mustBeGreaterThan=5), 5, ResultEnum.failed), + ] + + for quality, value, expected in cases: + connection = FakeConnection((value,)) + schema = SchemaObject(name="ORDERS", quality=[quality]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_quality_sql").result == expected + + +def test_row_count(): + connection = FakeConnection((10,)) + schema = SchemaObject(name="ORDERS", quality=[DataQuality(metric="rowCount", mustBe=10)]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "row_count").result == ResultEnum.passed + assert connection.executed[0][0] == 'SELECT COUNT(*) FROM "SALES"."ORDERS"' + + +def test_duplicate_values_field(): + connection = FakeConnection((0,)) + prop = SchemaProperty(name="ID", quality=[DataQuality(metric="duplicateValues", mustBe=0)]) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_duplicate_values").result == ResultEnum.passed + assert 'COUNT(DISTINCT "ID")' in connection.executed[0][0] + + +def test_duplicate_values_model(): + connection = FakeConnection((1,)) + quality = DataQuality(metric="duplicateValues", arguments={"properties": ["ID", "STATUS"]}, mustBe=0) + schema = SchemaObject(name="ORDERS", quality=[quality]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_duplicate_values").result == ResultEnum.failed + assert 'COUNT(DISTINCT "ID", "STATUS")' in connection.executed[0][0] + + +def test_null_values(): + connection = FakeConnection((0,)) + prop = SchemaProperty(name="ID", quality=[DataQuality(metric="nullValues", mustBe=0)]) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_null_values").result == ResultEnum.passed + assert '"ID" IS NULL' in connection.executed[0][0] + + +def test_invalid_values(): + connection = FakeConnection((0,)) + quality = DataQuality(metric="invalidValues", arguments={"validValues": ["OPEN", "PAID"]}, mustBe=0) + prop = SchemaProperty(name="STATUS", quality=[quality]) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_invalid_values").result == ResultEnum.passed + assert connection.executed[0][1] == ["OPEN", "PAID"] + + +def test_missing_values(): + connection = FakeConnection((2,)) + quality = DataQuality(metric="missingValues", arguments={"missingValues": ["", "N/A", None]}, mustBe=0) + prop = SchemaProperty(name="STATUS", quality=[quality]) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_quality_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_missing_values").result == ResultEnum.failed + assert connection.executed[0][1] == ["", "N/A"] + assert '"STATUS" IS NULL OR "STATUS" IN (?, ?)' in connection.executed[0][0] + + +def test_custom_sodacl_warning(): + quality = DataQuality(type="custom", engine="soda", implementation="checks for ORDERS: []") + schema = SchemaObject(name="ORDERS", quality=[quality]) + + checks = run_quality_checks(FakeConnection(), "SALES", schema) + + assert check_by_type(checks, "quality_custom_soda").result == ResultEnum.warning + + +def test_freshness_check(): + connection = FakeConnection((60,)) + contract = OpenDataContractStandard( + id="hana-test", + schema=[SchemaObject(name="ORDERS", properties=[SchemaProperty(name="UPDATED_AT")])], + slaProperties=[ + ServiceLevelAgreementProperty(property="freshness", element="ORDERS.UPDATED_AT", value=5, unit="m") + ], + ) + + checks = run_sla_checks(connection, "SALES", contract) + + assert check_by_type(checks, "servicelevel_freshness").result == ResultEnum.passed + assert "SECONDS_BETWEEN(MAX" in connection.executed[0][0] + + +def test_retention_check(): + connection = FakeConnection((4000,)) + contract = OpenDataContractStandard( + id="hana-test", + schema=[SchemaObject(name="ORDERS", properties=[SchemaProperty(name="CREATED_AT")])], + slaProperties=[ + ServiceLevelAgreementProperty(property="retention", element="ORDERS.CREATED_AT", value=2, unit="h") + ], + ) + + checks = run_sla_checks(connection, "SALES", contract) + + assert check_by_type(checks, "servicelevel_retention").result == ResultEnum.passed + assert "SECONDS_BETWEEN(MIN" in connection.executed[0][0] + + +def test_query_placeholder_replacement(): + placeholders = [ + "{model}", + "${model}", + '"{model}"', + "'${model}'", + "{table}", + "{object}", + ] + for placeholder in placeholders: + assert prepare_hana_query(f"SELECT * FROM {placeholder}", "SALES", "ORDERS") == 'SELECT * FROM "SALES"."ORDERS"' + + assert prepare_hana_query("SELECT * FROM {schema}", "SALES", "ORDERS") == 'SELECT * FROM "SALES"' + assert prepare_hana_query("SELECT {field}, {column}, {property} FROM {model}", "SALES", "ORDERS", "ID") == ( + 'SELECT "ID", "ID", "ID" FROM "SALES"."ORDERS"' + ) + + +def test_identifier_quoting_escapes_quotes(): + query = prepare_hana_query("SELECT {field} FROM {model}", 'S"1', 'T"1', 'C"1') + + assert query == 'SELECT "C""1" FROM "S""1"."T""1"' diff --git a/tests/test_hana_schema_check.py b/tests/test_hana_schema_check.py new file mode 100644 index 000000000..f28624afb --- /dev/null +++ b/tests/test_hana_schema_check.py @@ -0,0 +1,269 @@ +from open_data_contract_standard.model import SchemaObject, SchemaProperty + +from datacontract.engines.hana.hana_schema_check import ( + qualified_table_name, + quote_identifier, + run_schema_checks, +) +from datacontract.model.run import ResultEnum + + +class FakeConnection: + def __init__(self, responses=None): + self.responses = responses or [] + self.executed = [] + + def cursor(self): + return FakeCursor(self) + + def response_for(self, sql, params): + self.executed.append((sql, params)) + for matcher, response in self.responses: + if matcher(sql, params): + return response + return [] + + +class FakeCursor: + def __init__(self, connection): + self.connection = connection + self.response = [] + + def execute(self, sql, params=None): + self.response = self.connection.response_for(sql, params or []) + + def fetchall(self): + return self.response + + def fetchone(self): + return self.response[0] if self.response else None + + def close(self): + pass + + +def has_sql(fragment): + return lambda sql, params: fragment in " ".join(sql.split()) + + +def has_params(*expected): + return lambda sql, params: params == list(expected) + + +def catalog_response(rows, source="table"): + table_rows = rows if source == "table" else [] + view_rows = rows if source == "view" else [] + return [ + (has_sql("FROM SYS.TABLE_COLUMNS"), table_rows), + (has_sql("FROM SYS.VIEW_COLUMNS"), view_rows), + ] + + +def column(name="ID", data_type="INTEGER", nullable="FALSE"): + return (name, data_type, None, None, nullable, 1) + + +def check_by_type(checks, check_type): + return next(check for check in checks if check.type == check_type) + + +def test_model_exists_pass(): + connection = FakeConnection(catalog_response([column()])) + schema = SchemaObject(name="ORDERS", properties=[]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_exists").result == ResultEnum.passed + + +def test_model_exists_fail(): + connection = FakeConnection(catalog_response([])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", logicalType="integer")]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert len(checks) == 1 + assert check_by_type(checks, "model_exists").result == ResultEnum.failed + + +def test_field_present_pass(): + connection = FakeConnection(catalog_response([column()])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID")]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_is_present").result == ResultEnum.passed + + +def test_field_present_fail(): + connection = FakeConnection(catalog_response([column("OTHER")])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID")]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_is_present").result == ResultEnum.failed + + +def test_field_type_match(): + connection = FakeConnection(catalog_response([column(data_type="INTEGER")])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", logicalType="integer")]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_type").result == ResultEnum.passed + + +def test_field_type_mismatch(): + connection = FakeConnection(catalog_response([column(data_type="NVARCHAR")])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", logicalType="integer")]) + + checks = run_schema_checks(connection, "SALES", schema) + check = check_by_type(checks, "field_type") + + assert check.result == ResultEnum.failed + assert check.diagnostics == {"expected": "integer", "actual": "NVARCHAR"} + + +def test_field_required_nullable_false(): + connection = FakeConnection(catalog_response([column(nullable="FALSE")])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", required=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_required").result == ResultEnum.passed + + +def test_field_required_nullable_true(): + connection = FakeConnection(catalog_response([column(nullable="TRUE")])) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", required=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_required").result == ResultEnum.failed + + +def test_field_unique_pass(): + connection = FakeConnection(catalog_response([column()]) + [(has_sql("COUNT(*) - COUNT(DISTINCT"), [(0,)])]) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", unique=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_unique").result == ResultEnum.passed + + +def test_field_unique_fail(): + connection = FakeConnection(catalog_response([column()]) + [(has_sql("COUNT(*) - COUNT(DISTINCT"), [(2,)])]) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", unique=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_unique").result == ResultEnum.failed + + +def test_field_min_length_and_max_length(): + connection = FakeConnection( + catalog_response([column("NAME", "NVARCHAR")]) + + [ + (has_params(3), [(0,)]), + (has_params(10), [(1,)]), + ] + ) + prop = SchemaProperty(name="NAME", logicalType="string", logicalTypeOptions={"minLength": 3, "maxLength": 10}) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_min_length").result == ResultEnum.passed + assert check_by_type(checks, "field_max_length").result == ResultEnum.failed + + +def test_field_minimum_maximum_and_not_equal(): + connection = FakeConnection( + catalog_response([column("AMOUNT", "DECIMAL")]) + + [ + (has_params(1), [(0,)]), + (has_params(10), [(1,)]), + (has_params(0), [(0,)]), + ] + ) + prop = SchemaProperty( + name="AMOUNT", + logicalType="decimal", + logicalTypeOptions={"minimum": 1, "maximum": 10, "exclusiveMinimum": 0}, + ) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_minimum").result == ResultEnum.passed + assert check_by_type(checks, "field_maximum").result == ResultEnum.failed + assert check_by_type(checks, "field_not_equal").result == ResultEnum.passed + + +def test_field_enum_pass_and_fail(): + connection = FakeConnection(catalog_response([column("STATUS", "NVARCHAR")]) + [(has_sql("NOT IN"), [(1,)])]) + prop = SchemaProperty(name="STATUS", logicalType="string", logicalTypeOptions={"enum": ["PAID", "OPEN"]}) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_enum").result == ResultEnum.failed + + +def test_field_regex_uses_like_regexpr(): + connection = FakeConnection(catalog_response([column("EMAIL", "NVARCHAR")]) + [(has_sql("LIKE_REGEXPR"), [(0,)])]) + prop = SchemaProperty(name="EMAIL", logicalType="string", logicalTypeOptions={"pattern": ".*@example.com"}) + schema = SchemaObject(name="ORDERS", properties=[prop]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_regex").result == ResultEnum.passed + assert "LIKE_REGEXPR" in connection.executed[-1][0] + + +def test_primary_key_table(): + responses = catalog_response([column()]) + [(has_sql("FROM SYS.CONSTRAINTS"), [("ID",)])] + connection = FakeConnection(responses) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", primaryKey=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_primary_key").result == ResultEnum.passed + + +def test_primary_key_view_warning(): + connection = FakeConnection(catalog_response([column()], source="view")) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", primaryKey=True)]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_primary_key").result == ResultEnum.warning + + +def test_view_fallback(): + connection = FakeConnection(catalog_response([column()], source="view")) + schema = SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID")]) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "model_exists").result == ResultEnum.passed + assert any("FROM SYS.VIEW_COLUMNS" in sql for sql, params in connection.executed) + + +def test_physical_name_resolution(): + connection = FakeConnection(catalog_response([column("ORDER_ID")])) + schema = SchemaObject( + name="orders", + physicalName="ORDERS", + properties=[SchemaProperty(name="order_id", physicalName="ORDER_ID")], + ) + + checks = run_schema_checks(connection, "SALES", schema) + + assert check_by_type(checks, "field_is_present").model == "ORDERS" + assert check_by_type(checks, "field_is_present").field == "ORDER_ID" + + +def test_identifier_quoting_escapes_quotes(): + assert quote_identifier('A"B') == '"A""B"' + assert qualified_table_name('S"1', 'T"1') == '"S""1"."T""1"' diff --git a/tests/test_hana_type_mapping.py b/tests/test_hana_type_mapping.py new file mode 100644 index 000000000..8bd62573c --- /dev/null +++ b/tests/test_hana_type_mapping.py @@ -0,0 +1,75 @@ +from open_data_contract_standard.model import SchemaProperty + +from datacontract.engines.hana.hana_type_mapping import convert_type_to_hana, types_match + + +def test_convert_type_to_hana_string(): + field = SchemaProperty(name="col", logicalType="string") + + assert convert_type_to_hana(field) == "NVARCHAR(5000)" + + +def test_convert_type_to_hana_integer(): + field = SchemaProperty(name="col", logicalType="integer") + + assert convert_type_to_hana(field) == "INTEGER" + + +def test_convert_type_to_hana_decimal_with_params(): + field = SchemaProperty(name="col", logicalType="decimal", logicalTypeOptions={"precision": 15, "scale": 2}) + + assert convert_type_to_hana(field) == "DECIMAL(15,2)" + + +def test_convert_type_to_hana_custom_config(): + field = SchemaProperty( + name="col", + logicalType="string", + customProperties=[{"property": "hanaType", "value": "ST_GEOMETRY"}], + ) + + assert convert_type_to_hana(field) == "ST_GEOMETRY" + + +def test_convert_type_to_hana_all_mappings(): + expected = { + "varchar": "NVARCHAR(5000)", + "text": "NVARCHAR(5000)", + "int": "INTEGER", + "long": "BIGINT", + "bigint": "BIGINT", + "smallint": "SMALLINT", + "tinyint": "TINYINT", + "float": "REAL", + "double": "DOUBLE", + "numeric": "DECIMAL", + "number": "DECIMAL", + "boolean": "BOOLEAN", + "date": "DATE", + "time": "TIME", + "timestamp": "TIMESTAMP", + "timestamp_tz": "TIMESTAMP", + "timestamp_ntz": "SECONDDATE", + "binary": "VARBINARY", + "bytes": "VARBINARY", + "object": "NCLOB", + "record": "NCLOB", + "struct": "NCLOB", + "array": "NCLOB", + } + + for logical_type, hana_type in expected.items(): + assert convert_type_to_hana(SchemaProperty(name="col", logicalType=logical_type)) == hana_type + + +def test_types_match_strips_precision_and_uses_physical_type(): + assert types_match("NVARCHAR(255)", "NVARCHAR") + assert types_match("DECIMAL(15,2)", "DECIMAL") + assert not types_match("NVARCHAR(255)", "VARCHAR") + + +def test_types_match_logical_mapping(): + assert types_match("NVARCHAR(255)", "string") + assert types_match("INTEGER", "int") + assert types_match("SECONDDATE", "timestamp_ntz") + assert not types_match("DATE", "timestamp") diff --git a/tests/test_test_hana.py b/tests/test_test_hana.py new file mode 100644 index 000000000..5c59b78ec --- /dev/null +++ b/tests/test_test_hana.py @@ -0,0 +1,63 @@ +import os + +import pytest +from open_data_contract_standard.model import OpenDataContractStandard, SchemaObject, SchemaProperty, Server + +from datacontract.data_contract import DataContract +from datacontract.engines.data_contract_test import execute_data_contract_test +from datacontract.model.run import Check, ResultEnum, Run + + +def test_hana_test_flow_bypasses_soda(monkeypatch): + from datacontract.engines.hana import check_hana_execute as hana_module + + def fake_check_hana_execute(run, data_contract, server, schema_name="all", check_categories=None): + run.checks.append(Check(type="model_exists", result=ResultEnum.passed, engine="hana")) + + monkeypatch.setattr(hana_module, "check_hana_execute", fake_check_hana_execute) + data_contract = OpenDataContractStandard( + id="hana-test", + schema=[SchemaObject(name="ORDERS", properties=[SchemaProperty(name="ID", logicalType="integer")])], + servers=[Server(server="production", type="hana", host="hana.example.com", port=443, schema="SALES")], + ) + run = Run.create_run() + + execute_data_contract_test(data_contract, run) + + assert [check.engine for check in run.checks] == ["hana"] + assert run.checks[0].result == ResultEnum.passed + + +_required_hana_env = [ + "DATACONTRACT_HANA_HOST", + "DATACONTRACT_HANA_SCHEMA", + "DATACONTRACT_HANA_USERNAME", + "DATACONTRACT_HANA_PASSWORD", +] +hana_integration = pytest.mark.skipif( + any(os.environ.get(name) is None for name in _required_hana_env), + reason="SAP HANA Cloud integration environment variables are not set.", +) + + +@hana_integration +def test_hana_full_contract_pass(): + run = DataContract(data_contract_file="fixtures/hana/datacontract_hana_basic.yaml").test() + + assert run.result in (ResultEnum.passed, ResultEnum.warning) + assert not any(check.result in (ResultEnum.failed, ResultEnum.error) for check in run.checks) + + +@hana_integration +def test_hana_full_contract_fail_schema(): + run = DataContract(data_contract_file="fixtures/hana/datacontract_hana_fail_schema.yaml").test() + + assert run.result == ResultEnum.failed + assert all(check.engine == "hana" for check in run.checks) + + +@hana_integration +def test_hana_quality_check_live(): + run = DataContract(data_contract_file="fixtures/hana/datacontract_hana_quality.yaml").test() + + assert all(check.engine == "hana" for check in run.checks)