1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
|
import threading
import asdf_standard.integration
import pytest
import asdf
from asdf import get_config, resource
from asdf.extension import BuiltinExtension, ExtensionProxy
from asdf.resource import ResourceMappingProxy
def test_config_context():
assert get_config().validate_on_read is True
with asdf.config_context() as config:
config.validate_on_read = False
assert get_config().validate_on_read is False
assert get_config().validate_on_read is True
def test_config_context_nested():
assert get_config().validate_on_read is True
with asdf.config_context() as config1:
config1.validate_on_read = False
with asdf.config_context() as config2:
config2.validate_on_read = True
with asdf.config_context() as config3:
config3.validate_on_read = False
assert get_config().validate_on_read is False
assert get_config().validate_on_read is True
def test_config_context_threaded():
assert get_config().validate_on_read is True
thread_value = None
def worker():
nonlocal thread_value
thread_value = get_config().validate_on_read
with asdf.config_context() as config:
config.validate_on_read = False
with asdf.config_context() as config:
config.validate_on_read = False
thread = threading.Thread(target=worker)
thread.start()
thread.join()
assert thread_value is True
assert get_config().validate_on_read is True
def test_global_config():
assert get_config().validate_on_read is True
get_config().validate_on_read = False
assert get_config().validate_on_read is False
with asdf.config_context() as config:
assert config.validate_on_read is False
config.validate_on_read = True
assert get_config().validate_on_read is True
assert get_config().validate_on_read is False
# Global config is reset to defaults by autouse
# fixture in asdf/tests/conftest.py.
def test_validate_on_read():
with asdf.config_context() as config:
assert config.validate_on_read == asdf.config.DEFAULT_VALIDATE_ON_READ
config.validate_on_read = False
assert get_config().validate_on_read is False
config.validate_on_read = True
assert get_config().validate_on_read is True
def test_default_version():
with asdf.config_context() as config:
assert config.default_version == asdf.config.DEFAULT_DEFAULT_VERSION
assert "1.2.0" != asdf.config.DEFAULT_DEFAULT_VERSION
config.default_version = "1.2.0"
assert config.default_version == "1.2.0"
with pytest.raises(ValueError):
config.default_version = "0.1.5"
def test_legacy_fill_schema_defaults():
with asdf.config_context() as config:
assert config.legacy_fill_schema_defaults == asdf.config.DEFAULT_LEGACY_FILL_SCHEMA_DEFAULTS
config.legacy_fill_schema_defaults = False
assert get_config().legacy_fill_schema_defaults is False
config.legacy_fill_schema_defaults = True
assert get_config().legacy_fill_schema_defaults is True
def test_array_inline_threshold():
with asdf.config_context() as config:
assert config.array_inline_threshold == asdf.config.DEFAULT_ARRAY_INLINE_THRESHOLD
config.array_inline_threshold = 10
assert get_config().array_inline_threshold == 10
config.array_inline_threshold = None
assert get_config().array_inline_threshold is None
def test_resource_mappings():
with asdf.config_context() as config:
core_mappings = resource.get_json_schema_resource_mappings() + asdf_standard.integration.get_resource_mappings()
default_mappings = config.resource_mappings
assert len(default_mappings) >= len(core_mappings)
new_mapping = {"http://somewhere.org/schemas/foo-1.0.0": b"foo"}
config.add_resource_mapping(new_mapping)
assert len(config.resource_mappings) == len(default_mappings) + 1
assert any(m for m in config.resource_mappings if m.delegate is new_mapping)
# Adding a mapping should be idempotent:
config.add_resource_mapping(new_mapping)
# ... even if wrapped:
config.add_resource_mapping(ResourceMappingProxy(new_mapping))
assert len(config.resource_mappings) == len(default_mappings) + 1
# Adding a mapping should place it at the front of the line:
front_mapping = {"http://somewhere.org/schemas/baz-1.0.0": b"baz"}
config.add_resource_mapping(front_mapping)
assert len(config.resource_mappings) == len(default_mappings) + 2
assert config.resource_mappings[0].delegate is front_mapping
# ... even if the mapping is already in the list:
config.add_resource_mapping(new_mapping)
assert len(config.resource_mappings) == len(default_mappings) + 2
assert config.resource_mappings[0].delegate is new_mapping
# Reset should get rid of any additions:
config.reset_resources()
assert len(config.resource_mappings) == len(default_mappings)
# Should be able to remove a mapping:
config.add_resource_mapping(new_mapping)
config.remove_resource_mapping(new_mapping)
assert len(config.resource_mappings) == len(default_mappings)
# ... even if wrapped:
config.add_resource_mapping(new_mapping)
config.remove_resource_mapping(ResourceMappingProxy(new_mapping))
assert len(config.resource_mappings) == len(default_mappings)
# ... and also by the name of the package the mappings came from:
config.add_resource_mapping(ResourceMappingProxy(new_mapping, package_name="foo"))
config.add_resource_mapping(
ResourceMappingProxy({"http://somewhere.org/schemas/bar-1.0.0": b"bar"}, package_name="foo")
)
config.remove_resource_mapping(package="foo")
assert len(config.resource_mappings) == len(default_mappings)
# Can combine the package and mapping filters when removing:
config.add_resource_mapping(ResourceMappingProxy(new_mapping, package_name="foo"))
config.remove_resource_mapping(new_mapping, package="foo")
assert len(config.resource_mappings) == len(default_mappings)
# But not omit both:
with pytest.raises(ValueError):
config.remove_resource_mapping()
# Removing a mapping should be idempotent:
config.add_resource_mapping(new_mapping)
config.remove_resource_mapping(new_mapping)
config.remove_resource_mapping(new_mapping)
assert len(config.resource_mappings) == len(default_mappings)
def test_resource_manager():
with asdf.config_context() as config:
# Initial resource manager should contain just the entry points resources:
assert "http://stsci.edu/schemas/asdf/core/asdf-1.1.0" in config.resource_manager
assert (
b"http://stsci.edu/schemas/asdf/core/asdf-1.1.0"
in config.resource_manager["http://stsci.edu/schemas/asdf/core/asdf-1.1.0"]
)
assert "http://somewhere.org/schemas/foo-1.0.0" not in config.resource_manager
# Add a mapping and confirm that the manager now contains it:
new_mapping = {"http://somewhere.org/schemas/foo-1.0.0": b"foo"}
config.add_resource_mapping(new_mapping)
assert "http://stsci.edu/schemas/asdf/core/asdf-1.1.0" in config.resource_manager
assert (
b"http://stsci.edu/schemas/asdf/core/asdf-1.1.0"
in config.resource_manager["http://stsci.edu/schemas/asdf/core/asdf-1.1.0"]
)
assert "http://somewhere.org/schemas/foo-1.0.0" in config.resource_manager
assert config.resource_manager["http://somewhere.org/schemas/foo-1.0.0"] == b"foo"
# Remove a mapping and confirm that the manager no longer contains it:
config.remove_resource_mapping(new_mapping)
assert "http://stsci.edu/schemas/asdf/core/asdf-1.1.0" in config.resource_manager
assert (
b"http://stsci.edu/schemas/asdf/core/asdf-1.1.0"
in config.resource_manager["http://stsci.edu/schemas/asdf/core/asdf-1.1.0"]
)
assert "http://somewhere.org/schemas/foo-1.0.0" not in config.resource_manager
# Reset and confirm that the manager no longer contains the custom mapping:
config.add_resource_mapping(new_mapping)
config.reset_resources()
assert "http://stsci.edu/schemas/asdf/core/asdf-1.1.0" in config.resource_manager
assert (
b"http://stsci.edu/schemas/asdf/core/asdf-1.1.0"
in config.resource_manager["http://stsci.edu/schemas/asdf/core/asdf-1.1.0"]
)
assert "http://somewhere.org/schemas/foo-1.0.0" not in config.resource_manager
def test_extensions():
with asdf.config_context() as config:
original_extensions = config.extensions
assert any(isinstance(e.delegate, BuiltinExtension) for e in original_extensions)
class FooExtension:
types = []
tag_mapping = []
url_mapping = []
new_extension = FooExtension()
class BarExtension:
extension_uri = "asdf://somewhere.org/extensions/bar-1.0"
types = []
tag_mapping = []
url_mapping = []
uri_extension = BarExtension()
# Add an extension:
config.add_extension(new_extension)
assert len(config.extensions) == len(original_extensions) + 1
assert any(e for e in config.extensions if e.delegate is new_extension)
# Adding an extension should be idempotent:
config.add_extension(new_extension)
assert len(config.extensions) == len(original_extensions) + 1
# Even when wrapped:
config.add_extension(ExtensionProxy(new_extension))
assert len(config.extensions) == len(original_extensions) + 1
# Remove an extension:
config.remove_extension(new_extension)
assert len(config.extensions) == len(original_extensions)
# Removing should work when wrapped:
config.add_extension(new_extension)
config.remove_extension(ExtensionProxy(new_extension))
assert len(config.extensions) == len(original_extensions)
# And also by URI:
config.add_extension(uri_extension)
config.remove_extension(uri_extension.extension_uri)
assert len(config.extensions) == len(original_extensions)
# And also by URI pattern:
config.add_extension(uri_extension)
config.remove_extension("asdf://somewhere.org/extensions/*")
assert len(config.extensions) == len(original_extensions)
# Remove by the name of the extension's package:
config.add_extension(ExtensionProxy(new_extension, package_name="foo"))
config.add_extension(ExtensionProxy(uri_extension, package_name="foo"))
config.remove_extension(package="foo")
assert len(config.extensions) == len(original_extensions)
# Can combine remove filters:
config.add_extension(ExtensionProxy(new_extension, package_name="foo"))
config.add_extension(ExtensionProxy(uri_extension, package_name="foo"))
config.remove_extension(uri_extension.extension_uri, package="foo")
assert len(config.extensions) == len(original_extensions) + 1
# ... but not omit both:
with pytest.raises(ValueError):
config.remove_extension()
# Removing an extension should be idempotent:
config.add_extension(new_extension)
config.remove_extension(new_extension)
config.remove_extension(new_extension)
assert len(config.extensions) == len(original_extensions)
# Resetting should get rid of any additions:
config.add_extension(new_extension)
config.add_extension(FooExtension())
config.reset_extensions()
assert len(config.extensions) == len(original_extensions)
def test_config_repr():
with asdf.config_context() as config:
config.validate_on_read = True
config.default_version = "1.5.0"
config.io_block_size = 9999
config.legacy_fill_schema_defaults = False
config.array_inline_threshold = 14
assert "validate_on_read: True" in repr(config)
assert "default_version: 1.5.0" in repr(config)
assert "io_block_size: 9999" in repr(config)
assert "legacy_fill_schema_defaults: False" in repr(config)
assert "array_inline_threshold: 14" in repr(config)
|
