onnxruntime/server/test/integration_tests/model_zoo_tests.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

import unittest
import random
import os
import sys


class ModelZooTests(unittest.TestCase):
    server_ip = '127.0.0.1'
    server_port = 54321
    grpc_port = 56789
    url_pattern = 'http://{0}:{1}/v1/models/{2}/versions/{3}:predict'
    server_app_path = ''  # Required
    log_level = 'verbose'
    server_ready_in_seconds = 10
    server_off_in_seconds = 100
    need_data_preparation = False
    need_data_cleanup = False
    model_zoo_model_path = ''  # Required
    model_zoo_test_data_path = ''  # Required
    supported_opsets = ['opset7', 'opset8', 'opset9', 'opset_7', 'opset_8', 'opset_9']
    skipped_models = [
        ('opset7', 'tf_inception_v2'),  # Known issue
    ]

    def __test_model(self, model_path, data_paths):
        json_request_headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        pb_request_headers = {
            'Content-Type': 'application/octet-stream',
            'Accept': 'application/octet-stream'
        }

        server_app_proc = None
        try:
            onnx_file_path = ''
            for fname in os.listdir(model_path):
                if not fname.startswith(".") and fname.endswith(".onnx") and os.path.isfile(os.path.join(model_path, fname)):
                    onnx_file_path = os.path.join(model_path, fname)
                    break

            if onnx_file_path == '':
                raise FileNotFoundError('Could not find any *.onnx file in {0}'.format(model_path))
            
            cmd = [self.server_app_path, '--http_port', str(self.server_port), '--model_path', onnx_file_path, '--log_level', self.log_level, '--grpc_port', str(self.grpc_port)]
            test_util.test_log(cmd)
            server_app_proc = test_util.launch_server_app(cmd, self.server_ip, self.server_port,
                                                            self.server_ready_in_seconds)

            test_util.test_log('[{0}] Run tests...'.format(model_path))
            for test in data_paths:
                test_util.test_log('[{0}] Current: {0}'.format(model_path, test))

                test_util.test_log('[{0}] JSON payload testing ....'.format(model_path))
                url = self.url_pattern.format(self.server_ip, self.server_port, 'default_model', 12345)
                with open(os.path.join(test, 'request.json')) as f:
                    request_payload = f.read()
                resp = test_util.make_http_request(url, json_request_headers, request_payload)
                test_util.json_response_validation(self, resp, os.path.join(test, 'response.json'))

                test_util.test_log('[{0}] Protobuf payload testing ....'.format(model_path))
                url = self.url_pattern.format(self.server_ip, self.server_port, 'default_model', 54321)
                with open(os.path.join(test, 'request.pb'), 'rb') as f:
                    request_payload = f.read()
                resp = test_util.make_http_request(url, pb_request_headers, request_payload)
                test_util.pb_response_validation(self, resp, os.path.join(test, 'response.pb'))

                test_util.test_log('[{0}] GRPC testing ....'.format(model_path))
                uri = ("{}:{}".format(self.server_ip, self.grpc_port))
                with open(os.path.join(test, 'request.pb'), 'rb') as f:
                    request_payload = f.read()
                with grpc.insecure_channel(uri) as channel:
                    stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)
                    resp = stub.Predict(request_payload)
                test_util.pb_response_validation(self, resp, os.path.join(test, 'response.pb'))
        finally:
            test_util.shutdown_server_app(server_app_proc, self.server_off_in_seconds)


    def test_models_from_model_zoo(self):
        model_data_map = {}
        for opset in self.supported_opsets:
            test_data_folder = os.path.join(self.model_zoo_test_data_path, opset)
            model_file_folder = os.path.join(self.model_zoo_model_path, opset)

            if os.path.isdir(test_data_folder):
                for name in os.listdir(test_data_folder):
                    if (opset, name) in self.skipped_models:
                        test_util.test_log("  Skip {0}:{1}".format(opset, name))
                        continue

                    if os.path.isdir(os.path.join(test_data_folder, name)):
                        current_dir = os.path.join(test_data_folder, name)
                        model_data_map[os.path.join(model_file_folder, name)] = [os.path.join(current_dir, name) for name in os.listdir(current_dir) if os.path.isdir(os.path.join(current_dir, name))]

        test_util.test_log('Planned models and test data:')
        for model_data, data_paths in model_data_map.items():
            test_util.test_log(model_data)
            for data in data_paths:
                test_util.test_log('\t\t{0}'.format(data))
        test_util.test_log('-----------------------')

        self.server_port = random.randint(30000, 40000)
        self.grpc_port = self.server_port + 1
        for model_path, data_paths in model_data_map.items():
            self.__test_model(model_path, data_paths)


if __name__ == '__main__':
    sys.path.append(sys.argv[4])
    sys.path.append(sys.argv[5])

    import test_util

    loader = unittest.TestLoader()

    test_classes = [ModelZooTests]

    test_suites = []
    for tests in test_classes:
        tests.server_app_path = sys.argv[1]
        tests.model_zoo_model_path = sys.argv[2]
        tests.model_zoo_test_data_path = sys.argv[3]

        test_suites.append(loader.loadTestsFromTestCase(tests))

    suites = unittest.TestSuite(test_suites)
    runner = unittest.TextTestRunner(verbosity=2)

    results = runner.run(suites)