Add __richcmp__ to cygrpc.Timespec

2 files changed, 31 insertions, 2 deletions

diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
index d052b3f8bc..a9163c75bf 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi
@@ -194,6 +194,25 @@ cdef class Timespec:
   def infinite_past():
     return Timespec(float("-inf"))
 
+  def __richcmp__(Timespec self not None, Timespec other not None, int op):
+    cdef gpr_timespec self_c_time = self.c_time
+    cdef gpr_timespec other_c_time = other.c_time
+    cdef int result = gpr_time_cmp(self_c_time, other_c_time)
+    if op == 0:  # <
+      return result < 0
+    elif op == 2:  # ==
+      return result == 0
+    elif op == 4:  # >
+      return result > 0
+    elif op == 1:  # <=
+      return result <= 0
+    elif op == 3:  # !=
+      return result != 0
+    elif op == 5:  # >=
+      return result >= 0
+    else:
+      raise ValueError('__richcmp__ `op` contract violated')
+
 
 cdef class CallDetails:
 
diff --git a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
index 7aec316b95..a4d5015aa4 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
@@ -95,8 +95,18 @@ class TypeSmokeTest(unittest.TestCase):
 
     def testTimespec(self):
         now = time.time()
-        timespec = cygrpc.Timespec(now)
-        self.assertAlmostEqual(now, float(timespec), places=8)
+        now_timespec_a = cygrpc.Timespec(now)
+        now_timespec_b = cygrpc.Timespec(now)
+        self.assertAlmostEqual(now, float(now_timespec_a), places=8)
+        self.assertEqual(now_timespec_a, now_timespec_b)
+        self.assertLess(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
+        self.assertGreater(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
+        self.assertGreaterEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
+        self.assertGreaterEqual(cygrpc.Timespec(now), cygrpc.Timespec(now))
+        self.assertLessEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
+        self.assertLessEqual(cygrpc.Timespec(now), cygrpc.Timespec(now))
+        self.assertNotEqual(cygrpc.Timespec(now - 1), cygrpc.Timespec(now))
+        self.assertNotEqual(cygrpc.Timespec(now + 1), cygrpc.Timespec(now))
 
     def testCompletionQueueUpDown(self):
         completion_queue = cygrpc.CompletionQueue()