test_domain_schema.py
python
| 1 | """Tests for Phase 2 domain schema declaration and plugin registry lookup. |
| 2 | |
| 3 | Verifies that: |
| 4 | - ``MusicPlugin.schema()`` returns a fully-typed ``DomainSchema``. |
| 5 | - The four dimensions have the correct element schema types. |
| 6 | - The schema is JSON round-trippable (all values are JSON-serialisable). |
| 7 | - ``schema_for()`` in the plugin registry performs the correct lookup. |
| 8 | - The protocol assertion still holds after adding ``schema()``. |
| 9 | """ |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import json |
| 13 | |
| 14 | import pytest |
| 15 | |
| 16 | from muse.core.schema import ( |
| 17 | DomainSchema, |
| 18 | SequenceSchema, |
| 19 | SetSchema, |
| 20 | TensorSchema, |
| 21 | TreeSchema, |
| 22 | ) |
| 23 | from muse.domain import MuseDomainPlugin |
| 24 | from muse.plugins.music.plugin import MusicPlugin |
| 25 | from muse.plugins.registry import registered_domains, schema_for |
| 26 | |
| 27 | |
| 28 | # --------------------------------------------------------------------------- |
| 29 | # Helpers |
| 30 | # --------------------------------------------------------------------------- |
| 31 | |
| 32 | |
| 33 | @pytest.fixture() |
| 34 | def music_plugin() -> MusicPlugin: |
| 35 | return MusicPlugin() |
| 36 | |
| 37 | |
| 38 | @pytest.fixture() |
| 39 | def music_schema(music_plugin: MusicPlugin) -> DomainSchema: |
| 40 | return music_plugin.schema() |
| 41 | |
| 42 | |
| 43 | # =========================================================================== |
| 44 | # MusicPlugin.schema() structure |
| 45 | # =========================================================================== |
| 46 | |
| 47 | |
| 48 | class TestMusicPluginSchema: |
| 49 | def test_schema_returns_domain_schema(self, music_schema: DomainSchema) -> None: |
| 50 | assert isinstance(music_schema, dict) |
| 51 | assert music_schema["domain"] == "music" |
| 52 | |
| 53 | def test_schema_version_is_1(self, music_schema: DomainSchema) -> None: |
| 54 | assert music_schema["schema_version"] == 1 |
| 55 | |
| 56 | def test_schema_has_four_dimensions(self, music_schema: DomainSchema) -> None: |
| 57 | assert len(music_schema["dimensions"]) == 4 |
| 58 | |
| 59 | def test_dimension_names(self, music_schema: DomainSchema) -> None: |
| 60 | names = [d["name"] for d in music_schema["dimensions"]] |
| 61 | assert names == ["melodic", "harmonic", "dynamic", "structural"] |
| 62 | |
| 63 | def test_top_level_is_set_schema(self, music_schema: DomainSchema) -> None: |
| 64 | top = music_schema["top_level"] |
| 65 | assert top["kind"] == "set" |
| 66 | assert isinstance(top, dict) |
| 67 | |
| 68 | def test_top_level_set_schema_fields(self, music_schema: DomainSchema) -> None: |
| 69 | top = music_schema["top_level"] |
| 70 | assert top["kind"] == "set" |
| 71 | # Narrow to SetSchema for field access |
| 72 | if top["kind"] == "set": |
| 73 | assert top["element_type"] == "audio_file" |
| 74 | assert top["identity"] == "by_content" |
| 75 | |
| 76 | def test_melodic_dimension_is_sequence(self, music_schema: DomainSchema) -> None: |
| 77 | melodic = next(d for d in music_schema["dimensions"] if d["name"] == "melodic") |
| 78 | schema = melodic["schema"] |
| 79 | assert schema["kind"] == "sequence" |
| 80 | |
| 81 | def test_melodic_dimension_element_type(self, music_schema: DomainSchema) -> None: |
| 82 | melodic = next(d for d in music_schema["dimensions"] if d["name"] == "melodic") |
| 83 | schema = melodic["schema"] |
| 84 | if schema["kind"] == "sequence": |
| 85 | assert schema["element_type"] == "note_event" |
| 86 | assert schema["diff_algorithm"] == "lcs" |
| 87 | |
| 88 | def test_harmonic_dimension_is_sequence(self, music_schema: DomainSchema) -> None: |
| 89 | harmonic = next(d for d in music_schema["dimensions"] if d["name"] == "harmonic") |
| 90 | schema = harmonic["schema"] |
| 91 | assert schema["kind"] == "sequence" |
| 92 | |
| 93 | def test_dynamic_dimension_is_tensor(self, music_schema: DomainSchema) -> None: |
| 94 | dynamic = next(d for d in music_schema["dimensions"] if d["name"] == "dynamic") |
| 95 | schema = dynamic["schema"] |
| 96 | assert schema["kind"] == "tensor" |
| 97 | |
| 98 | def test_dynamic_tensor_schema_fields(self, music_schema: DomainSchema) -> None: |
| 99 | dynamic = next(d for d in music_schema["dimensions"] if d["name"] == "dynamic") |
| 100 | schema = dynamic["schema"] |
| 101 | if schema["kind"] == "tensor": |
| 102 | assert schema["dtype"] == "float32" |
| 103 | assert schema["rank"] == 1 |
| 104 | assert schema["epsilon"] == 1.0 |
| 105 | assert schema["diff_mode"] == "sparse" |
| 106 | |
| 107 | def test_structural_dimension_is_tree(self, music_schema: DomainSchema) -> None: |
| 108 | structural = next(d for d in music_schema["dimensions"] if d["name"] == "structural") |
| 109 | schema = structural["schema"] |
| 110 | assert schema["kind"] == "tree" |
| 111 | |
| 112 | def test_structural_tree_schema_fields(self, music_schema: DomainSchema) -> None: |
| 113 | structural = next(d for d in music_schema["dimensions"] if d["name"] == "structural") |
| 114 | schema = structural["schema"] |
| 115 | if schema["kind"] == "tree": |
| 116 | assert schema["node_type"] == "track_node" |
| 117 | assert schema["diff_algorithm"] == "zhang_shasha" |
| 118 | |
| 119 | def test_melodic_independent_merge_is_true(self, music_schema: DomainSchema) -> None: |
| 120 | melodic = next(d for d in music_schema["dimensions"] if d["name"] == "melodic") |
| 121 | assert melodic["independent_merge"] is True |
| 122 | |
| 123 | def test_structural_independent_merge_is_false(self, music_schema: DomainSchema) -> None: |
| 124 | structural = next(d for d in music_schema["dimensions"] if d["name"] == "structural") |
| 125 | assert structural["independent_merge"] is False |
| 126 | |
| 127 | def test_merge_mode_is_three_way(self, music_schema: DomainSchema) -> None: |
| 128 | assert music_schema["merge_mode"] == "three_way" |
| 129 | |
| 130 | def test_schema_round_trips_json(self, music_schema: DomainSchema) -> None: |
| 131 | serialised = json.dumps(music_schema) |
| 132 | restored = json.loads(serialised) |
| 133 | assert restored["domain"] == music_schema["domain"] |
| 134 | assert restored["schema_version"] == music_schema["schema_version"] |
| 135 | assert len(restored["dimensions"]) == len(music_schema["dimensions"]) |
| 136 | # top_level round-trips |
| 137 | assert restored["top_level"]["kind"] == music_schema["top_level"]["kind"] |
| 138 | |
| 139 | def test_schema_description_is_non_empty(self, music_schema: DomainSchema) -> None: |
| 140 | assert isinstance(music_schema["description"], str) |
| 141 | assert len(music_schema["description"]) > 0 |
| 142 | |
| 143 | def test_all_dimension_schemas_have_kind(self, music_schema: DomainSchema) -> None: |
| 144 | for dim in music_schema["dimensions"]: |
| 145 | assert "kind" in dim["schema"] |
| 146 | |
| 147 | |
| 148 | # =========================================================================== |
| 149 | # Plugin registry schema lookup |
| 150 | # =========================================================================== |
| 151 | |
| 152 | |
| 153 | class TestPluginRegistrySchemaLookup: |
| 154 | def test_schema_for_music_returns_domain_schema(self) -> None: |
| 155 | result = schema_for("music") |
| 156 | assert result is not None |
| 157 | assert result["domain"] == "music" |
| 158 | |
| 159 | def test_schema_for_unknown_domain_returns_none(self) -> None: |
| 160 | result = schema_for("nonexistent_domain_xyz") |
| 161 | assert result is None |
| 162 | |
| 163 | def test_schema_for_returns_same_type_as_plugin_schema(self) -> None: |
| 164 | plugin = MusicPlugin() |
| 165 | direct = plugin.schema() |
| 166 | via_registry = schema_for("music") |
| 167 | assert via_registry is not None |
| 168 | assert via_registry["domain"] == direct["domain"] |
| 169 | assert via_registry["schema_version"] == direct["schema_version"] |
| 170 | |
| 171 | def test_registered_domains_still_contains_music(self) -> None: |
| 172 | assert "music" in registered_domains() |
| 173 | |
| 174 | def test_schema_for_all_registered_domains_returns_non_none(self) -> None: |
| 175 | for domain in registered_domains(): |
| 176 | result = schema_for(domain) |
| 177 | assert result is not None, f"schema_for({domain!r}) returned None" |
| 178 | |
| 179 | |
| 180 | # =========================================================================== |
| 181 | # Protocol conformance |
| 182 | # =========================================================================== |
| 183 | |
| 184 | |
| 185 | class TestProtocolConformance: |
| 186 | def test_music_plugin_satisfies_protocol(self) -> None: |
| 187 | plugin = MusicPlugin() |
| 188 | assert isinstance(plugin, MuseDomainPlugin) |
| 189 | |
| 190 | def test_schema_method_is_callable(self) -> None: |
| 191 | plugin = MusicPlugin() |
| 192 | assert callable(plugin.schema) |
| 193 | |
| 194 | def test_schema_returns_dict(self) -> None: |
| 195 | plugin = MusicPlugin() |
| 196 | result = plugin.schema() |
| 197 | assert isinstance(result, dict) |