Coverage for tests/test_universes.py: 100%

1import pytest 

2 

3from bbconf.exceptions import BEDFileNotFoundError 

4 

5from .conftest import SERVICE_UNAVAILABLE 

6from .utils import BED_TEST_ID, ContextManagerDBTesting 

7 

8 

9@pytest.mark.skipif(SERVICE_UNAVAILABLE, reason="Database is not available") 

10class TestUniverses: 

11 def test_add(self, bbagent_obj): 

12 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): 

13 bbagent_obj.bed.add_universe( 

14 BED_TEST_ID, bedset_id=None, construct_method="hp31" 

15 ) 

16 

17 assert bbagent_obj.bed.exists(BED_TEST_ID) 

18 assert bbagent_obj.bed.exists_universe(BED_TEST_ID) 

19 universe_meta = bbagent_obj.bed.get(BED_TEST_ID) 

20 assert universe_meta is not None 

21 assert universe_meta.is_universe is True 

22 

23 def test_delete_universe(self, bbagent_obj): 

24 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): 

25 bbagent_obj.bed.add_universe( 

26 BED_TEST_ID, bedset_id=None, construct_method="hp31" 

27 ) 

28 

29 assert bbagent_obj.bed.exists_universe(BED_TEST_ID) 

30 

31 bbagent_obj.bed.delete_universe(BED_TEST_ID) 

32 

33 assert not bbagent_obj.bed.exists_universe(BED_TEST_ID) 

34 

35 def test_add_universe_error(self, bbagent_obj): 

36 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): 

37 with pytest.raises(BEDFileNotFoundError): 

38 bbagent_obj.bed.add_universe( 

39 "not_f", bedset_id=None, construct_method="hp31" 

40 ) 

41 

42 def test_add_get_tokenized(self, bbagent_obj, mocker): 

43 with ContextManagerDBTesting(config=bbagent_obj.config, add_data=True): 

44 bbagent_obj.bed.add_universe( 

45 BED_TEST_ID, bedset_id=None, construct_method="hp31" 

46 ) 

47 

48 assert bbagent_obj.bed.exists_universe(BED_TEST_ID) 

49 

50 saved_path = "test/1/1" 

51 

52 zarr_mock = mocker.patch( 

53 "bbconf.modules.bedfiles.BedAgentBedFile._add_zarr_s3", 

54 return_value=saved_path, 

55 ) 

56 

57 bbagent_obj.bed.add_tokenized( 

58 bed_id=BED_TEST_ID, universe_id=BED_TEST_ID, token_vector=[1, 2, 3] 

59 ) 

60 

61 zarr_path = bbagent_obj.bed._get_tokenized_path( 

62 BED_TEST_ID, universe_id=BED_TEST_ID 

63 ) 

64 

65 assert zarr_mock.called 

66 assert f"s3://bedbase/{saved_path}" == zarr_path 

67 

68 def test_get_tokenized(self, bbagent_obj): 

69 # how to test it? 

70 ...