# -*- coding: utf-8 -*-

from datafactory.pipeline import Pipeline
from datafactory.operators import ReadXLS
from datafactory.operators import WriteFiles
from datafactory.operators import TransposeTable
from datafactory.operators import ReadGspread

TEST_DIR=”tests/test_data/”

def test_pipeline(put_file, reconnect_gspread) -> None:
“””Teste de criação de um pipeline.”””
RECONNECTION_TIMEOUT = 60

# Pipeline test
pipeline_config = {
“stages”: [
{
“name”: “reading_stage”,
“operator”: “ReadXLS”,
“config”: {
“sheet_name”: “data”,
“src_cache”: TEST_DIR + “test_data_{src_cache}.xlsx”,
},
},
{
“name”: “transposing_stage”,
“operator”: “TransposeTable”,
“config”: {},
},
{
“name”: “writing_stage”,
“operator”: “WriteFiles”,
“config”: {
“output_file”: TEST_DIR + “{data_name}.csv”,
}
},
],
}
pipeline = Pipeline(**pipeline_config)

# Adicionando ReadXLS no pipeline
pipeline.add_op(ReadXLS, config={“sheet_name”: “data”,
“src_cache”: TEST_DIR + “test_data_{src_cache}.xlsx”})
# Adicionando TransposeTable
pipeline.add_op(TransposeTable)
# Adicionando WriteFiles
pipeline.add_op(WriteFiles,
config={“output_file”: TEST_DIR + “{data_name}.csv”})

# Executando pipeline
pipeline.run(**{“src_cache”: put_file},
reconnect_gspread=reconnect_gspread,
timeout=RECONNECTION_TIMEOUT)

# Verificando se o pipeline aconteceu corretamente
assert pipeline.status() == “SUCCESS”, “Erro ao execução pipeline”

from tempfile import NamedTemporaryFile
from datafactory.src import gspread

def test_gspread_api(reinstall_gspread_client: bool) -> None:
“””Testa a comunicação com o Google Drive”””

# Input
GSPREAD_CONFIG = {
“file_id”: None,
}
GSHEET_CONFIG = {
“name”: “API_TESTS_TMP_SPREADSHEET”, # Sheet Name
“header_row”: 0,
}

# Se reinstala o client do Gspread (util quando existem problemas de conexão)
if reinstall_gspread_client:
gspread.reinstall_client()

# Cria o spreadsheet temporal
spreadsheet = gspread.GspreadHelper.create_spreadsheet(
GSPREAD_CONFIG[“file_id”], GSPREAD_CONFIG
)
# Marca este Open Spreadsheet
os_marked = gspread.GspreadHelper.open_spreadsheet(
GSPREAD_CONFIG[“file_id”], GSPREAD_CONFIG
)
# Adiciona worksheet
worksheet = gspread.GspreadHelper.add_worksheet(
GSHEET_CONFIG[“name”], os_marked, GSHEET_CONFIG
)

# Input Data a serem escritos
with NamedTemporaryFile(“w”, suffix=”.csv”) as write_file:
write_file.file.write(“lnA,lnB,lnCnA1,B1,C1nA2,B2,C2”)
write_file.file.seek(0)
# Escreve dentro do worksheet
gspread.GspreadHelper.write_worksheet_from_csv(
write_file.file.name, worksheet, write_file
)
rw_marked = gspread.GspreadHelper.read_worksheet(
os_marked, GSHEET_CONFIG[“name”], GSHEET_CONFIG, rows_limit=None
)
assert rw_marked[0] == [“lnA”, “lnB”, “lnC”], “Valor da lina esperado não foi encontrado”
assert rw_marked[1:] == [[“A1”, “B1”, “C1”], [“A2”, “B2”, “C2”]],
“Valores esperados da tabela não foram encontrados”