Coverage for python/lsst/images/cli/_fixtures.py: 94%

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 

13__all__ = ("fixtures",) 

14 

15import json 

16from pathlib import Path 

17 

18import click 

19import pydantic 

20 

21from ..serialization import ArchiveReadError 

22from ..tests import ( 

23 SchemaFixtureError, 

24 check_schema_fixtures, 

25 format_coverage_report, 

26 freeze_schema_fixtures, 

27 refresh_schema_fixtures, 

28 schema_coverage, 

29) 

30 

31# read_fixture_tree (called internally by both refresh_schema_fixtures and 

32# freeze_schema_fixtures) can raise either of these for a fixture that no 

33# longer validates, in addition to SchemaFixtureError for a conflict these 

34# commands detect themselves; all three get the same clean-exception 

35# treatment rather than a raw traceback. 

36_FIXTURE_ERRORS = (SchemaFixtureError, ArchiveReadError, pydantic.ValidationError) 

37 

38_DIR_OPTION = click.option( 

39 "--dir", 

40 "directory", 

41 type=click.Path(file_okay=False, path_type=Path), 

42 default=Path("tests/data/schemas"), 

43 show_default=True, 

44 help="Directory holding the fixture tree.", 

45) 

46 

47_PACKAGE_OPTION = click.option( 

48 "--package", 

49 default="lsst.images", 

50 show_default=True, 

51 help="Act only on schemas whose model classes are defined in this package; " 

52 "external packages providing schemas through the 'lsst.images.schemas' " 

53 "entry point group can use this to manage their own fixtures.", 

54) 

55 

56 

57@click.group(name="fixtures") 

58def fixtures() -> None: 

59 """Manage the committed schema fixtures.""" 

60 

61 

62@fixtures.command(name="check") 

63@_DIR_OPTION 

64@_PACKAGE_OPTION 

65@click.option( 

66 "--schema-dir", 

67 "schema_directory", 

68 type=click.Path(file_okay=False, path_type=Path), 

69 required=True, 

70 help="Directory holding the frozen schema documents. Fixtures and schema " 

71 "documents are checked for pairing in both directions.", 

72) 

73@click.option( 

74 "--exempt", 

75 multiple=True, 

76 help="Schema name allowed to have no fixture; repeatable.", 

77) 

78def check( 

79 directory: Path, 

80 package: str, 

81 schema_directory: Path, 

82 exempt: tuple[str, ...], 

83) -> None: 

84 """Exit nonzero if any fixture is missing, stale, or misplaced.""" 

85 problems = check_schema_fixtures( 

86 directory, schema_directory=schema_directory, package=package, exempt=exempt 

87 ) 

88 for problem in problems: 

89 click.echo(problem) 

90 if problems: 

91 raise click.ClickException("the fixture tree does not match the current models") 

92 

93 

94@fixtures.command(name="refresh") 

95@_DIR_OPTION 

96@_PACKAGE_OPTION 

97def refresh(directory: Path, package: str) -> None: 

98 """Rewrite every development fixture in canonical form. 

99 

100 Seeds a missing fixture from the newest existing version of that schema, 

101 and regenerates the canonical twin of a development as_shipped fixture. 

102 Refuses to rewrite a fixture at a finalized version; bump SCHEMA_VERSION 

103 instead. 

104 """ 

105 try: 

106 changed = refresh_schema_fixtures(directory, package=package) 

107 except _FIXTURE_ERRORS as exc: 

108 raise click.ClickException(str(exc)) from exc 

109 for path in changed: 

110 click.echo(f"wrote {path}") 

111 if not changed: 

112 click.echo("all fixtures are already up to date") 

113 

114 

115@fixtures.command(name="freeze") 

116@_DIR_OPTION 

117@_PACKAGE_OPTION 

118def freeze(directory: Path, package: str) -> None: 

119 """Move each ordinary development fixture to its finalized version. 

120 

121 Run this after dropping the .devN suffix from SCHEMA_VERSION and running 

122 'lsst-images-admin schemas write'. Files are written and deleted; staging 

123 that in version control is left to you. Development ``as_shipped`` 

124 fixtures are refused because their preserved bytes cannot be re-stamped. 

125 """ 

126 try: 

127 frozen = freeze_schema_fixtures(directory, package=package) 

128 except _FIXTURE_ERRORS as exc: 

129 raise click.ClickException(str(exc)) from exc 

130 for written, removed in frozen: 130 ↛ 131line 130 didn't jump to line 131 because the loop on line 130 never started

131 click.echo(f"wrote {written}") 

132 click.echo(f"removed {removed}") 

133 if not frozen: 133 ↛ 136line 133 didn't jump to line 136 because the condition on line 133 was always true

134 click.echo("nothing to freeze") 

135 else: 

136 click.echo("now run 'lsst-images-admin schemas write' if you have not already") 

137 

138 

139@fixtures.command(name="coverage") 

140@_DIR_OPTION 

141@_PACKAGE_OPTION 

142@click.option("--schema", default=None, help="Report only this schema name.") 

143@click.option( 

144 "--format", 

145 "output_format", 

146 type=click.Choice(["text", "json"]), 

147 default="text", 

148 show_default=True, 

149 help="Output format.", 

150) 

151def coverage( 

152 directory: Path, 

153 package: str, 

154 schema: str | None, 

155 output_format: str, 

156) -> None: 

157 """Report what the committed fixtures do and do not exercise. 

158 

159 Coverage is credited to the schema that owns each stamped subtree, so a 

160 sub-model embedded by several containers is credited by all of them. A 

161 'gap' line marks a sub-schema position some candidate never reaches; that 

162 is information, not a failure, since a composite model need not hold every 

163 candidate its schema admits. This command always exits zero. 

164 """ 

165 report = schema_coverage(directory, package=package) 

166 if output_format == "json": 

167 click.echo( 

168 json.dumps( 

169 { 

170 f"{name} {version}": { 

171 "sources": sorted(entry.sources), 

172 "expressed": sorted(entry.expressed), 

173 "absent": sorted(entry.absent_roots), 

174 "positions": [ 

175 { 

176 "path": position.path, 

177 "candidates": sorted(position.candidates), 

178 "reached": sorted(position.reached), 

179 "missing": sorted(position.missing), 

180 } 

181 for position in entry.positions 

182 ], 

183 } 

184 for (name, version), entry in sorted(report.schemas.items()) 

185 if schema is None or name == schema 

186 }, 

187 indent=2, 

188 ) 

189 ) 

190 return 

191 if text := format_coverage_report(report, schema=schema): 

192 click.echo(text) 

193 else: 

194 click.echo(f"no schema matches {schema!r}" if schema else "no schemas in scope")