Coverage for tests / test_clustering.py: 36%

1# This file is part of drp_pipe. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22from __future__ import annotations 

23 

24import os 

25import unittest 

26 

27from lsst.ctrl.bps import BpsConfig, check_clustering_config 

28from lsst.pipe.base import Pipeline 

29 

30# Pipelines to check the clustering YAMLs 

31# In everyday practice, pipelines and clustering configs do not map 1:1. 

32# But for testing purposes, choose a pipeline that the clustering is 

33# expected to work with. Keys are clustering yamls. Values are pipelines. 

34CLUSTERING_PIPELINE_MAPPING = { 

35 'LSSTCam/DRP-clustering.yaml': 'LSSTCam/DRP.yaml', 

36 'HSC/DRP-RC2-clustering.yaml': 'HSC/DRP-RC2.yaml', 

37 'LSSTCam-imSim/DRP-DC2-clustering.yaml': 'LSSTCam-imSim/DRP-test-med-1.yaml', 

38 'LSSTCam-imSim/DRP-OR5-clustering.yaml': 'LSSTCam-imSim/DRP.yaml', 

39} 

40 

41PIPELINES_DIR = os.path.join(os.path.dirname(__file__), "..", "pipelines") 

42CLUSTERING_DIR = os.path.join(os.path.dirname(__file__), "..", "bps", "clustering") 

43 

44 

45class ClusteringTestCase(unittest.TestCase): 

46 

47 def test_clusters(self) -> None: 

48 """Check clustering yamls. 

49 

50 Confirm that they do not produce cycles and that the 

51 two required keys, 'pipetasks' and 'dimensions', are 

52 are present in each cluster definition. 

53 """ 

54 for clustering_path, pipeline_path in CLUSTERING_PIPELINE_MAPPING.items(): 

55 # construct pipeline graph from URI 

56 full_pipeline_path = os.path.join(PIPELINES_DIR, pipeline_path) 

57 pipeline = Pipeline.from_uri(full_pipeline_path) 

58 pipeline_graph = pipeline.to_graph() 

59 

60 # construct bps config from clustering yaml 

61 full_clustering_path = os.path.join(CLUSTERING_DIR, clustering_path) 

62 bps_config = BpsConfig(full_clustering_path) 

63 

64 try: 

65 check_clustering_config(bps_config["cluster"], pipeline_graph.make_task_xgraph()) 

66 except Exception as e: 

67 self.fail(f"Clustering config in {clustering_path} is not compatible with pipeline" 

68 f" {pipeline_path}: {e}") 

69 

70 for cluster_name, cluster_info in bps_config["cluster"].items(): 

71 for required_key in ["pipetasks", "dimensions"]: 

72 self.assertIn(required_key, cluster_info, 

73 f"Missing '{required_key}' in {clustering_path} for {cluster_name}") 

74 

75 

76if __name__ == "__main__": 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 unittest.main()