calculator_service_test.py

import time
import unittest

import grpc

import calculator_pb2
import grpc_testing
from calculator_pb2 import CalculationRequest, CalculationResponse
from server.calculator_service import CalculatorService

target_service = calculator_pb2.DESCRIPTOR.services_by_name['Calculator']


class CalculatorServerTest(unittest.TestCase):
    def setUp(self):
        self._real_time = grpc_testing.strict_real_time()
        self._fake_time = grpc_testing.strict_fake_time(time.time())
        service = CalculatorService()
        descriptors_to_services = {
            target_service: service
        }
        self._real_time_server = grpc_testing.server_from_dictionary(
            descriptors_to_services, self._real_time)
        self._fake_time_server = grpc_testing.server_from_dictionary(
            descriptors_to_services, self._fake_time)

    def test_successful_Square(self):
        request = CalculationRequest(value=2)

        rpc = self._real_time_server.invoke_unary_unary(
            target_service.methods_by_name['Square'], (),
            request, None)

        actual, trailing_metadata, code, details = rpc.termination()
        expected = CalculationResponse(value=4)

        self.assertEqual(expected, actual)
        self.assertIs(code, grpc.StatusCode.OK)

    def test_successful_NaturalNumberGenerator(self):
        request = CalculationRequest(value=2)

        rpc = self._real_time_server.invoke_unary_stream(
            target_service.methods_by_name['NaturalNumberGenerator'], (),
            request, None)

        actual = [
            rpc.take_response(),
            rpc.take_response(),
        ]
        trailing_metadata, code, details = rpc.termination()
        expected = [
            CalculationResponse(value=1),
            CalculationResponse(value=2)
        ]

        self.assertEqual(expected[0], actual[0])
        self.assertEqual(expected[1], actual[1])
        self.assertIs(code, grpc.StatusCode.OK)

    def test_successful_Summation(self):
        rpc = self._real_time_server.invoke_stream_unary(
            target_service.methods_by_name['Summation'], (), None)
        rpc.send_request(CalculationRequest(value=2))
        rpc.send_request(CalculationRequest(value=3))
        rpc.requests_closed()

        actual, trailing_metadata, code, details = rpc.termination()
        expected = CalculationResponse(value=5)

        self.assertEqual(expected, actual)
        self.assertIs(code, grpc.StatusCode.OK)

    def test_successful_Buffer3Sum(self):
        rpc = self._real_time_server.invoke_stream_stream(
            target_service.methods_by_name['Buffer3Sum'], (), None)
        rpc.send_request(CalculationRequest(value=1))
        rpc.send_request(CalculationRequest(value=2))
        rpc.send_request(CalculationRequest(value=3))

        actual = rpc.take_response()
        expected = CalculationResponse(value=6)

        self.assertEqual(expected, actual)

        rpc.send_request(CalculationRequest(value=4))
        rpc.send_request(CalculationRequest(value=5))
        rpc.send_request(CalculationRequest(value=6))

        actual = rpc.take_response()
        expected = CalculationResponse(value=15)

        self.assertEqual(expected, actual)

        rpc.send_request(CalculationRequest(value=7))
        rpc.send_request(CalculationRequest(value=8))
        rpc.requests_closed()

        trailing_metadata, code, details = rpc.termination()
        actual = rpc.take_response()
        expected = CalculationResponse(value=15)

        self.assertEqual(expected, actual)
        self.assertIs(code, grpc.StatusCode.OK)


if __name__ == "__main__":
    unittest.main()