import pandas as pd import sys import os import tableauserverclient as TSC from datetime import datetime, timedelta, timezone from tableauhyperapi import TableDefinition, SqlType, Nullability, TableName, escape_name from dotenv import load_dotenv from pathlib import Path import pytest load_dotenv() sys.path.append(os.path.join(os.path.dirname(__file__), '../djagitit')) import tableau def safely_getenv(var): """Checks if environment variable is defined before returning it""" if not os.getenv(var): raise EnvironmentError(f'Environment variable {var} not defined!') return os.getenv(var) # Example data creation df1 = pd.DataFrame({ 'report_date': pd.to_datetime(['2022-01-01', '2022-01-02']), 'isrc': ['ABC123', 'DEF456'], 'country_code': ['US', 'UK'], 'num_streams': [100, 200], 'avg_stream_duration_seconds': [150.5, 200.75], 'is_local_track': [True, False], }) df2 = pd.DataFrame({ 'report_date': pd.to_datetime(['2022-01-03', '2022-01-04']), 'isrc': ['GHI789', 'JKL123'], 'country_code': ['FI', 'FR'], 'num_streams': [300, 400], 'avg_stream_duration_seconds': [140.5, 201.75], 'is_local_track': [False, False], }) expected_hyperschema = [ TableDefinition.Column('report_date', SqlType.date(), Nullability.NULLABLE), TableDefinition.Column('isrc', SqlType.text(), Nullability.NULLABLE), TableDefinition.Column('country_code', SqlType.text(), Nullability.NULLABLE), TableDefinition.Column('num_streams', SqlType.big_int(), Nullability.NULLABLE), TableDefinition.Column('avg_stream_duration_seconds', SqlType.double(), Nullability.NULLABLE), TableDefinition.Column('is_local_track', SqlType.bool(), Nullability.NULLABLE), ] testfilepathobject_hyper = Path(safely_getenv('DATADIR')) / 'test' / 'test.hyper' testfilepathobject_package = Path(safely_getenv('DATADIR')) / 'test' / 'packaged_data.tdsx' testfilepath_hyper = testfilepathobject_hyper.resolve() testfilepath_package = testfilepathobject_package.resolve() infamous_id = 'e6d1af34-b15d-490f-a972-cbabcb8e7cf1' class TestHyperSchema: @staticmethod def test_hyperschema(): generated_hyperschema = tableau.generate_hyper_schema(df1) generated_columns = [(col.name, col.type) for col in generated_hyperschema] expected_columns = [(col.name, col.type) for col in expected_hyperschema] assert generated_columns == expected_columns @pytest.mark.parametrize("schema, table", [ ('kikkelis', 'kokkelis'), ]) class TestHyperIndividual: @staticmethod def test_write_hyper_new(schema, table): tableau.write_hyper(data=df1, hyper_path=testfilepath_hyper, hyperSchema=expected_hyperschema, schemaName=schema, tableName=table) assert os.path.exists(testfilepath_hyper) @staticmethod def test_query_hyper(schema, table): result = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table) assert isinstance(result, pd.DataFrame) nrow1 = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, select='count(*)', scalar=True) assert nrow1 == 2 @staticmethod def test_write_hyper_existing(schema, table): tableau.write_hyper(data=df2, hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, overwrite=False) nrow2 = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, select='count(*)', scalar=True) assert nrow2 == 4 @staticmethod def test_command_hyper(schema, table): result = tableau.command_hyper(hyper_path=testfilepath_hyper, command=f"delete from {TableName(schema, table)} where {escape_name('isrc')} in ('GHI789', 'JKL123')") assert result == 2 @pytest.mark.parametrize("schema, table", [ ('kikkelis', 'kokkelis'), ('public', 'extract'), ('test_schema_1', 'test_table_1'), ('test_schema_2', 'test_table_2'), ]) class TestHyperCumulative: @staticmethod def test_hyper(schema, table): tableau.write_hyper(data=df1, hyper_path=testfilepath_hyper, hyperSchema=expected_hyperschema, schemaName=schema, tableName=table) assert os.path.exists(testfilepath_hyper) result = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table) assert isinstance(result, pd.DataFrame) nrow1 = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, select='count(*)', scalar=True) assert nrow1 == 2 tableau.write_hyper(data=df2, hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, overwrite=False) nrow2 = tableau.query_hyper(hyper_path=testfilepath_hyper, schemaName=schema, tableName=table, select='count(*)', scalar=True) assert nrow2 == 4 result = tableau.command_hyper(hyper_path=testfilepath_hyper, command=f"delete from {TableName(schema, table)} where {escape_name('isrc')} in ('GHI789', 'JKL123')") assert result == 2 class TestServer: @staticmethod def test_publish_hyper_new(): global testdatasource_id testdatasource = tableau.publish_datasource(file_path=testfilepath_hyper, site='int_mktng', project_id=infamous_id) testdatasource_id = testdatasource.id @staticmethod def test_publish_hyper_existing(): testdatasource = tableau.publish_datasource(file_path=testfilepath_hyper, site='int_mktng', datasource_id=testdatasource_id) @staticmethod def test_publish_packaged_new(): global testdatasource_id2 testdatasource2 = tableau.publish_datasource(file_path=testfilepath_hyper, site='int_mktng', package_path=testfilepath_package, project_id=infamous_id) testdatasource_id2 = testdatasource2.id @staticmethod def test_publish_packaged_existing(): testdatasource2 = tableau.publish_datasource(file_path=testfilepath_hyper, site='int_mktng', package_path=testfilepath_package, datasource_id=testdatasource_id2) @staticmethod def test_get_data_sources(): limit = datetime.now(timezone.utc) - timedelta(minutes=1) timefilter = limit.strftime('%Y-%m-%dT%H:%M:%SZ') req_option = TSC.RequestOptions(pagesize=1000) req_option.filter.add( TSC.Filter( TSC.RequestOptions.Field.UpdatedAt, TSC.RequestOptions.Operator.GreaterThan, timefilter ) ) result = tableau.get_datasource_ids('int_mktng', options=req_option) assert isinstance(result, dict), 'Failed, excpected dictionary' assert len(result) > 0, 'Failed, empty dict' print(testdatasource_id) print(result) assert testdatasource_id in result.values(), 'Failed, recently published data source not found' @staticmethod def test_get_projects(): result = tableau.get_project_ids('int_mktng') assert isinstance(result, dict), 'Failed, excpected dictionary' assert len(result) > 0, 'Failed, empty dict' assert infamous_id in result.values(), 'Failed, project Infamous not found' @staticmethod def test_delete_data_source(): tableau.delete_datasource(testdatasource_id, 'int_mktng', sure='Yes')