fix(validation): report unreadable input or schema, not a traceback

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 "<unicode string>". 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("<unicode string>", result.stdout)
+                self.assertNotIn("Traceback", result.stderr)
+                if has_snippet:
+                    self.assertIn("^", result.stdout)
+
 
 if __name__ == "__main__":
     unittest.main()

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 "<unicode string>", 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 = []