diff --git a/validation/test_validate.py b/validation/test_validate.py index bf1b2c2b..170e1bb2 100644 --- a/validation/test_validate.py +++ b/validation/test_validate.py @@ -23,6 +23,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os import subprocess import sys import tempfile @@ -34,6 +35,20 @@ from validate import UniqueKeyLoader +# chmod(0o000) makes a file unreadable only under POSIX permissions, and not +# for root, which bypasses them. Elsewhere (notably Windows) chmod only toggles +# the read-only flag and the file stays readable. +_MODE_BLOCKS_READS = hasattr(os, "geteuid") and os.geteuid() != 0 + +_MINIMAL_MODEL = ( + "version: 0.2.0.dev0\n" + "name: sales\n" + "datasets:\n" + " - name: orders\n" + " source: analytics.orders\n" +) + + class UniqueKeyLoaderTest(unittest.TestCase): def load(self, content: str): return yaml.load(content, Loader=UniqueKeyLoader) @@ -352,6 +367,106 @@ def test_duplicate_key_exits_nonzero(self): self.assertIn("Error: Invalid YAML", result.stdout) self.assertIn("found duplicate key 'name'", result.stdout) + def run_validator_args(self, *args) -> subprocess.CompletedProcess[str]: + """Run validate.py with arbitrary arguments, for bad-path cases.""" + return subprocess.run( + [sys.executable, Path(__file__).with_name("validate.py"), *args], + check=False, + capture_output=True, + text=True, + ) + + def test_directory_as_input_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + result = self.run_validator_args(temp_dir) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Could not read", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + @unittest.skipUnless( + _MODE_BLOCKS_READS, + "chmod(0o000) only blocks reads for a non-root POSIX user", + ) + def test_unreadable_input_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_text("version: 0.2.0.dev0\n") + model_path.chmod(0o000) + result = self.run_validator_args(model_path) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Could not read", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_directory_as_schema_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_text(_MINIMAL_MODEL) + result = self.run_validator_args(model_path, "--schema", temp_dir) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Could not read schema", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_schema_that_is_not_json_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_text(_MINIMAL_MODEL) + schema_path = Path(temp_dir) / "schema.json" + schema_path.write_text("not json {{{\n") + result = self.run_validator_args(model_path, "--schema", schema_path) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Invalid JSON in schema", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + @unittest.skipUnless( + _MODE_BLOCKS_READS, + "chmod(0o000) only blocks reads for a non-root POSIX user", + ) + def test_unreadable_schema_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_text(_MINIMAL_MODEL) + schema_path = Path(temp_dir) / "schema.json" + schema_path.write_text("{}") + schema_path.chmod(0o000) + result = self.run_validator_args(model_path, "--schema", schema_path) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Could not read schema", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_binary_input_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_bytes(b"\xff\xfe\x00binary") + result = self.run_validator_args(model_path) + + self.assertEqual(result.returncode, 1) + self.assertIn("is not valid UTF-8 text", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_binary_schema_is_reported(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "model.yaml" + model_path.write_text(_MINIMAL_MODEL) + schema_path = Path(temp_dir) / "schema.json" + schema_path.write_bytes(b"\xff\xfe\x00binary") + result = self.run_validator_args(model_path, "--schema", schema_path) + + self.assertEqual(result.returncode, 1) + self.assertIn("is not valid UTF-8 text", result.stdout) + self.assertNotIn("Traceback", result.stderr) + + def test_missing_input_is_still_reported_as_not_found(self): + """The existing guard keeps its own message, rather than the read error.""" + result = self.run_validator_args("no-such-file.yaml") + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: File not found", result.stdout) + def test_valid_model_still_passes(self): result = self.run_validator( "version: 0.2.0.dev0\n" @@ -453,6 +568,36 @@ def test_empty_relationship_endpoint_is_rejected_by_schema(self): self.assertIn("Validation FAILED", result.stdout) self.assertIn("[Schema] relationships -> 0 -> to: '' should be non-empty", result.stdout) + def test_yaml_diagnostics_name_the_file_and_show_the_line(self): + # A parser mark is only actionable if it names the model: a str source + # is otherwise reported as "". Parsing a str also gets + # the offending line quoted under the mark, which a stream source does + # not record, so assert the caret wherever PyYAML produces one. + cases = ( + ("name: x\n bad: [\n", "mapping values are not allowed here", True), + ( + "version: 0.2.0.dev0\nname: sales\nname: finance\n", + "found duplicate key 'name'", + True, + ), + # Raised while the loader is constructed, before it can be named, + # and carrying a position rather than a mark with a snippet. + ("version: 0.2.0.dev0\nname: sa\x00les\n", "special characters are not allowed", False), + ) + for content, detail, has_snippet in cases: + with self.subTest(detail=detail): + result = self.run_validator(content) + + self.assertEqual(result.returncode, 1) + self.assertIn("Error: Invalid YAML", result.stdout) + self.assertIn(detail, result.stdout) + self.assertIn('in "', result.stdout) + self.assertIn("model.yaml", result.stdout) + self.assertNotIn("", result.stdout) + self.assertNotIn("Traceback", result.stderr) + if has_snippet: + self.assertIn("^", result.stdout) + if __name__ == "__main__": unittest.main() diff --git a/validation/validate.py b/validation/validate.py index 1899e094..9fb63b16 100644 --- a/validation/validate.py +++ b/validation/validate.py @@ -157,6 +157,29 @@ def _check_unique_keys(self, node: yaml.Node, visited: set) -> None: self._check_unique_keys(child, visited) +def load_yaml_named(text: str, name: str): + """Parse text with UniqueKeyLoader, naming the source in any parser mark. + + yaml.load() names a str source "", which drops the path from + malformed-YAML and duplicate-key diagnostics. Naming the loader restores it + and keeps the offending line, which PyYAML only records for a str source: + Reader.get_mark() passes its buffer to the Mark only when the input was not + a stream, so a named StringIO would name the file but lose the snippet. + """ + try: + loader = UniqueKeyLoader(text) + except yaml.reader.ReaderError as e: + # Reader.__init__ rejects an unprintable character (a NUL decodes as + # valid UTF-8 and gets this far) before the name below can be set. + e.name = name + raise + loader.name = name + try: + return loader.get_single_data() + finally: + loader.dispose() + + # Ossie schemas reference one another by raw GitHub URL or canonical $id. # Resolve those URLs onto files in this checkout only when a reference needs # them. The decorator parses and caches each retrieved schema across calls. @@ -559,6 +582,18 @@ def inherited_relationships(concept: str) -> set[str]: return errors +def read_text_or_exit(path: Path, description: str) -> str: + """Read path as UTF-8 text, or report why it could not be read and exit 1.""" + try: + return path.read_text(encoding="utf-8") + except OSError as e: + print(f"Error: Could not read {description}: {e}") + sys.exit(1) + except UnicodeDecodeError as e: + print(f"Error: {description} is not valid UTF-8 text: {e}") + sys.exit(1) + + def main(): if len(sys.argv) < 2: print(__doc__) @@ -583,20 +618,26 @@ def main(): print(f"Error: Schema not found: {schema_path}") sys.exit(1) - # Load files - with open(schema_path) as f: - schema = json.load(f) + # Read first, then parse. The exists() checks above pass for a path that cannot be + # read as text — a directory, one without read permission, or a binary file — so + # reading is guarded to report like every other bad input rather than raise. + schema_text = read_text_or_exit(schema_path, f"schema {schema_path}") + try: + schema = json.loads(schema_text) + except json.JSONDecodeError as e: + print(f"Error: Invalid JSON in schema {schema_path}: {e}") + sys.exit(1) - with open(yaml_path) as f: - try: - data = yaml.load(f, Loader=UniqueKeyLoader) - except yaml.YAMLError as e: - print(f"Error: Invalid YAML: {e}") - sys.exit(1) - except RecursionError: - # Deeply nested input surfaces as RecursionError, not YAMLError. - print("Error: Invalid YAML: input is too deeply nested to parse") - sys.exit(1) + model_text = read_text_or_exit(yaml_path, str(yaml_path)) + try: + data = load_yaml_named(model_text, str(yaml_path)) + except yaml.YAMLError as e: + print(f"Error: Invalid YAML: {e}") + sys.exit(1) + except RecursionError: + # Deeply nested input surfaces as RecursionError, not YAMLError. + print("Error: Invalid YAML: input is too deeply nested to parse") + sys.exit(1) # Run validations errors = []