This method tests if the embeddings are generated successfully
(self)
| 112 | |
| 113 | |
| 114 | def test_embedding(self): |
| 115 | """ |
| 116 | This method tests if the embeddings are generated successfully |
| 117 | """ |
| 118 | try: |
| 119 | self.setUp() |
| 120 | with grpc.insecure_channel("localhost:50051") as channel: |
| 121 | stub = backend_pb2_grpc.BackendStub(channel) |
| 122 | response = stub.LoadModel(backend_pb2.ModelOptions(Model="intfloat/e5-mistral-7b-instruct")) |
| 123 | self.assertTrue(response.success) |
| 124 | embedding_request = backend_pb2.PredictOptions(Embeddings="This is a test sentence.") |
| 125 | embedding_response = stub.Embedding(embedding_request) |
| 126 | self.assertIsNotNone(embedding_response.embeddings) |
| 127 | # assert that is a list of floats |
| 128 | self.assertIsInstance(embedding_response.embeddings, list) |
| 129 | # assert that the list is not empty |
| 130 | self.assertTrue(len(embedding_response.embeddings) > 0) |
| 131 | except Exception as err: |
| 132 | print(err) |
| 133 | self.fail("Embedding service failed") |
| 134 | finally: |
| 135 | self.tearDown() |
| 136 | |
| 137 | def test_concurrent_requests(self): |
| 138 | """ |