1+ import sys
2+ import os
3+ import pytest
4+ import json
5+ import yaml
6+
7+ sys .path .insert (0 , os .path .abspath (os .path .join (os .path .dirname (__file__ ), '..' )))
8+
9+ from validation_rules_yaml import validate_entry , load_validation_rules
10+
11+ # @pytest.fixture(scope='module')
12+ # def expected_results():
13+ # return load_expected_results()
14+ @pytest .fixture (scope = 'module' )
15+ def expected_results_negative ():
16+ return load_expected_results_negative ()
17+
18+ @pytest .fixture (scope = 'module' )
19+ def validation_rules ():
20+ return load_validation_rules ()
21+
22+ def fetch_results_as_dict (results ):
23+ return [dict (row ) for row in results ]
24+
25+ # def load_expected_results():
26+ # json_path = os.path.join(os.path.dirname(__file__), '..', 'expected_results/expected_results.json')
27+ # with open(json_path) as f:
28+ # return json.load(f)
29+ def load_expected_results_negative ():
30+ json_path = os .path .join (os .path .dirname (__file__ ), '..' , 'expected_results/expected_results_negative.json' )
31+ with open (json_path ) as f :
32+ return json .load (f )
33+
34+ def load_sql_query (file_name ):
35+ sql_path = os .path .join (os .path .dirname (__file__ ), '..' , 'sql_queries' , file_name )
36+ with open (sql_path , 'r' ) as file :
37+ return file .read ()
38+
39+ # # Positive Testing
40+ # def run_query_test(db_connection, expected_results, validation_rules, query_name, sql_file):
41+ # query_sql = load_sql_query(sql_file)
42+ # results = db_connection.execute(query_sql)
43+ # results_dict = fetch_results_as_dict(results)
44+
45+ # # Obtener las reglas específicas de la consulta
46+ # query_rules = validation_rules["queries"].get(query_name, {}).get("rules", {})
47+
48+ # # Combinar las reglas comunes con las reglas específicas de la consulta
49+ # combined_rules = {**validation_rules["common_rules"], **query_rules}
50+
51+ # # Filtrar resultados válidos
52+ # filtered_results = [result for result in results_dict if validate_entry(result, combined_rules)]
53+
54+ # # Convert lists to sets for comparison:
55+ # """
56+ # En lugar de hacer:
57+ # # Comparar con los resultados esperados: expected = expected_results[query_name]["expected"]
58+ # Mejor hacer:
59+ # Ignorar el orden de las filas y se enfoque en el contenido de los datos.
60+ # Convertir Diccionarios en Tuplas Ordenadas: tuple(sorted(d.items())) convierte cada diccionario
61+ # en una lista de pares clave-valor ordenados, luego en una tupla. Esto asegura que
62+ # cada diccionario tenga una representación única e independiente del orden de los campos.
63+ # Crear Conjuntos: set(...) convierte las listas de tuplas en conjuntos.
64+ # Esto permite que la comparación ignore el orden de las filas.
65+ # Comparación de Conjuntos: La comparación de los conjuntos filtered y expected asegura que ambos
66+ # contienen exactamente los mismos elementos, independientemente del orden en que aparecen.
67+ # """
68+ # expected = set(tuple(sorted(d.items())) for d in expected_results[query_name]["expected"])
69+ # filtered = set(tuple(sorted(d.items())) for d in filtered_results)
70+
71+ # assert filtered == expected, f"Error in {query_name}: {filtered} != {expected}"
72+
73+ # def test_active_sites_positive(db_connection, expected_results, validation_rules):
74+ # run_query_test(db_connection, expected_results, validation_rules, "active_sites_positive", "active_sites_positive.sql")
75+
76+ # def test_randomized_patients_positive(db_connection, expected_results, validation_rules):
77+ # run_query_test(db_connection, expected_results, validation_rules, "randomized_patients_positive", "randomized_patients_positive.sql")
78+
79+ # def test_active_patients_date(db_connection, expected_results, validation_rules):
80+ # run_query_test(db_connection, expected_results, validation_rules, "active_patients_date", "active_patients_date.sql")
81+
82+ # Negative Testing
83+ def run_query_test_negative (db_connection_negative , expected_results_negative , validation_rules , query_name , sql_file ):
84+ query_sql = load_sql_query (sql_file )
85+ results = db_connection_negative .execute (query_sql )
86+ results_dict = fetch_results_as_dict (results )
87+
88+ # Obtener las reglas específicas de la consulta
89+ query_rules = validation_rules ["queries" ].get (query_name , {}).get ("rules" , {})
90+
91+ # Combinar las reglas comunes con las reglas específicas de la consulta
92+ combined_rules = {** validation_rules ["common_rules" ], ** query_rules }
93+
94+ # Filtrar resultados válidos
95+ filtered_results = [result for result in results_dict if validate_entry (result , combined_rules )]
96+
97+ # Convert lists to sets for comparison:
98+ """
99+ En lugar de hacer:
100+ # Comparar con los resultados esperados: expected = expected_results[query_name]["expected"]
101+ Mejor hacer:
102+ Ignorar el orden de las filas y se enfoque en el contenido de los datos.
103+ Convertir Diccionarios en Tuplas Ordenadas: tuple(sorted(d.items())) convierte cada diccionario
104+ en una lista de pares clave-valor ordenados, luego en una tupla. Esto asegura que
105+ cada diccionario tenga una representación única e independiente del orden de los campos.
106+ Crear Conjuntos: set(...) convierte las listas de tuplas en conjuntos.
107+ Esto permite que la comparación ignore el orden de las filas.
108+ Comparación de Conjuntos: La comparación de los conjuntos filtered y expected asegura que ambos
109+ contienen exactamente los mismos elementos, independientemente del orden en que aparecen.
110+ """
111+ expected = set (tuple (sorted (d .items ())) for d in expected_results_negative [query_name ]["expected" ])
112+ filtered = set (tuple (sorted (d .items ())) for d in filtered_results )
113+
114+ assert filtered == expected , f"Error in { query_name } { filtered } { expected }
115+ def test_active_patients_date_negative (db_connection_negative , expected_results_negative , validation_rules ):
116+ run_query_test_negative (db_connection_negative , expected_results_negative , validation_rules , "active_patients_date_negative" , "active_patients_date.sql" )
0 commit comments