import datetime import sys from typing import Any, Union import path import pandas as pd import pytest from _pytest.config.argparsing import Parser from pytest import FixtureRequest import streamlit as st # directory reach directory = path.Path(__file__).abspath() # setting path sys.path.append(directory.parent.parent) from snowflake.snowpark.session import Session from common.local_connection import get_connection_parameters def pytest_addoption(parser: Parser)-> None: parser.addoption("--snowflake-session", action="store", default="live") @pytest.fixture(scope='module') def session(request: FixtureRequest) -> Union[Session, Any]: """ Pytest fixture to provide either live or dummy Snowflake connection for testing :param request: :return: Snowflake session builder config """ if request.config.getoption('--snowflake-session') == 'local': return Session.builder.config('local_testing', True).create() else: connection_params = get_connection_parameters("fan_data_export") return Session.builder.configs(connection_params).create() @pytest.fixture def products()-> pd.DataFrame: """ Pytest fixture to be reused by test functions :return: Pandas dataframe """ PRODUCT_ID = ["A", "B", "C", "D"] PRODUCT_NAME = ["Jeans", "Red Vinyl", "Blue Vinyl", "Double CD"] TYPE = ["Clothing", "Vinyl", "Vinyl", "CD"] products = {'PRODUCT_ID': PRODUCT_ID, 'PRODUCT_NAME': PRODUCT_NAME, 'TYPE': TYPE} df = pd.DataFrame(products) return df @pytest.fixture def fan_ids_for_export()-> pd.DataFrame: """ Pytest fixture to be reused by test functions :return: Pandas dataframe """ CRM_FAN_ID = ["a0O0800000vsL81EAE", "ba4a6a2e6b35fdc74726600b72ed094d4a12f932e80c9daf962c4b8c8cb097f3"] fan_ids = {'FAN_ID': CRM_FAN_ID} df = pd.DataFrame(fan_ids) return df @pytest.fixture def fans()-> pd.DataFrame: """ Pytest fixture to be reused by test functions :return: Pandas dataframe """ CRM_FAN_ID = ["a0O0800000vsL81EAE", "ba4a6a2e6b35fdc74726600b72ed094d4a12f932e80c9daf962c4b8c8cb097f3", "dsfdsfd"] EMAILS = ["example1@email.com", "example2@email.com", "example3@gmail.com"] PHONES = [None, "372 44222", None] fans = {'ID': CRM_FAN_ID, 'EMAIL_C': EMAILS, 'MOBILE_PHONE_C': PHONES} df = pd.DataFrame(fans) return df @pytest.fixture def fans2()-> pd.DataFrame: """ Pytest fixture to be reused by test functions :return: Pandas dataframe """ CRM_FAN_ID = ["a0O0800000vsL81EAE", "ba4a6a2e6b35fdc74726600b72ed094d4a12f932e80c9daf962c4b8c8cb097f3", "dsfdsfd"] EMAILS = ["example1@email.com", "example2@email.com", "example3@gmail.com"] PHONES = [None, "372 44222", None] fans = {'ID': CRM_FAN_ID, 'EMAIL': EMAILS, 'MOBILE_PHONE_C': PHONES} df = pd.DataFrame(fans) return df @pytest.fixture(autouse=True) def reset_streamlit_session_state()-> None: """ Pytest fixture to be reused by test functions """ st.session_state.clear() @pytest.fixture def forms()-> pd.DataFrame: FORM_ID = ["form_1", "form_2"] FORM_NAME_C = ["Artist A FORM 1", "Artist A FORM 2"] TLA_ID_C = ["tla_1", "tla_1"] FORM_MIGRATION_ID = [1123, 2323] last_modified_date = datetime.date(2023, 4, 25) FORM_LAST_MODIFIED_DATE = [last_modified_date, last_modified_date] forms = {'ID': FORM_ID, 'FORM_NAME_C': FORM_NAME_C, 'TLA_ID_C': TLA_ID_C, 'MIGRATION_ID_C': FORM_MIGRATION_ID, 'LAST_MODIFIED_DATE': FORM_LAST_MODIFIED_DATE} df = pd.DataFrame(forms) return df @pytest.fixture def mailing_lists()-> pd.DataFrame: MAILING_LIST_ID = ["ml_1", "ml_2", "ml_3"] MAILING_LIST_NAME_C = ["mailing list nr 1", "mailing list nr 2", "mailing list nr 3"] TLA_ID_C = ["tla_1", "tla_1", "tla_3"] mailing_lists = {'ID': MAILING_LIST_ID, 'MAILING_LIST_NAME_C': MAILING_LIST_NAME_C, 'TLA_ID_C': TLA_ID_C} df = pd.DataFrame(mailing_lists) return df @pytest.fixture def tlas()-> pd.DataFrame: TLA_ID = ["tla_1"] TL_ID_C = ["tla_1"] LABEL_ID_C = ["label_1"] ARTIST_ID_C = ["artist_1"] TERRITORY_ID_C = ["territory_1"] tlas = {'ID': TLA_ID, 'TL_ID_C': TL_ID_C, 'LABEL_ID_C': LABEL_ID_C, 'ARTIST_ID_C': ARTIST_ID_C, 'TERRITORY_ID_C': TERRITORY_ID_C} df = pd.DataFrame(tlas) return df @pytest.fixture def tls()-> pd.DataFrame: TL_ID_C = ["tla_1"] TERRITORY_ID_C = ["territory_1"] tls = {'ID': TL_ID_C, 'TERRITORY_ID_C': TERRITORY_ID_C} df = pd.DataFrame(tls) return df @pytest.fixture def territories()-> pd.DataFrame: ID = ["territory_1"] TERRITORY_C = ["GB"] territories = {'ID': ID, 'TERRITORY_C': TERRITORY_C} df = pd.DataFrame(territories) return df @pytest.fixture def labels()-> pd.DataFrame: LABEL_NAME = ["label one"] LABEL_ID = ["label_1"] labels = {'NAME': LABEL_NAME, 'ID': LABEL_ID} df = pd.DataFrame(labels) return df @pytest.fixture def artists()-> pd.DataFrame: ARTIST_ID = ["artist_1"] ARTIST_C = ["Artist one"] artists = {'ID': ARTIST_ID, 'ARTIST_C': ARTIST_C} df = pd.DataFrame(artists) return df