Unable to run tests where streams are returned

[ArjunBansil]

I've been writing a unit test to test my grpc server. If I'm receiving unary responses, the tests work just fine. However, when it comes to receiving streams, the test just hangs and won't even fail. Is there anything I can do or is this just a limitation?

import pytest
from .context import pred_pb2
from .context import pred_pb2_grpc
from pred_pb2 import PredictRequest
from .context import predictor
from .context import grpc

pytest -k test_local_client.py -v --disable-pytest-warnings

@pytest.fixture(scope="module")
def grpc_add_to_server():
from pred_pb2_grpc import add_PredictorServicer_to_server

return add_PredictorServicer_to_server

@pytest.fixture(scope="module")
def grpc_servicer():
from predictor import PredictorService
return PredictorService()

@pytest.fixture(scope="module")
def grpc_stub(grpc_channel=grpc.insecure_channel("localhost:8000")):
from pred_pb2_grpc import PredictorStub
return PredictorStub(grpc_channel)

def test_single_prediction(grpc_stub):
request = PredictRequest(jobTitle="Nothing", jobDescription="Nothing")
response = grpc_stub.make_prediction(request)
assert len(f'pred-{response.word1}') > 0

def test_send_stream_to_unary(grpc_stub):
jd = "nonsense"
jt = "nonsense"
features = []
for i in range(20):
features.append(PredictRequest(jobTitle=jt, jobDescription=jd))

request = iter(features)
with pytest.raises(grpc.RpcError):
    grpc_stub.make_prediction(request)

with pytest.raises(grpc.RpcError):
    grpc_stub.unary_stream_prediction(request)

def test_unary_stream_prediction(grpc_stub):
jd = "nonsense"
jt = "nonsense"
request = PredictRequest(jobTitle=jt, jobDescription=jd)
responses = grpc_stub.unary_stream_prediction(request)
for r in responses:
print(r.word1)
assert True

def test_send_unary_to_stream(grpc_stub):
request = PredictRequest(jobTitle="nonsense", jobDescription="nonsense")
with pytest.raises(grpc.RpcError):
grpc_stub.stream_unary_prediction(request)

@pytest.mark.timeout(20)
def test_stream_stream():
jt = "Software Engineer"
jd = "A person who types on a keyboard all the time"
from pred_pb2_grpc import PredictorStub

 stub = PredictorStub(grpc.insecure_channel("localhost:8000"))
 j_d_list = jd.split()
 features = []
 for string in j_d_list:
     features.append(PredictRequest(jobTitle=jt, jobDescription=string))
 request = iter(features)
 responses = stub.stream_stream_prediction(request)
 for r in responses:
     print(r.word1)

 assert True

[NickYin0125]

I got the same issue, have you figured it out yet?