Change `Status.error_message()` to use `.decode(..., errors='replace')`.

To avoid clobbering the error to be reported with a `UnicodeDecodeError`.

PiperOrigin-RevId: 474941891

pybind11_abseil/register_status_bindings.cc

@@ -94,6 +94,17 @@ void def_status_factory(
       arg("message"));
 }
 
+// TODO(b/225205409): Move to utility library.
+// To avoid clobbering potentially critical error messages with
+// `UnicodeDecodeError`.
+str decode_utf8_replace(absl::string_view s) {
+  PyObject* u = PyUnicode_DecodeUTF8(s.data(), s.size(), "replace");
+  if (u == nullptr) {
+    throw error_already_set();
+  }
+  return reinterpret_steal<str>(u);
+}
+
 }  // namespace
 
 namespace internal {
@@ -123,26 +134,43 @@ void RegisterStatusBindings(module m) {
       .def(init<absl::StatusCode, std::string>())
       .def("ok", &absl::Status::ok)
       .def("code", &absl::Status::code)
-      .def("code_int", [](const absl::Status& self) {
-          return static_cast<int>(self.code());
-      })
-      .def("message", &absl::Status::message)
-      .def(
-          "update",
-          (void (absl::Status::*)(const absl::Status &)) & absl::Status::Update,
-          arg("other"))
-      .def("to_string", [](const absl::Status& s) { return s.ToString(); })
-      .def("__repr__", [](const absl::Status& s) { return s.ToString(); })
+      .def("code_int",
+           [](const absl::Status& self) {
+             return static_cast<int>(self.code());
+           })
+      .def("message",
+           [](const absl::Status& self) {
+             return decode_utf8_replace(self.message());
+           })
+      .def("message_bytes",
+           [](const absl::Status& self) {
+             return bytes(self.message().data(), self.message().size());
+           })
+      .def("update",
+           (void(absl::Status::*)(const absl::Status&)) & absl::Status::Update,
+           arg("other"))
+      .def("to_string",
+           [](const absl::Status& s) {
+             return decode_utf8_replace(s.ToString());
+           })
+      .def("__repr__",
+           [](const absl::Status& s) {
+             return decode_utf8_replace(s.ToString());
+           })
       .def("to_string_status_not_ok",
            [](const absl::Status& s) {
-             return s.ToString();
+             return decode_utf8_replace(s.ToString());
            })
       .def_static("OkStatus", DoNotThrowStatus(&absl::OkStatus))
       .def("raw_code", &absl::Status::raw_code)
-      .def("CanonicalCode", [](const absl::Status& self) {
-          return static_cast<int>(self.code());
-      })
-      .def("error_message", &absl::Status::message)
+      .def("CanonicalCode",
+           [](const absl::Status& self) {
+             return static_cast<int>(self.code());
+           })
+      .def("error_message",
+           [](const absl::Status& self) {
+             return decode_utf8_replace(self.message());
+           })
       .def("IgnoreError", &absl::Status::IgnoreError);
 
   m.def("is_ok", &IsOk, arg("status_or"),

pybind11_abseil/tests/status_test.py

@@ -166,6 +166,18 @@ class StatusTest(absltest.TestCase):
     self.assertEqual(ok_status.error_message(), '')
     self.assertIsNone(ok_status.IgnoreError())
 
+  def test_error_message_malformed_utf8(self):
+    malformed_utf8 = b'\x80'
+    stx80 = status.invalid_argument_error(malformed_utf8)
+    self.assertEqual(stx80.message(), '�')
+    self.assertEqual(stx80.message_bytes(), malformed_utf8)
+    self.assertEqual(stx80.error_message(), '�')
+    self.assertEqual(stx80.to_string(), 'INVALID_ARGUMENT: �')
+    self.assertEqual(str(stx80), 'INVALID_ARGUMENT: �')
+    self.assertEqual(repr(stx80), 'INVALID_ARGUMENT: �')
+    e = status.StatusNotOk(stx80)
+    self.assertEqual(str(e), 'INVALID_ARGUMENT: �')
+
   def test_raw_code_ne_code(self):
     st500 = status_example.status_from_int_code(500, 'Not a canonical code.')
     self.assertEqual(st500.raw_code(), 500)