Coverage for tests/test_serialization_basic_info.py: 38%

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

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

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

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11from __future__ import annotations 

12 

13import json 

14import os 

15import tempfile 

16import unittest 

17 

18import numpy as np 

19import pydantic 

20 

21from lsst.images import Box, Image 

22from lsst.images import fits as images_fits 

23from lsst.images import json as images_json 

24from lsst.images.fits import FitsInputArchive 

25from lsst.images.json import JsonInputArchive 

26from lsst.images.serialization import ArchiveInfo, InputArchive 

27 

28try: 

29 import h5py 

30 

31 from lsst.images import ndf as images_ndf 

32 from lsst.images.ndf import NdfInputArchive 

33 

34 HAVE_H5PY = True 

35except ImportError: 

36 HAVE_H5PY = False 

37 

38 

39class ArchiveInfoTestCase(unittest.TestCase): 

40 """Tests for the ArchiveInfo model and its schema_url parsing.""" 

41 

42 def test_from_schema_url(self) -> None: 

43 info = ArchiveInfo.from_schema_url( 

44 "https://images.lsst.io/schemas/visit_image-1.2.3", format_version=1 

45 ) 

46 self.assertEqual(info.schema_name, "visit_image") 

47 self.assertEqual(info.schema_version, "1.2.3") 

48 self.assertEqual(info.schema_url, "https://images.lsst.io/schemas/visit_image-1.2.3") 

49 self.assertEqual(info.format_version, 1) 

50 

51 def test_from_schema_url_none_format(self) -> None: 

52 info = ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/image-1.0.0", format_version=None) 

53 self.assertEqual(info.schema_name, "image") 

54 self.assertIsNone(info.format_version) 

55 

56 def test_frozen(self) -> None: 

57 info = ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/image-1.0.0", format_version=None) 

58 with self.assertRaises(pydantic.ValidationError): 

59 info.schema_name = "other" # type: ignore[misc] 

60 

61 def test_from_schema_url_invalid(self) -> None: 

62 with self.assertRaises(ValueError): 

63 ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/noversion", format_version=None) 

64 

65 def test_from_schema_url_foreign_host(self) -> None: 

66 # A DATAMODL header pointing at another host must be rejected so a 

67 # file not written by lsst.images cannot select an arbitrary schema. 

68 with self.assertRaises(ValueError): 

69 ArchiveInfo.from_schema_url("https://evil.example.com/schemas/image-1.0.0", format_version=None) 

70 

71 def test_get_basic_info_base_raises(self) -> None: 

72 with self.assertRaises(NotImplementedError): 

73 InputArchive.get_basic_info("x.fits") 

74 

75 

76class JsonBasicInfoTestCase(unittest.TestCase): 

77 """get_basic_info for the JSON backend.""" 

78 

79 def setUp(self) -> None: 

80 self.tmp = tempfile.mkdtemp() 

81 self.image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

82 

83 def test_json_basic_info(self) -> None: 

84 path = os.path.join(self.tmp, "x.json") 

85 images_json.write(self.image, path) 

86 info = JsonInputArchive.get_basic_info(path) 

87 self.assertEqual(info.schema_name, "image") 

88 self.assertEqual(info.schema_version, "1.0.0") 

89 self.assertEqual(info.schema_url, "https://images.lsst.io/schemas/image-1.0.0") 

90 self.assertIsNone(info.format_version) 

91 

92 

93class FitsBasicInfoTestCase(unittest.TestCase): 

94 """get_basic_info for the FITS backend.""" 

95 

96 def setUp(self) -> None: 

97 tmp = tempfile.TemporaryDirectory() 

98 self.addCleanup(tmp.cleanup) 

99 self.tmp = tmp.name 

100 self.image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

101 

102 def test_fits_basic_info(self) -> None: 

103 path = os.path.join(self.tmp, "x.fits") 

104 images_fits.write(self.image, path) 

105 info = FitsInputArchive.get_basic_info(path) 

106 self.assertEqual(info.schema_name, "image") 

107 self.assertEqual(info.schema_version, "1.0.0") 

108 self.assertEqual(info.schema_url, "https://images.lsst.io/schemas/image-1.0.0") 

109 self.assertEqual(info.format_version, 1) 

110 

111 

112@unittest.skipUnless(HAVE_H5PY, "h5py is not available.") 

113class NdfBasicInfoTestCase(unittest.TestCase): 

114 """get_basic_info for the NDF backend.""" 

115 

116 def setUp(self) -> None: 

117 tmp = tempfile.TemporaryDirectory() 

118 self.addCleanup(tmp.cleanup) 

119 self.tmp = tmp.name 

120 self.image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

121 

122 def test_ndf_basic_info(self) -> None: 

123 path = os.path.join(self.tmp, "x.sdf") 

124 images_ndf.write(self.image, path) 

125 info = NdfInputArchive.get_basic_info(path) 

126 self.assertEqual(info.schema_name, "image") 

127 self.assertEqual(info.schema_version, "1.0.0") 

128 self.assertEqual(info.schema_url, "https://images.lsst.io/schemas/image-1.0.0") 

129 self.assertEqual(info.format_version, 1) 

130 

131 def test_ndf_basic_info_ignores_nested_json(self) -> None: 

132 path = os.path.join(self.tmp, "nested.sdf") 

133 images_ndf.write(self.image, path) 

134 # Inject a nested pointer-tree JSON alongside the top-level one; it 

135 # sorts before "JSON" so a depth-first scan would hit it first. 

136 payload = json.dumps({"schema_url": "https://images.lsst.io/schemas/aaa_child-9.9.9"}) 

137 with h5py.File(path, "a") as handle: 

138 child = handle.require_group("MORE/LSST/AAA") 

139 child.create_dataset("JSON", data=np.frombuffer(payload.encode("utf-8"), dtype=np.uint8)) 

140 info = NdfInputArchive.get_basic_info(path) 

141 self.assertEqual(info.schema_name, "image") 

142 self.assertEqual(info.schema_url, "https://images.lsst.io/schemas/image-1.0.0")