diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3c85b4d397c..8940db1f8e6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -117,8 +117,10 @@ jobs:
           python-version: 3.7
       - name: Install test dependencies
         run: ./ci/install_test_deps.sh
+      - name: Install fastai
+        run: python -m pip install fastai
       - name: Run tests
-        run: ./ci/fastai_integration_tests.sh
+        run: ./ci/test_project.sh tests/integration/projects/fastai2
       - name: Upload test coverage to Codecov
         uses: codecov/codecov-action@v1.0.12
 
@@ -141,7 +143,7 @@ jobs:
         uses: codecov/codecov-action@v1.0.12
 
   api_server_integration_tests:
-    name: ${{ matrix.os }} API Server Integration Tests
+    name: API Server Integration Tests (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
@@ -158,12 +160,14 @@ jobs:
       - name: Install test dependencies
         run: ./ci/install_test_deps.sh
       - name: Run tests
-        run: ./ci/integration_tests.sh
+        run: ./ci/test_project.sh tests/integration/projects/general
+      - name: Run tests
+        run: ./ci/test_project.sh tests/integration/projects/general_non_batch
       - name: Upload test coverage to Codecov
         uses: codecov/codecov-action@v1.0.12
 
   back_compatibility_integration_tests:
-    name: ${{ matrix.os }} Back Compatibility Integration Tests
+    name: Backward Compatibility Integration Tests (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
diff --git a/bentoml/adapters/base_input.py b/bentoml/adapters/base_input.py
index 53f4b9e7d0b..60663a338ca 100644
--- a/bentoml/adapters/base_input.py
+++ b/bentoml/adapters/base_input.py
@@ -43,6 +43,7 @@ class BaseInputAdapter:
     def __init__(self, http_input_example=None, **base_config):
         self._config = base_config
         self._http_input_example = http_input_example
+        self.custom_request_schema = base_config.get('request_schema')
 
     @property
     def config(self):
diff --git a/bentoml/service/inference_api.py b/bentoml/service/inference_api.py
index b7152322c8a..d88dba6f86a 100644
--- a/bentoml/service/inference_api.py
+++ b/bentoml/service/inference_api.py
@@ -203,7 +203,11 @@ def request_schema(self):
         """
         :return: the HTTP API request schema in OpenAPI/Swagger format
         """
-        schema = self.input_adapter.request_schema
+        if self.input_adapter.custom_request_schema is None:
+            schema = self.input_adapter.request_schema
+        else:
+            schema = self.input_adapter.custom_request_schema
+
         if schema.get('application/json'):
             schema.get('application/json')[
                 'example'
diff --git a/ci/fastai_integration_tests.sh b/ci/fastai_integration_tests.sh
deleted file mode 100755
index eb8271d8611..00000000000
--- a/ci/fastai_integration_tests.sh
+++ /dev/null
@@ -1,16 +0,0 @@
-#!/usr/bin/env bash
-set -x
-
-# https://stackoverflow.com/questions/42218009/how-to-tell-if-any-command-in-bash-script-failed-non-zero-exit-status/42219754#42219754
-# Set err to 1 if pytest failed.
-error=0
-trap 'error=1' ERR
-
-GIT_ROOT=$(git rev-parse --show-toplevel)
-cd "$GIT_ROOT" || exit
-
-# Install PyTorch
-pip install fastai
-pytest "$GIT_ROOT"/tests/integration/test_fastai2_model_artifact.py --cov=bentoml --cov-config=.coveragerc
-
-test $error = 0 # Return non-zero if pytest failed
diff --git a/ci/integration_tests.sh b/ci/test_project.sh
similarity index 50%
rename from ci/integration_tests.sh
rename to ci/test_project.sh
index d91e86fc5b3..4ff066d85e1 100755
--- a/ci/integration_tests.sh
+++ b/ci/test_project.sh
@@ -9,28 +9,15 @@ GIT_ROOT=$(git rev-parse --show-toplevel)
 cd "$GIT_ROOT" || exit
 
 # Run test
-PROJECT_PATH="$GIT_ROOT/tests/integration/projects/general"
+PROJECT_PATH="$GIT_ROOT/$1"
 BUILD_PATH="$PROJECT_PATH/build"
 python "$PROJECT_PATH/model/model.py" "$BUILD_PATH/artifacts"
 python "$PROJECT_PATH/service.py" "$BUILD_PATH/artifacts" "$BUILD_PATH/dist"
 if [ "$(uname)" == "Darwin" ]; then
 	export OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES
-	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist"
+	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist" --cov=bentoml --cov-config=.coveragerc
 else
-	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist" --docker
-fi
-rm -r $BUILD_PATH
-
-# test the non batch service
-PROJECT_PATH="$GIT_ROOT/tests/integration/projects/general_non_batch"
-BUILD_PATH="$PROJECT_PATH/build"
-python "$PROJECT_PATH/model/model.py" "$BUILD_PATH/artifacts"
-python "$PROJECT_PATH/service.py" "$BUILD_PATH/artifacts" "$BUILD_PATH/dist"
-if [ "$(uname)" == "Darwin" ]; then
-	export OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES
-	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist"
-else
-	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist" --docker
+	python -m pytest -s "$PROJECT_PATH" --bento-dist "$BUILD_PATH/dist" --docker --cov=bentoml --cov-config=.coveragerc
 fi
 rm -r $BUILD_PATH
 
diff --git a/tests/conftest.py b/tests/conftest.py
index f76b5f33bfb..9c0ebfac3ae 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,6 +2,8 @@
 import glob
 import inspect
 import os
+import subprocess
+from typing import Callable, List, Optional, Union
 
 import imageio
 import numpy as np
@@ -11,47 +13,80 @@
 from tests.bento_service_examples.example_bento_service import ExampleBentoService
 
 
+# Assert HTTP server responses. Sending request asynchronously.
+async def assert_http_request(
+    method: str,
+    url: str,
+    headers: Optional[dict] = None,
+    data: Optional[str] = None,
+    timeout: Optional[int] = None,
+    assert_status: Union[Callable, int, None] = None,
+    assert_data: Union[Callable, bytes, None] = None,
+):
+    if assert_status is None:
+        assert_status = 200
+
+    import aiohttp
+
+    try:
+        async with aiohttp.ClientSession() as sess:
+            async with sess.request(
+                method, url, data=data, headers=headers, timeout=timeout
+            ) as r:
+                r_body = await r.read()
+    except RuntimeError:
+        # the event loop has been closed due to previous task failed, ignore
+        return
+
+    if callable(assert_status):
+        assert assert_status(r.status), f"{r.status} {r_body}"
+    else:
+        assert r.status == assert_status, f"{r.status} {r_body}"
+
+    if assert_data is not None:
+        if callable(assert_data):
+            assert assert_data(r_body), r_body
+        else:
+            assert r_body == assert_data
+
+
+# Assert system command std output.
+def assert_command(
+    cmd: List[str],
+    assert_out: Union[Callable, str, None] = None,
+    assert_err: Union[Callable, str, None] = None,
+    strip=True,
+):
+    with subprocess.Popen(
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=os.environ,
+    ) as proc:
+        stdout = proc.stdout.read().decode('utf-8')
+        stderr = proc.stderr.read().decode('utf-8')
+
+    if assert_out is not None:
+        if callable(assert_out):
+            assert assert_out(stdout)
+        elif strip:
+            assert stdout.strip() == assert_out.strip()
+        else:
+            assert stdout == assert_out
+
+    if assert_err is not None:
+        if callable(assert_err):
+            assert assert_err(stderr)
+        elif strip:
+            assert stderr.strip() == assert_err.strip()
+        else:
+            assert stderr == assert_err
+
+
 def pytest_configure():
     '''
     global constants for tests
     '''
-    # async request client
-    async def assert_request(
-        method,
-        url,
-        headers=None,
-        data=None,
-        timeout=None,
-        assert_status=None,
-        assert_data=None,
-    ):
-        if assert_status is None:
-            assert_status = 200
-
-        import aiohttp
-
-        try:
-            async with aiohttp.ClientSession() as sess:
-                async with sess.request(
-                    method, url, data=data, headers=headers, timeout=timeout
-                ) as r:
-                    r_body = await r.read()
-        except RuntimeError:
-            # the event loop has been closed due to previous task failed, ignore
-            return
-
-        if callable(assert_status):
-            assert assert_status(r.status), f"{r.status} {r_body}"
-        else:
-            assert r.status == assert_status, f"{r.status} {r_body}"
-
-        if assert_data is not None:
-            if callable(assert_data):
-                assert assert_data(r_body), r_body
-            else:
-                assert r_body == assert_data
 
-    pytest.assert_request = assert_request
+    pytest.assert_request = assert_http_request
+    pytest.assert_command = assert_command
 
     # dataframe json orients
     pytest.DF_ORIENTS = {
diff --git a/tests/integration/fastai_utils.py b/tests/integration/fastai_utils.py
deleted file mode 100644
index 193383e9f66..00000000000
--- a/tests/integration/fastai_utils.py
+++ /dev/null
@@ -1,5 +0,0 @@
-import numpy as np
-
-
-def get_items(x):
-    return np.ones([5, 5], np.float32)
diff --git a/tests/integration/projects/conftest.py b/tests/integration/projects/conftest.py
index 95fa41f3144..d5c070a1bff 100644
--- a/tests/integration/projects/conftest.py
+++ b/tests/integration/projects/conftest.py
@@ -24,6 +24,14 @@ def clean_context():
         yield stack
 
 
+@pytest.fixture(scope="session")
+def bundle(pytestconfig):
+    test_svc_bundle = pytestconfig.getoption("bento_dist") or os.path.join(
+        sys.argv[1], "build", "dist"
+    )
+    return test_svc_bundle
+
+
 @pytest.fixture(params=[True, False], scope="module")
 def enable_microbatch(request):
     pytest.enable_microbatch = request.param
@@ -31,18 +39,20 @@ def enable_microbatch(request):
 
 
 @pytest.fixture(scope="module")
-def host(pytestconfig, clean_context, enable_microbatch):
-    test_svc_bundle = pytestconfig.getoption("bento_dist") or os.path.join(
-        sys.argv[1], "build", "dist"
-    )
-    print(test_svc_bundle)
-
+def host(pytestconfig, bundle, clean_context, enable_microbatch):
     if pytestconfig.getoption("docker"):
         image = clean_context.enter_context(
-            build_api_server_docker_image(test_svc_bundle, "example_service")
+            build_api_server_docker_image(bundle, "example_service")
         )
         with run_api_server_docker_container(image, enable_microbatch) as host:
             yield host
     else:
-        with run_api_server(test_svc_bundle, enable_microbatch) as host:
+        with run_api_server(bundle, enable_microbatch) as host:
             yield host
+
+
+@pytest.fixture(scope="module")
+def service(bundle):
+    import bentoml
+
+    return bentoml.load_from_dir(bundle)
diff --git a/tests/integration/projects/fastai2/model/__init__.py b/tests/integration/projects/fastai2/model/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/tests/integration/projects/fastai2/model/model.py b/tests/integration/projects/fastai2/model/model.py
new file mode 100644
index 00000000000..3e6006cce5f
--- /dev/null
+++ b/tests/integration/projects/fastai2/model/model.py
@@ -0,0 +1,55 @@
+import pathlib
+import sys
+
+import numpy as np
+import torch
+from fastai.data.block import DataBlock
+from fastai.learner import Learner
+from fastai.torch_core import Module
+from torch import nn
+
+
+def get_items(_x):
+    return np.ones([5, 5], np.float32)
+
+
+class Model(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.fc = nn.Linear(5, 1, bias=False)
+        torch.nn.init.ones_(self.fc.weight)
+
+    def forward(self, x):
+        return self.fc(x)
+
+
+class Loss(Module):
+    reduction = 'none'
+
+    def forward(self, x, _y):
+        return x
+
+    def activation(self, x):
+        return x
+
+    def decodes(self, x):
+        return x
+
+
+def pack_models(path):
+    from bentoml.frameworks.fastai import FastaiModelArtifact
+
+    model = Model()
+    loss = Loss()
+
+    dblock = DataBlock(get_items=get_items, get_y=np.sum)
+    dls = dblock.datasets(None).dataloaders()
+    learner = Learner(dls, model, loss)
+
+    FastaiModelArtifact("model").pack(learner).save(path)
+
+
+if __name__ == "__main__":
+    artifacts_path = sys.argv[1]
+    pathlib.Path(artifacts_path).mkdir(parents=True, exist_ok=True)
+    pack_models(artifacts_path)
diff --git a/tests/bento_service_examples/fastai2_classifier.py b/tests/integration/projects/fastai2/service.py
similarity index 56%
rename from tests/bento_service_examples/fastai2_classifier.py
rename to tests/integration/projects/fastai2/service.py
index 670b5a98a40..9acb62c03bb 100644
--- a/tests/bento_service_examples/fastai2_classifier.py
+++ b/tests/integration/projects/fastai2/service.py
@@ -1,5 +1,9 @@
-import bentoml
+import pathlib
+import sys
+
 import numpy as np
+
+import bentoml
 from bentoml.adapters import DataframeInput
 from bentoml.frameworks.fastai import FastaiModelArtifact
 
@@ -13,3 +17,16 @@ def predict(self, df):
         _, _, output = self.artifacts.model.predict(input_data)
 
         return output.squeeze().item()
+
+
+if __name__ == "__main__":
+    artifacts_path = sys.argv[1]
+    bento_dist_path = sys.argv[2]
+    service = FastaiClassifier()
+
+    from model.model import Loss, Model  # noqa # pylint: disable=unused-import
+
+    service.artifacts.load_all(artifacts_path)
+
+    pathlib.Path(bento_dist_path).mkdir(parents=True, exist_ok=True)
+    service.save_to_dir(bento_dist_path)
diff --git a/tests/integration/projects/fastai2/tests/test_service.py b/tests/integration/projects/fastai2/tests/test_service.py
new file mode 100644
index 00000000000..08f44410075
--- /dev/null
+++ b/tests/integration/projects/fastai2/tests/test_service.py
@@ -0,0 +1,7 @@
+import pandas
+
+test_df = pandas.DataFrame([[1] * 5])
+
+
+def test_fastai2_artifact_pack(service):
+    assert service.predict(test_df) == 5.0, 'Run inference before saving'
diff --git a/tests/integration/projects/general/service.py b/tests/integration/projects/general/service.py
index e79c7d2ffd0..d7565b4bf94 100644
--- a/tests/integration/projects/general/service.py
+++ b/tests/integration/projects/general/service.py
@@ -70,6 +70,23 @@ def predict_json(self, input_datas):
     def customezed_route(self, input_datas):
         return input_datas
 
+    CUSTOM_SCHEMA = {
+        "application/json": {
+            "schema": {
+                "type": "object",
+                "required": ["field1", "field2"],
+                "properties": {
+                    "field1": {"type": "string"},
+                    "field2": {"type": "uuid"},
+                },
+            },
+        }
+    }
+
+    @bentoml.api(input=JsonInput(request_schema=CUSTOM_SCHEMA), batch=True)
+    def customezed_schema(self, input_datas):
+        return input_datas
+
     @bentoml.api(input=JsonInput(), batch=True)
     def predict_strict_json(self, input_datas, tasks: Sequence[InferenceTask] = None):
         filtered_jsons = []
diff --git a/tests/integration/projects/general/tests/test_meta.py b/tests/integration/projects/general/tests/test_meta.py
index bc601f1bce3..9f599bb4cb6 100644
--- a/tests/integration/projects/general/tests/test_meta.py
+++ b/tests/integration/projects/general/tests/test_meta.py
@@ -34,3 +34,17 @@ def path_in_docs(response_body):
         data=json.dumps("hello"),
         assert_data=bytes('"hello"', 'ascii'),
     )
+
+
+@pytest.mark.asyncio
+async def test_customized_request_schema(host):
+    def has_customized_schema(doc_bytes):
+        json_str = doc_bytes.decode()
+        return "field1" in json_str
+
+    await pytest.assert_request(
+        "GET",
+        f"http://{host}/docs.json",
+        headers=(("Content-Type", "application/json"),),
+        assert_data=has_customized_schema,
+    )
diff --git a/tests/integration/projects/general/tests/test_microbatch.py b/tests/integration/projects/general/tests/test_microbatch.py
index 7910b26e406..3a1a5e9ea76 100644
--- a/tests/integration/projects/general/tests/test_microbatch.py
+++ b/tests/integration/projects/general/tests/test_microbatch.py
@@ -33,6 +33,15 @@ async def test_slow_server(host):
     await asyncio.gather(*tasks)
     assert time.time() - time_start < 12
 
+
+@pytest.mark.asyncio
+async def test_fast_server(host):
+    if not pytest.enable_microbatch:
+        pytest.skip()
+
+    A, B = 0.0002, 0.01
+    data = '{"a": %s, "b": %s}' % (A, B)
+
     req_count = 100
     tasks = tuple(
         pytest.assert_request(
@@ -46,17 +55,8 @@ async def test_slow_server(host):
     )
     await asyncio.gather(*tasks)
 
-
-@pytest.mark.asyncio
-async def test_fast_server(host):
-    if not pytest.enable_microbatch:
-        pytest.skip()
-
-    A, B = 0.0002, 0.01
-    data = '{"a": %s, "b": %s}' % (A, B)
-
     time_start = time.time()
-    req_count = 500
+    req_count = 200
     tasks = tuple(
         pytest.assert_request(
             "POST",
@@ -70,4 +70,4 @@ async def test_fast_server(host):
         for i in range(req_count)
     )
     await asyncio.gather(*tasks)
-    assert time.time() - time_start < 5
+    assert time.time() - time_start < 2
diff --git a/tests/integration/test_fastai2_model_artifact.py b/tests/integration/test_fastai2_model_artifact.py
deleted file mode 100644
index 651c8fb301f..00000000000
--- a/tests/integration/test_fastai2_model_artifact.py
+++ /dev/null
@@ -1,64 +0,0 @@
-import pytest
-
-import torch
-from torch import nn
-from fastai.torch_core import Module
-from fastai.learner import Learner
-from fastai.data.block import DataBlock
-import numpy as np
-import pandas
-
-import bentoml
-from bentoml.yatai.client import YataiClient
-from tests.bento_service_examples.fastai2_classifier import FastaiClassifier
-from tests.integration.fastai_utils import get_items
-
-
-class Model(nn.Module):
-    def __init__(self):
-        super().__init__()
-        self.fc = nn.Linear(5, 1, bias=False)
-        torch.nn.init.ones_(self.fc.weight)
-
-    def forward(self, x):
-        return self.fc(x)
-
-
-class Loss(Module):
-    reduction = 'none'
-
-    def forward(self, x, y):
-        return x
-
-    def activation(self, x):
-        return x
-
-    def decodes(self, x):
-        return x
-
-
-@pytest.fixture
-def fastai_learner():
-    model = Model()
-    loss = Loss()
-
-    dblock = DataBlock(get_items=get_items, get_y=np.sum)
-    dls = dblock.datasets(None).dataloaders()
-    learner = Learner(dls, model, loss)
-    return learner
-
-
-test_df = pandas.DataFrame([[1] * 5])
-
-
-def test_fastai2_artifact_pack(fastai_learner):
-    svc = FastaiClassifier()
-    svc.pack('model', fastai_learner)
-    assert svc.predict(test_df) == 5.0, 'Run inference before saving'
-
-    saved_path = svc.save()
-    loaded_svc = bentoml.load(saved_path)
-    assert loaded_svc.predict(test_df) == 5.0, 'Run inference from saved model'
-
-    yc = YataiClient()
-    yc.repository.delete(f'{svc.name}:{svc.version}')
diff --git a/tests/integration/utils.py b/tests/integration/utils.py
index 2fd91589fab..c6e076e960d 100644
--- a/tests/integration/utils.py
+++ b/tests/integration/utils.py
@@ -16,20 +16,21 @@
 
 def _wait_until_api_server_ready(host_url, timeout, container=None, check_interval=1):
     start_time = time.time()
+    proxy_handler = urllib.request.ProxyHandler({})
+    opener = urllib.request.build_opener(proxy_handler)
+    ex = None
     while time.time() - start_time < timeout:
         try:
-            if (
-                urllib.request.urlopen(f'http://{host_url}/healthz', timeout=1).status
-                == 200
-            ):
-                break
+            if opener.open(f'http://{host_url}/healthz', timeout=1).status == 200:
+                return
             elif container.status != "running":
                 break
             else:
                 logger.info("Waiting for host %s to be ready..", host_url)
                 time.sleep(check_interval)
         except Exception as e:  # pylint:disable=broad-except
-            logger.info(f"'{e}', retrying to connect to the host {host_url}...")
+            logger.info(f"retrying to connect to the host {host_url}...")
+            ex = e
             time.sleep(check_interval)
         finally:
             if container:
@@ -40,7 +41,8 @@ def _wait_until_api_server_ready(host_url, timeout, container=None, check_interv
                         logger.info(f">>> {log_record}")
     else:
         raise AssertionError(
-            f"Timed out waiting {timeout} seconds for Server {host_url} to be ready"
+            f"Timed out waiting {timeout} seconds for Server {host_url} to be ready, "
+            f"exception: {ex}"
         )
 
 
@@ -148,6 +150,8 @@ def print_log(p):
     ) as p:
         host_url = f"127.0.0.1:{port}"
         threading.Thread(target=print_log, args=(p,), daemon=True).start()
-        _wait_until_api_server_ready(host_url, timeout=timeout)
-        yield host_url
-        p.terminate()
+        try:
+            _wait_until_api_server_ready(host_url, timeout=timeout)
+            yield host_url
+        finally:
+            p.terminate()