Initial commit
This commit is contained in:
git
3
tests/__init__.py
Executable file
3
tests/__init__.py
Executable file
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
Test suite for Data Regression Testing Framework
|
||||
"""
|
||||
207
tests/test_config.py
Executable file
207
tests/test_config.py
Executable file
@@ -0,0 +1,207 @@
|
||||
"""
|
||||
Unit tests for configuration management
|
||||
"""
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from drt.config.models import (
|
||||
DatabaseConnection,
|
||||
DatabasePair,
|
||||
ComparisonSettings,
|
||||
RowCountSettings,
|
||||
SchemaSettings,
|
||||
AggregateSettings,
|
||||
ReportingSettings,
|
||||
LoggingSettings,
|
||||
Config
|
||||
)
|
||||
|
||||
|
||||
class TestDatabaseConnection:
|
||||
"""Test DatabaseConnection model"""
|
||||
|
||||
def test_database_connection_minimal(self):
|
||||
"""Test creating a minimal database connection"""
|
||||
conn = DatabaseConnection(
|
||||
server="SQLSERVER01",
|
||||
database="TestDB"
|
||||
)
|
||||
assert conn.server == "SQLSERVER01"
|
||||
assert conn.database == "TestDB"
|
||||
assert conn.timeout.connection == 30
|
||||
assert conn.timeout.query == 300
|
||||
|
||||
def test_database_connection_with_timeout(self):
|
||||
"""Test database connection with custom timeout"""
|
||||
conn = DatabaseConnection(
|
||||
server="SQLSERVER01",
|
||||
database="TestDB",
|
||||
timeout={"connection": 60, "query": 600}
|
||||
)
|
||||
assert conn.timeout.connection == 60
|
||||
assert conn.timeout.query == 600
|
||||
|
||||
|
||||
class TestDatabasePair:
|
||||
"""Test DatabasePair model"""
|
||||
|
||||
def test_database_pair_creation(self):
|
||||
"""Test creating a database pair"""
|
||||
pair = DatabasePair(
|
||||
name="Test_Pair",
|
||||
enabled=True,
|
||||
baseline=DatabaseConnection(
|
||||
server="SQLSERVER01",
|
||||
database="PROD_DB"
|
||||
),
|
||||
target=DatabaseConnection(
|
||||
server="SQLSERVER01",
|
||||
database="TEST_DB"
|
||||
)
|
||||
)
|
||||
assert pair.name == "Test_Pair"
|
||||
assert pair.enabled is True
|
||||
assert pair.baseline.database == "PROD_DB"
|
||||
assert pair.target.database == "TEST_DB"
|
||||
|
||||
|
||||
class TestComparisonSettings:
|
||||
"""Test ComparisonSettings model"""
|
||||
|
||||
def test_comparison_settings_health_check(self):
|
||||
"""Test health check mode settings"""
|
||||
settings = ComparisonSettings(
|
||||
mode="health_check",
|
||||
row_count=RowCountSettings(enabled=True, tolerance_percent=0.0),
|
||||
schema=SchemaSettings(
|
||||
enabled=True,
|
||||
checks={
|
||||
"column_names": True,
|
||||
"data_types": True
|
||||
}
|
||||
),
|
||||
aggregates=AggregateSettings(enabled=False)
|
||||
)
|
||||
assert settings.mode == "health_check"
|
||||
assert settings.row_count.enabled is True
|
||||
assert settings.aggregates.enabled is False
|
||||
|
||||
def test_comparison_settings_full_mode(self):
|
||||
"""Test full mode settings"""
|
||||
settings = ComparisonSettings(
|
||||
mode="full",
|
||||
row_count=RowCountSettings(enabled=True, tolerance_percent=0.0),
|
||||
schema=SchemaSettings(enabled=True),
|
||||
aggregates=AggregateSettings(enabled=True, tolerance_percent=0.01)
|
||||
)
|
||||
assert settings.mode == "full"
|
||||
assert settings.aggregates.enabled is True
|
||||
assert settings.aggregates.tolerance_percent == 0.01
|
||||
|
||||
|
||||
class TestReportingSettings:
|
||||
"""Test ReportingSettings model"""
|
||||
|
||||
def test_reporting_settings_defaults(self):
|
||||
"""Test default reporting settings"""
|
||||
settings = ReportingSettings()
|
||||
assert settings.output_dir == "./reports"
|
||||
assert settings.formats.html is True
|
||||
assert settings.formats.csv is True
|
||||
assert settings.formats.pdf is False
|
||||
assert settings.include_timestamp is True
|
||||
|
||||
def test_reporting_settings_custom(self):
|
||||
"""Test custom reporting settings"""
|
||||
settings = ReportingSettings(
|
||||
output_dir="./custom_reports",
|
||||
filename_prefix="custom_test",
|
||||
formats={"html": True, "csv": False, "pdf": True}
|
||||
)
|
||||
assert settings.output_dir == "./custom_reports"
|
||||
assert settings.filename_prefix == "custom_test"
|
||||
assert settings.formats.pdf is True
|
||||
|
||||
|
||||
class TestLoggingSettings:
|
||||
"""Test LoggingSettings model"""
|
||||
|
||||
def test_logging_settings_defaults(self):
|
||||
"""Test default logging settings"""
|
||||
settings = LoggingSettings()
|
||||
assert settings.level == "INFO"
|
||||
assert settings.output_dir == "./logs"
|
||||
assert settings.console.enabled is True
|
||||
assert settings.file.enabled is True
|
||||
|
||||
def test_logging_settings_custom(self):
|
||||
"""Test custom logging settings"""
|
||||
settings = LoggingSettings(
|
||||
level="DEBUG",
|
||||
console={"enabled": True, "level": "WARNING"}
|
||||
)
|
||||
assert settings.level == "DEBUG"
|
||||
assert settings.console.level == "WARNING"
|
||||
|
||||
|
||||
class TestConfig:
|
||||
"""Test Config model"""
|
||||
|
||||
def test_config_minimal(self):
|
||||
"""Test creating a minimal config"""
|
||||
config = Config(
|
||||
database_pairs=[
|
||||
DatabasePair(
|
||||
name="Test",
|
||||
enabled=True,
|
||||
baseline=DatabaseConnection(
|
||||
server="SERVER01",
|
||||
database="PROD"
|
||||
),
|
||||
target=DatabaseConnection(
|
||||
server="SERVER01",
|
||||
database="TEST"
|
||||
)
|
||||
)
|
||||
],
|
||||
comparison=ComparisonSettings(
|
||||
mode="health_check",
|
||||
row_count=RowCountSettings(enabled=True),
|
||||
schema=SchemaSettings(enabled=True),
|
||||
aggregates=AggregateSettings(enabled=False)
|
||||
),
|
||||
tables=[]
|
||||
)
|
||||
assert len(config.database_pairs) == 1
|
||||
assert config.comparison.mode == "health_check"
|
||||
assert len(config.tables) == 0
|
||||
|
||||
def test_config_with_tables(self):
|
||||
"""Test config with table definitions"""
|
||||
from drt.models.table import TableInfo
|
||||
|
||||
config = Config(
|
||||
database_pairs=[
|
||||
DatabasePair(
|
||||
name="Test",
|
||||
enabled=True,
|
||||
baseline=DatabaseConnection(server="S1", database="D1"),
|
||||
target=DatabaseConnection(server="S1", database="D2")
|
||||
)
|
||||
],
|
||||
comparison=ComparisonSettings(
|
||||
mode="health_check",
|
||||
row_count=RowCountSettings(enabled=True),
|
||||
schema=SchemaSettings(enabled=True),
|
||||
aggregates=AggregateSettings(enabled=False)
|
||||
),
|
||||
tables=[
|
||||
TableInfo(
|
||||
schema="dbo",
|
||||
name="TestTable",
|
||||
enabled=True,
|
||||
expected_in_target=True
|
||||
)
|
||||
]
|
||||
)
|
||||
assert len(config.tables) == 1
|
||||
assert config.tables[0].name == "TestTable"
|
||||
186
tests/test_models.py
Executable file
186
tests/test_models.py
Executable file
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
Unit tests for data models
|
||||
"""
|
||||
import pytest
|
||||
from drt.models.enums import Status, CheckType
|
||||
from drt.models.table import TableInfo, ColumnInfo
|
||||
from drt.models.results import CheckResult, ComparisonResult
|
||||
|
||||
|
||||
class TestStatus:
|
||||
"""Test Status enum"""
|
||||
|
||||
def test_status_values(self):
|
||||
"""Test status enum values"""
|
||||
assert Status.PASS.value == "PASS"
|
||||
assert Status.FAIL.value == "FAIL"
|
||||
assert Status.WARNING.value == "WARNING"
|
||||
assert Status.ERROR.value == "ERROR"
|
||||
assert Status.INFO.value == "INFO"
|
||||
assert Status.SKIP.value == "SKIP"
|
||||
|
||||
def test_status_severity(self):
|
||||
"""Test status severity comparison"""
|
||||
assert Status.FAIL.severity > Status.WARNING.severity
|
||||
assert Status.WARNING.severity > Status.PASS.severity
|
||||
assert Status.ERROR.severity > Status.FAIL.severity
|
||||
|
||||
|
||||
class TestCheckType:
|
||||
"""Test CheckType enum"""
|
||||
|
||||
def test_check_type_values(self):
|
||||
"""Test check type enum values"""
|
||||
assert CheckType.TABLE_EXISTENCE.value == "TABLE_EXISTENCE"
|
||||
assert CheckType.ROW_COUNT.value == "ROW_COUNT"
|
||||
assert CheckType.SCHEMA.value == "SCHEMA"
|
||||
assert CheckType.AGGREGATE.value == "AGGREGATE"
|
||||
|
||||
|
||||
class TestTableInfo:
|
||||
"""Test TableInfo model"""
|
||||
|
||||
def test_table_info_creation(self):
|
||||
"""Test creating a TableInfo instance"""
|
||||
table = TableInfo(
|
||||
schema="dbo",
|
||||
name="TestTable",
|
||||
enabled=True,
|
||||
expected_in_target=True
|
||||
)
|
||||
assert table.schema == "dbo"
|
||||
assert table.name == "TestTable"
|
||||
assert table.enabled is True
|
||||
assert table.expected_in_target is True
|
||||
assert table.aggregate_columns == []
|
||||
|
||||
def test_table_info_with_aggregates(self):
|
||||
"""Test TableInfo with aggregate columns"""
|
||||
table = TableInfo(
|
||||
schema="dbo",
|
||||
name="FactSales",
|
||||
enabled=True,
|
||||
expected_in_target=True,
|
||||
aggregate_columns=["Amount", "Quantity"]
|
||||
)
|
||||
assert len(table.aggregate_columns) == 2
|
||||
assert "Amount" in table.aggregate_columns
|
||||
|
||||
|
||||
class TestColumnInfo:
|
||||
"""Test ColumnInfo model"""
|
||||
|
||||
def test_column_info_creation(self):
|
||||
"""Test creating a ColumnInfo instance"""
|
||||
column = ColumnInfo(
|
||||
name="CustomerID",
|
||||
data_type="int",
|
||||
is_nullable=False,
|
||||
is_primary_key=True
|
||||
)
|
||||
assert column.name == "CustomerID"
|
||||
assert column.data_type == "int"
|
||||
assert column.is_nullable is False
|
||||
assert column.is_primary_key is True
|
||||
|
||||
|
||||
class TestCheckResult:
|
||||
"""Test CheckResult model"""
|
||||
|
||||
def test_check_result_pass(self):
|
||||
"""Test creating a passing check result"""
|
||||
result = CheckResult(
|
||||
check_type=CheckType.ROW_COUNT,
|
||||
status=Status.PASS,
|
||||
message="Row counts match",
|
||||
baseline_value=1000,
|
||||
target_value=1000
|
||||
)
|
||||
assert result.status == Status.PASS
|
||||
assert result.baseline_value == 1000
|
||||
assert result.target_value == 1000
|
||||
|
||||
def test_check_result_fail(self):
|
||||
"""Test creating a failing check result"""
|
||||
result = CheckResult(
|
||||
check_type=CheckType.ROW_COUNT,
|
||||
status=Status.FAIL,
|
||||
message="Row count mismatch",
|
||||
baseline_value=1000,
|
||||
target_value=950
|
||||
)
|
||||
assert result.status == Status.FAIL
|
||||
assert result.baseline_value != result.target_value
|
||||
|
||||
|
||||
class TestComparisonResult:
|
||||
"""Test ComparisonResult model"""
|
||||
|
||||
def test_comparison_result_creation(self):
|
||||
"""Test creating a ComparisonResult instance"""
|
||||
result = ComparisonResult(
|
||||
schema="dbo",
|
||||
table="TestTable"
|
||||
)
|
||||
assert result.schema == "dbo"
|
||||
assert result.table == "TestTable"
|
||||
assert len(result.checks) == 0
|
||||
|
||||
def test_add_check_result(self):
|
||||
"""Test adding check results"""
|
||||
comparison = ComparisonResult(
|
||||
schema="dbo",
|
||||
table="TestTable"
|
||||
)
|
||||
|
||||
check = CheckResult(
|
||||
check_type=CheckType.ROW_COUNT,
|
||||
status=Status.PASS,
|
||||
message="Row counts match"
|
||||
)
|
||||
|
||||
comparison.checks.append(check)
|
||||
assert len(comparison.checks) == 1
|
||||
assert comparison.checks[0].status == Status.PASS
|
||||
|
||||
def test_overall_status_all_pass(self):
|
||||
"""Test overall status when all checks pass"""
|
||||
comparison = ComparisonResult(
|
||||
schema="dbo",
|
||||
table="TestTable"
|
||||
)
|
||||
|
||||
comparison.checks.append(CheckResult(
|
||||
check_type=CheckType.TABLE_EXISTENCE,
|
||||
status=Status.PASS,
|
||||
message="Table exists"
|
||||
))
|
||||
|
||||
comparison.checks.append(CheckResult(
|
||||
check_type=CheckType.ROW_COUNT,
|
||||
status=Status.PASS,
|
||||
message="Row counts match"
|
||||
))
|
||||
|
||||
assert comparison.overall_status == Status.PASS
|
||||
|
||||
def test_overall_status_with_failure(self):
|
||||
"""Test overall status when one check fails"""
|
||||
comparison = ComparisonResult(
|
||||
schema="dbo",
|
||||
table="TestTable"
|
||||
)
|
||||
|
||||
comparison.checks.append(CheckResult(
|
||||
check_type=CheckType.TABLE_EXISTENCE,
|
||||
status=Status.PASS,
|
||||
message="Table exists"
|
||||
))
|
||||
|
||||
comparison.checks.append(CheckResult(
|
||||
check_type=CheckType.ROW_COUNT,
|
||||
status=Status.FAIL,
|
||||
message="Row count mismatch"
|
||||
))
|
||||
|
||||
assert comparison.overall_status == Status.FAIL
|
||||
83
tests/test_utils.py
Executable file
83
tests/test_utils.py
Executable file
@@ -0,0 +1,83 @@
|
||||
"""
|
||||
Unit tests for utility functions
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
from drt.utils.timestamps import format_timestamp, format_duration
|
||||
from drt.utils.patterns import matches_pattern
|
||||
|
||||
|
||||
class TestTimestamps:
|
||||
"""Test timestamp utilities"""
|
||||
|
||||
def test_format_timestamp(self):
|
||||
"""Test timestamp formatting"""
|
||||
dt = datetime(2024, 1, 15, 14, 30, 45)
|
||||
formatted = format_timestamp(dt)
|
||||
assert formatted == "20240115_143045"
|
||||
|
||||
def test_format_timestamp_current(self):
|
||||
"""Test formatting current timestamp"""
|
||||
formatted = format_timestamp()
|
||||
# Should be in YYYYMMDD_HHMMSS format
|
||||
assert len(formatted) == 15
|
||||
assert formatted[8] == "_"
|
||||
|
||||
def test_format_duration_seconds(self):
|
||||
"""Test duration formatting for seconds"""
|
||||
duration = format_duration(45.5)
|
||||
assert duration == "45.50s"
|
||||
|
||||
def test_format_duration_minutes(self):
|
||||
"""Test duration formatting for minutes"""
|
||||
duration = format_duration(125.0)
|
||||
assert duration == "2m 5.00s"
|
||||
|
||||
def test_format_duration_hours(self):
|
||||
"""Test duration formatting for hours"""
|
||||
duration = format_duration(3725.0)
|
||||
assert duration == "1h 2m 5.00s"
|
||||
|
||||
|
||||
class TestPatterns:
|
||||
"""Test pattern matching utilities"""
|
||||
|
||||
def test_exact_match(self):
|
||||
"""Test exact pattern matching"""
|
||||
assert matches_pattern("TestTable", "TestTable") is True
|
||||
assert matches_pattern("TestTable", "OtherTable") is False
|
||||
|
||||
def test_wildcard_star(self):
|
||||
"""Test wildcard * pattern"""
|
||||
assert matches_pattern("TestTable", "Test*") is True
|
||||
assert matches_pattern("TestTable", "*Table") is True
|
||||
assert matches_pattern("TestTable", "*est*") is True
|
||||
assert matches_pattern("TestTable", "Other*") is False
|
||||
|
||||
def test_wildcard_question(self):
|
||||
"""Test wildcard ? pattern"""
|
||||
assert matches_pattern("Test1", "Test?") is True
|
||||
assert matches_pattern("TestA", "Test?") is True
|
||||
assert matches_pattern("Test12", "Test?") is False
|
||||
assert matches_pattern("Test", "Test?") is False
|
||||
|
||||
def test_combined_wildcards(self):
|
||||
"""Test combined wildcard patterns"""
|
||||
assert matches_pattern("Test_Table_01", "Test_*_??") is True
|
||||
assert matches_pattern("Test_Table_1", "Test_*_??") is False
|
||||
|
||||
def test_case_sensitivity(self):
|
||||
"""Test case-sensitive matching"""
|
||||
assert matches_pattern("TestTable", "testtable") is False
|
||||
assert matches_pattern("TestTable", "TestTable") is True
|
||||
|
||||
def test_empty_pattern(self):
|
||||
"""Test empty pattern"""
|
||||
assert matches_pattern("TestTable", "") is False
|
||||
assert matches_pattern("", "") is True
|
||||
|
||||
def test_special_characters(self):
|
||||
"""Test patterns with special characters"""
|
||||
assert matches_pattern("Test.Table", "Test.Table") is True
|
||||
assert matches_pattern("Test_Table", "Test_*") is True
|
||||
assert matches_pattern("Test-Table", "Test-*") is True
|
||||
Reference in New Issue
Block a user