numpy · seberg · Mar 9, 2025 · Feb 13, 2025 · Mar 1, 2025 · Mar 1, 2025
diff --git a/numpy/lib/_format_impl.py b/numpy/lib/_format_impl.py
@@ -872,6 +872,14 @@ def read_array(fp, allow_pickle=False, pickle_kwargs=None, *,
                    data = _read_bytes(fp, read_size, "array data")
                    array[i:i + read_count] = numpy.frombuffer(data, dtype=dtype,
                                                             count=read_count)
+
+        if array.size != count:
+            raise ValueError(
+                "Failed to read all data for array. "
+                f"Expected {shape} = {count} elements, "
+                f"could only read {array.size} elements. "
+                "(file seems not fully written?)"
+            )

        if fortran_order:
            array.shape = shape[::-1]

diff --git a/numpy/lib/tests/test_format.py b/numpy/lib/tests/test_format.py
@@ -428,7 +428,6 @@ def roundtrip_truncated(arr):
    arr2 = format.read_array(f2)
    return arr2

-
 def assert_equal_(o1, o2):
    assert_(o1 == o2)

@@ -451,6 +450,30 @@ def test_roundtrip_truncated():
        if arr.dtype != object:
            assert_raises(ValueError, roundtrip_truncated, arr)

+def test_file_truncated(tmp_path):
+    path = tmp_path / "a.npy"
+    for arr in basic_arrays:
+        if arr.dtype != object:
+            with open(path, 'wb') as f:
+                format.write_array(f, arr)
+            #truncate the file by one byte
+            with open(path, 'rb+') as f:
+                f.seek(-1, os.SEEK_END)
+                f.truncate()
+            with open(path, 'rb') as f:
+                with pytest.raises(
+                    ValueError, 
+                    match = (
+                        r"EOF: reading array header, "
+                        r"expected (\d+) bytes got (\d+)"
+                    ) if arr.size == 0 else (
+                        r"Failed to read all data for array\. "
+                        r"Expected \(.*?\) = (\d+) elements, "
+                        r"could only read (\d+) elements\. "
+                        r"\(file seems not fully written\?\)"
+                    )
+                ):
+                    _ = format.read_array(f)

 def test_long_str():
    # check items larger than internal buffer size, gh-4027