Repository: numpy/numpy
File: numpy/_core/tests/test_multiarray.py (Python)

Code before:
            offset_items = len(x.flat) // 4
            offset_bytes = x.dtype.itemsize * offset_items
            y = np.fromfile(
                f, dtype=x.dtype, count=count_items, offset=offset_bytes
            )
            assert_array_equal(
                y, x.flat[offset_items:offset_items + count_items]
            )

            # subsequent seeks should stack
            offset_bytes = x.dtype.itemsize
            z = np.fromfile(f, dtype=x.dtype, offset=offset_bytes)
            assert_array_equal(z, x.flat[offset_items + count_items + 1:])

        with open(tmp_filename, 'wb') as f:
            x.tofile(f, sep=",")

        with open(tmp_filename, 'rb') as f:
            assert_raises_regex(
                    TypeError,
                    "'offset' argument only permitted for binary files",
                    np.fromfile, tmp_filename, dtype=x.dtype,
                    sep=",", offset=1)

    @pytest.mark.skipif(IS_PYPY, reason="bug in PyPy's PyNumber_AsSsize_t")
    @pytest.mark.thread_unsafe(reason="os.dup thread-unsafe?")
    def test_fromfile_bad_dup(self, tmp_filename):
        def dup_str(fd):
            return 'abc'

        def dup_bigint(fd):
            return 2**68

        x = self._create_data()
        old_dup = os.dup
        try:
            with open(tmp_filename, 'wb') as f:
                x.tofile(f)
                for dup, exc in ((dup_str, TypeError), (dup_bigint, OSError)):
                    os.dup = dup
                    assert_raises(exc, np.fromfile, f)
        finally:
            os.dup = old_dup

    def _check_from(self, s, value, filename, **kw):
        if 'sep' not in kw:
            y = np.frombuffer(s, **kw)
        else:
            y = np.fromstring(s, **kw)
        assert_array_equal(y, value)


Review comment (suggestion):
this test is thread-unsafe because of monkeypatching, not because of os.dup

Code after:
                    pos = f.tell()
                assert_equal(pos, 10, err_msg=err_msg)

    def test_file_position_after_tofile(self, tmp_path, param_filename):
        # gh-4118
        sizes = [io.DEFAULT_BUFFER_SIZE // 8,
                 io.DEFAULT_BUFFER_SIZE,
                 io.DEFAULT_BUFFER_SIZE * 8]
        tmp_filename = normalize_filename(tmp_path, param_filename)

        for size in sizes:
            err_msg = "%d" % (size,)

            with open(tmp_filename, 'wb') as f:
                f.seek(size - 1)
                f.write(b'\0')
                f.seek(10)
                f.write(b'12')
                np.array([0], dtype=np.float64).tofile(f)
                pos = f.tell()
            assert_equal(pos, 10 + 2 + 8, err_msg=err_msg)

            with open(tmp_filename, 'r+b') as f:
                f.read(2)
                f.seek(0, 1)  # seek between read&write required by ANSI C
                np.array([0], dtype=np.float64).tofile(f)
                pos = f.tell()
            assert_equal(pos, 10, err_msg=err_msg)

    def test_load_object_array_fromfile(self, tmp_path, param_filename):
        # gh-12300
        tmp_filename = normalize_filename(tmp_path, param_filename)
        with open(tmp_filename, 'w') as f:
            # Ensure we have a file with consistent contents
            pass

        with open(tmp_filename, 'rb') as f:
            assert_raises_regex(ValueError, "Cannot read into object array",
                                np.fromfile, f, dtype=object)

        assert_raises_regex(ValueError, "Cannot read into object array",
                            np.fromfile, tmp_filename, dtype=object)

    def test_fromfile_offset(self, tmp_path, param_filename):
        tmp_filename = normalize_filename(tmp_path, param_filename)
        x = self._create_data()
        with open(tmp_filename, 'wb') as f:
            x.tofile(f)

        with open(tmp_filename, 'rb') as f:
            y = np.fromfile(f, dtype=x.dtype, offset=0)