Coverage for tests/conftest.py: 100%

0 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-01 22:07 +0000

1# import os 

2# 

3# import peppy 

4# import pytest 

5# from sqlalchemy import create_engine, text 

6# 

7# from pepdbagent import PEPDatabaseAgent 

8# 

9# DNS = "postgresql+psycopg://postgres:docker@localhost:5432/pep-db" 

10# 

11# 

12# DATA_PATH = os.path.join( 

13# os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 

14# "tests", 

15# "data", 

16# ) 

17# 

18# 

19# def get_path_to_example_file(namespace, project_name): 

20# return os.path.join(DATA_PATH, namespace, project_name, "project_config.yaml") 

21# 

22# 

23# @pytest.fixture 

24# def list_of_available_peps(): 

25# pep_namespaces = os.listdir(DATA_PATH) 

26# projects = {} 

27# for np in pep_namespaces: 

28# pep_name = os.listdir(os.path.join(DATA_PATH, np)) 

29# projects[np] = {p: get_path_to_example_file(np, p) for p in pep_name} 

30# return projects 

31# 

32# 

33# @pytest.fixture(scope="function") 

34# def initiate_pepdb_con( 

35# list_of_available_peps, 

36# ): 

37# sa_engine = create_engine(DNS) 

38# with sa_engine.begin() as conn: 

39# conn.execute(text("DROP table IF EXISTS projects CASCADE")) 

40# conn.execute(text("DROP table IF EXISTS samples CASCADE")) 

41# conn.execute(text("DROP table IF EXISTS subsamples CASCADE")) 

42# conn.execute(text("DROP table IF EXISTS stars CASCADE")) 

43# conn.execute(text("DROP table IF EXISTS users CASCADE")) 

44# conn.execute(text("DROP table IF EXISTS views CASCADE")) 

45# conn.execute(text("DROP table IF EXISTS views_samples CASCADE")) 

46# 

47# pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=True) 

48# for namespace, item in list_of_available_peps.items(): 

49# if namespace == "private_test": 

50# private = True 

51# else: 

52# private = False 

53# for name, path in item.items(): 

54# prj = peppy.Project(path) 

55# pepdb_con.project.create( 

56# namespace=namespace, 

57# name=name, 

58# tag="default", 

59# is_private=private, 

60# project=prj, 

61# overwrite=True, 

62# pep_schema="random_schema_name", 

63# ) 

64# 

65# yield pepdb_con 

66# 

67# 

68# @pytest.fixture(scope="function") 

69# def initiate_empty_pepdb_con( 

70# list_of_available_peps, 

71# ): 

72# """ 

73# create connection without adding peps to the db 

74# """ 

75# # sa_engine = create_engine(DNS) 

76# # with sa_engine.begin() as conn: 

77# # conn.execute(text("DROP table IF EXISTS projects CASCADE")) 

78# # conn.execute(text("DROP table IF EXISTS samples CASCADE")) 

79# # conn.execute(text("DROP table IF EXISTS subsamples CASCADE")) 

80# 

81# pepdb_con = PEPDatabaseAgent(dsn=DNS, echo=False) 

82# 

83# yield pepdb_con