ronantakizawa/codereview-bench huggingface dataset:
Repository: jina-ai/jina
File: jina/serve/gateway.py (Python)

Code before:
            deployments_no_reduce=deployments_no_reduce,
            timeout_send=timeout_send,
            retries=args.retries,
            compression=args.compression,
            runtime_name=runtime_name,
            prefetch=args.prefetch,
            logger=self.logger,
            metrics_registry=metrics_registry,
            meter=meter,
            aio_tracing_client_interceptors=aio_tracing_client_interceptors,
            tracing_client_interceptor=tracing_client_interceptor,
        )

    @abc.abstractmethod
    async def setup_server(self):
        """Setup server"""
        ...

    @abc.abstractmethod
    async def run_server(self):
        """Run server forever"""
        ...

    @abc.abstractmethod
    async def shutdown(self):
        """Free other resources allocated with the server, e.g, gateway object, ..."""
        ...

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


Review comment (suggestion):
```suggestion
        """Shutdown the server and free other allocated resources, e.g, streamer object, health check service, ..."""
```

Code after:
        self.streamer = GatewayStreamer(
            graph_representation=graph_description,
            executor_addresses=deployments_addresses,
            graph_conditions=graph_conditions,
            deployments_metadata=deployments_metadata,
            deployments_no_reduce=deployments_no_reduce,
            timeout_send=timeout_send,
            retries=args.retries,
            compression=args.compression,
            runtime_name=runtime_name,
            prefetch=args.prefetch,
            logger=self.logger,
            metrics_registry=metrics_registry,
            meter=meter,
            aio_tracing_client_interceptors=aio_tracing_client_interceptors,
            tracing_client_interceptor=tracing_client_interceptor,
        )

    @abc.abstractmethod
    async def setup_server(self):
        """Setup server"""
        ...

    @abc.abstractmethod
    async def run_server(self):
        """Run server forever"""
        ...

    @abc.abstractmethod
    async def shutdown(self):
        """Shutdown the server and free other allocated resources, e.g, streamer object, health check service, ..."""
        ...

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def _set_single_port_protocol(self):
        if len(self.runtime_args.port) < 1 or len(self.runtime_args.protocol) < 1:
            raise ValueError(f'{self.__class__} expects at least 1 port and 1 protcol')
        self.port = self.runtime_args.port[0]
Repository: jina-ai/jina
File: tests/unit/test_importer.py (Python)

Code before:
    with ImportExtensions(required=False, logger=default_logger) as ie:
        ie._tags = fake_tags
        from ngt.abc import edf

    assert ie._tags == fake_tags

    with ImportExtensions(required=False, logger=default_logger) as ie:
        import abcdefg

    assert not ie._tags


def test_no_suppress_other_exception():
    with pytest.raises(Exception):
        with ImportExtensions(required=False, logger=default_logger):
            raise Exception

    with pytest.raises(Exception):
        with ImportExtensions(required=True, logger=default_logger):
            raise Exception


def test_path_importer():
    from jina.importer import PathImporter
    with pytest.raises(ImportError):
        PathImporter.add_modules('unit.error_package')


Review comment (suggestion):
The `ModuleNotFoundError` is not covered in this case. 

Code after:
    with ImportExtensions(required=False, logger=default_logger) as ie:
        ie._tags = fake_tags
        from ngt.abc import edf

    assert ie._tags == fake_tags

    with ImportExtensions(required=False, logger=default_logger) as ie:
        import abcdefg

    assert not ie._tags


def test_no_suppress_other_exception():
    with pytest.raises(Exception):
        with ImportExtensions(required=False, logger=default_logger):
            raise Exception

    with pytest.raises(Exception):
        with ImportExtensions(required=True, logger=default_logger):
            raise Exception


def test_path_importer(tmpdir):
    tmpmodule = f'package.py'
    with open(tmpdir / tmpmodule, 'w') as f:
        f.write("raise ImportError")

    from jina.importer import PathImporter
    with pytest.raises(ImportError):
        PathImporter.add_modules(tmpdir / tmpmodule)
    with pytest.raises(FileNotFoundError):
        PathImporter.add_modules('some_package.py')
Repository: jina-ai/jina
File: tests/integration/issues/github_4488/test_deployment_protocol.py (Python)

Code before:
        pass


@pytest.fixture
def cert_prefix():
    cur_dir = os.path.dirname(os.path.abspath(__file__))
    return f'{cur_dir}/../../../unit/serve/runtimes/gateway/grpc/cert/'


@pytest.fixture
def cert_pem(cert_prefix):
    """This is the cert entry of a self-signed local cert"""
    return cert_prefix + '/server.crt'


@pytest.fixture
def key_pem(cert_prefix):
    """This is the key entry of a self-signed local cert"""
    return cert_prefix + '/server.key'


@pytest.mark.parametrize('protocol', ['http', 'websocket', 'grpc'])
@pytest.mark.parametrize('tls', [True, False])
@pytest.mark.parametrize(
    'uses',
    ['jinahub+sandbox://DummyHubExecutor', 'jinaai+sandbox://jina-ai/DummyHubExecutor'],
)
def test_deployment_protocol(protocol, tls, cert_pem, key_pem, uses):
    cert = cert_pem if tls else None
    key = key_pem if tls else None
    f = (
        Flow(protocol=protocol, ssl_certfile=cert, ssl_keyfile=key)
        .add(uses=MyExec)
        .add(uses=uses)
    )
    with f:
        for node, v in f._deployment_nodes.items():
            p = v.protocol.lower()
            if node == 'gateway':
                assert p == protocol + ('s' if tls else '')
            elif node == 'executor0':
                assert p == 'grpc'
            elif node == 'executor1':
                assert p == 'grpcs'


Review comment (suggestion):
```suggestion
    ['jinaai+sandbox://jina-ai/DummyHubExecutor'],
```

Code after:
        pass


@pytest.fixture
def cert_prefix():
    cur_dir = os.path.dirname(os.path.abspath(__file__))
    return f'{cur_dir}/../../../unit/serve/runtimes/gateway/grpc/cert/'


@pytest.fixture
def cert_pem(cert_prefix):
    """This is the cert entry of a self-signed local cert"""
    return cert_prefix + '/server.crt'


@pytest.fixture
def key_pem(cert_prefix):
    """This is the key entry of a self-signed local cert"""
    return cert_prefix + '/server.key'


@pytest.mark.parametrize('protocol', ['http', 'websocket', 'grpc'])
@pytest.mark.parametrize('tls', [True, False])
@pytest.mark.parametrize(
    'uses',
    ['jinaai+sandbox://jina-ai/DummyHubExecutor'],
)
def test_deployment_protocol(protocol, tls, cert_pem, key_pem, uses):
    cert = cert_pem if tls else None
    key = key_pem if tls else None
    f = (
        Flow(protocol=protocol, ssl_certfile=cert, ssl_keyfile=key)
        .add(uses=MyExec)
        .add(uses=uses)
    )
    with f:
        for node, v in f._deployment_nodes.items():
            p = v.protocol.lower()
            if node == 'gateway':
                assert p == protocol + ('s' if tls else '')
            elif node == 'executor0':
                assert p == 'grpc'
            elif node == 'executor1':
                assert p == 'grpcs'
Repository: jina-ai/jina
File: tests/integration/sandbox/test_sandbox.py (Python)

Code before:
import pytest

from jina import Document, Flow


@pytest.mark.parametrize('endpoint', ['foo', 'bar'])
@pytest.mark.parametrize(
    'uses', ['jinahub+sandbox://Hello', 'jinaai+sandbox://jina-ai/Hello']
)
def test_sandbox(endpoint, uses):
    with Flow().add(uses=uses) as f:
        da = f.post(
            endpoint,
            [
                Document(text="dog world"),
                Document(text="cat world"),
                Document(id="a", text="elephant world"),
                Document(id="b", text="monkey world"),
            ],
        )
        assert da.texts == [
            'hello dog world',
            'hello cat world',
            'hello elephant world',
            'hello monkey world',
        ]


Review comment (suggestion):
```suggestion
    'uses', ['jinaai+sandbox://jina-ai/Hello']
```

Code after:
import pytest

from jina import Document, Flow


@pytest.mark.parametrize('endpoint', ['foo', 'bar'])
@pytest.mark.parametrize(
    'uses', ['jinaai+sandbox://jina-ai/Hello']
)
def test_sandbox(endpoint, uses):
    with Flow().add(uses=uses) as f:
        da = f.post(
            endpoint,
            [
                Document(text="dog world"),
                Document(text="cat world"),
                Document(id="a", text="elephant world"),
                Document(id="b", text="monkey world"),
            ],
        )
        assert da.texts == [
            'hello dog world',
            'hello cat world',
            'hello elephant world',
            'hello monkey world',
        ]
Repository: jina-ai/jina
File: tests/unit/orchestrate/deployments/config/test_helper.py (Python)

Code before:
        # current version is published already
        count = 3
    requests_mock.get(
        'https://registry.hub.docker.com/v2/repositories/jinaai/jina/tags',
        text=json.dumps(
            {
                'count': count,
                'next': 'abc',
                'previous': 'def',
                'results': [{'a': 'b', 'c': 'd'}],
            }
        ),
    )
    v = get_base_executor_version()
    if is_master:
        assert v == 'master'
    else:
        assert v == __version__


def test_to_compatible_name():
    assert to_compatible_name('executor/hey-ha_HO') == 'executor-hey-ha-ho'


@pytest.mark.parametrize(
    'uses', ['jinahub://DummyExecutor', 'jinaai://jina-ai/DummyExecutor']
)
def test_get_image_name(mocker, monkeypatch, uses):
    mock = mocker.Mock()

    def _mock_fetch(
        name,
        tag,
        image_required=True,
        rebuild_image=True,
        *,
        secret=None,
        force=False,
    ):
        mock(name=name, rebuild_image=rebuild_image)

        return (
            HubExecutor(
                uuid='hello',
                name=name,
                tag='v0',
                image_name=f'jinahub/{name}',
                md5sum=None,
                visibility=True,
                archive_url=None,
            ),

Review comment (suggestion):
```suggestion
    'uses', ['jinaai://jina-ai/DummyExecutor']
```

Code after:
        # current version is published already
        count = 3
    requests_mock.get(
        'https://registry.hub.docker.com/v2/repositories/jinaai/jina/tags',
        text=json.dumps(
            {
                'count': count,
                'next': 'abc',
                'previous': 'def',
                'results': [{'a': 'b', 'c': 'd'}],
            }
        ),
    )
    v = get_base_executor_version()
    if is_master:
        assert v == 'master'
    else:
        assert v == __version__


def test_to_compatible_name():
    assert to_compatible_name('executor/hey-ha_HO') == 'executor-hey-ha-ho'


@pytest.mark.parametrize(
    'uses', ['jinaai://jina-ai/DummyExecutor']
)
def test_get_image_name(mocker, monkeypatch, uses):
    mock = mocker.Mock()

    def _mock_fetch(
        name,
        tag,
        image_required=True,
        rebuild_image=True,
        *,
        secret=None,
        force=False,
    ):
        mock(name=name, rebuild_image=rebuild_image)

        return (
            HubExecutor(
                uuid='hello',
                name=name,
                tag='v0',
                image_name=f'jinahub/{name}',
                md5sum=None,
                visibility=True,
                archive_url=None,
            ),
Repository: jina-ai/jina
File: tests/unit/orchestrate/deployments/config/test_helper.py (Python)

Code before:
        image_required=True,
        rebuild_image=True,
        *,
        secret=None,
        force=False,
    ):
        mock(name=name, rebuild_image=rebuild_image)

        return (
            HubExecutor(
                uuid='hello',
                name=name,
                tag='v0',
                image_name=f'jinahub/{name}',
                md5sum=None,
                visibility=True,
                archive_url=None,
            ),
            False,
        )

    monkeypatch.setattr(HubIO, 'fetch_meta', _mock_fetch)

    image_name = get_image_name(uses)

    assert image_name == 'jinahub/DummyExecutor'

    _, mock_kwargs = mock.call_args_list[0]
    assert mock_kwargs['rebuild_image'] is True  # default value must be True

    os.environ['JINA_HUB_NO_IMAGE_REBUILD'] = '1'

    get_image_name(uses)

    del os.environ['JINA_HUB_NO_IMAGE_REBUILD']

    _, mock_kwargs = mock.call_args_list[1]
    assert mock_kwargs['rebuild_image'] is False  # env var is set, so it must be False


Review comment (suggestion):
```suggestion
    assert image_name in {'jinahub/DummyExecutor', 'jinahub/jina-ai/DummyExecutor'}
```

Code after:
        image_required=True,
        rebuild_image=True,
        *,
        secret=None,
        force=False,
    ):
        mock(name=name, rebuild_image=rebuild_image)

        return (
            HubExecutor(
                uuid='hello',
                name=name,
                tag='v0',
                image_name=f'jinahub/{name}',
                md5sum=None,
                visibility=True,
                archive_url=None,
            ),
            False,
        )

    monkeypatch.setattr(HubIO, 'fetch_meta', _mock_fetch)

    image_name = get_image_name(uses)

    assert image_name in {'jinahub/DummyExecutor', 'jinahub/jina-ai/DummyExecutor'}

    _, mock_kwargs = mock.call_args_list[0]
    assert mock_kwargs['rebuild_image'] is True  # default value must be True

    os.environ['JINA_HUB_NO_IMAGE_REBUILD'] = '1'

    get_image_name(uses)

    del os.environ['JINA_HUB_NO_IMAGE_REBUILD']

    _, mock_kwargs = mock.call_args_list[1]
    assert mock_kwargs['rebuild_image'] is False  # env var is set, so it must be False
Repository: jina-ai/jina
File: tests/unit/orchestrate/pods/test_pod_factory.py (Python)

Code before:
import pytest
from hubble.executor.hubio import HubIO

from jina.orchestrate.pods.factory import PodFactory
from jina.parsers import set_pod_parser


@pytest.mark.parametrize(
    'uses', ['jinahub+docker://DummyExecutor', 'jinaai+docker://jina-ai/DummyExecutor']
)
def test_container_pod(mocker, monkeypatch, uses):
    mock = mocker.Mock()

    def _mock_pull(self):
        return 'docker://jinahub/dummy_executor'

    monkeypatch.setattr(HubIO, 'pull', _mock_pull)

    args = set_pod_parser().parse_args(['--uses', uses])
    pod = PodFactory.build_pod(args)
    assert pod.args.uses == 'docker://jinahub/dummy_executor'
    assert pod.name == 'ContainerPod'


Review comment (suggestion):
```suggestion
    'uses', ['jinaai+docker://jina-ai/DummyExecutor']
```

Code after:
import pytest
from hubble.executor.hubio import HubIO

from jina.orchestrate.pods.factory import PodFactory
from jina.parsers import set_pod_parser


@pytest.mark.parametrize(
    'uses', ['jinaai+docker://jina-ai/DummyExecutor']
)
def test_container_pod(mocker, monkeypatch, uses):
    mock = mocker.Mock()

    def _mock_pull(self):
        return 'docker://jinahub/dummy_executor'

    monkeypatch.setattr(HubIO, 'pull', _mock_pull)

    args = set_pod_parser().parse_args(['--uses', uses])
    pod = PodFactory.build_pod(args)
    assert pod.args.uses == 'docker://jinahub/dummy_executor'
    assert pod.name == 'ContainerPod'
Repository: jina-ai/jina
File: tests/unit/serve/executors/test_executor.py (Python)

Code before:

    e = threading.Event()

    kwargs = {'port_expose': exposed_port, 'stop_event': e}
    enable_dynamic_batching = request.param
    if enable_dynamic_batching:
        kwargs['uses_dynamic_batching'] = {
            '/bar': {'preferred_batch_size': 4, 'timeout': 5000}
        }

    t = threading.Thread(
        name='serve-exec',
        target=serve_exec,
        kwargs=kwargs,
    )
    t.start()
    time.sleep(3)  # allow Flow to start

    yield

    e.set()  # set event and stop (unblock) the Flow
    t.join()


@pytest.mark.parametrize(
    'uses', ['jinahub://DummyHubExecutor', 'jinaai://jina-ai/DummyHubExecutor']
)
def test_executor_load_from_hub(uses):
    exec = Executor.from_hub(uses, uses_metas={'name': 'hello123'})
    da = DocumentArray([Document()])
    exec.foo(da)
    assert da.texts == ['hello']
    assert exec.metas.name == 'hello123'


def test_executor_import_with_external_dependencies(capsys):
    ex = Executor.load_config('../../hubble-executor/config.yml')
    assert ex.bar == 123
    ex.foo()
    out, err = capsys.readouterr()
    assert 'hello' in out


def test_executor_with_pymodule_path():
    with pytest.raises(FileNotFoundError):
        _ = Executor.load_config(
            '''
        jtype: BaseExecutor
        py_modules:
            - jina.no_valide.executor
        '''

Review comment (suggestion):
```suggestion
    'uses', ['jinaai://jina-ai/DummyHubExecutor']
```

Code after:

    e = threading.Event()

    kwargs = {'port_expose': exposed_port, 'stop_event': e}
    enable_dynamic_batching = request.param
    if enable_dynamic_batching:
        kwargs['uses_dynamic_batching'] = {
            '/bar': {'preferred_batch_size': 4, 'timeout': 5000}
        }

    t = threading.Thread(
        name='serve-exec',
        target=serve_exec,
        kwargs=kwargs,
    )
    t.start()
    time.sleep(3)  # allow Flow to start

    yield

    e.set()  # set event and stop (unblock) the Flow
    t.join()


@pytest.mark.parametrize(
    'uses', ['jinaai://jina-ai/DummyHubExecutor']
)
def test_executor_load_from_hub(uses):
    exec = Executor.from_hub(uses, uses_metas={'name': 'hello123'})
    da = DocumentArray([Document()])
    exec.foo(da)
    assert da.texts == ['hello']
    assert exec.metas.name == 'hello123'


def test_executor_import_with_external_dependencies(capsys):
    ex = Executor.load_config('../../hubble-executor/config.yml')
    assert ex.bar == 123
    ex.foo()
    out, err = capsys.readouterr()
    assert 'hello' in out


def test_executor_with_pymodule_path():
    with pytest.raises(FileNotFoundError):
        _ = Executor.load_config(
            '''
        jtype: BaseExecutor
        py_modules:
            - jina.no_valide.executor
        '''
Repository: jina-ai/jina
File: tests/integration/concurrent_clients/test_multithread_client.py (Python)

Code before:

import threading
import time
from concurrent.futures import ThreadPoolExecutor

from jina import Client, Document, Executor, Flow, requests


class MyExecutor(Executor):
    @requests
    def foo(self, docs, **kwargs):
        for doc in docs:
            doc.text = 'I am coming from MyExecutor'
                
def flow_run(flow, stop_event):
    with flow:
        flow.block(stop_event)


def client_post(c):
    doc = Document(id='doc', text='hello world')
    result = c.post(on='/', inputs=doc)[0]
    return result


def test_multithread_client(capsys):
    stop_event = threading.Event()
    flow = Flow(port=12345).add(uses=MyExecutor)
    t = threading.Thread(target=flow_run, args=(flow, stop_event))

    c = Client(port=12345)

    try:
        with capsys.disabled():
            t.start()
            time.sleep(5)

        with ThreadPoolExecutor(max_workers=50) as pool:
            tasks = []
            for _ in range(1000):
                task = pool.submit(client_post, c)
                tasks.append(task)

            for task in tasks:
                result = task.result()
                assert result.id == 'doc'
                assert result.text == 'I am coming from MyExecutor'

Review comment (suggestion):
```suggestion
def client_post(client):
    doc = Document(id='doc', text='hello world')
    result = client.post(on='/', inputs=doc)[0]
```

Code after:

import threading
import time
from concurrent.futures import ThreadPoolExecutor

from jina import Client, Document, Executor, Flow, requests
from jina.helper import random_port


class MyExecutor(Executor):
    @requests
    def foo(self, docs, **kwargs):
        for doc in docs:
            doc.text = 'I am coming from MyExecutor'
                
def flow_run(flow, stop_event):
    with flow:
        flow.block(stop_event)


def client_post(doc, client):
    result = client.post(on='/', inputs=doc)[0]
    return result


def test_multithread_client(capsys):
    port = random_port()
    stop_event = threading.Event()
    flow = Flow(port=port).add(uses=MyExecutor)
    t = threading.Thread(target=flow_run, args=(flow, stop_event))

    c = Client(port=port)

    try:
        with capsys.disabled():
            t.start()
            time.sleep(5)

        with ThreadPoolExecutor(max_workers=50) as pool:
            tasks = []
            for i in range(1000):
                doc = Document(id=f'{i}', text='hello world')
                task = pool.submit(client_post, doc, c)
                tasks.append(task)

            for i,task in enumerate(tasks):
                result = task.result()
Repository: jina-ai/jina
File: jina/helper.py (Python)

Code before:
    .. warn::
        Set `timeout` to a large number will block the Flow.

    """
    import urllib.request

    results = []

    def _get_ip(url):
        try:
            req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
            with urllib.request.urlopen(req, timeout=timeout) as fp:
                _ip = fp.read().decode().strip()
                results.append(_ip)

        except:
            pass  # intentionally ignored, public ip is not showed

    ip_server_list = [
        'https://api.ipify.org',
        'https://ident.me',
        'https://checkip.amazonaws.com/',
    ]

    for idx, ip in enumerate(ip_server_list):
        results.append(_get_ip(ip))

    for r in results:
        if r:
            return r


def convert_tuple_to_list(d: Dict):
    """
    Convert all the tuple type values from a dict to list.

    :param d: Dict type of data.
    """
    for k, v in d.items():
        if isinstance(v, tuple):
            d[k] = list(v)
        elif isinstance(v, dict):
            convert_tuple_to_list(v)


def is_jupyter() -> bool:  # pragma: no cover
    """
    Check if we're running in a Jupyter notebook, using magic command `get_ipython` that only available in Jupyter.

    :return: True if run in a Jupyter notebook else False.
    """

Review comment (suggestion):
```suggestion
        r = _get_ip(ip)
        if r:
            return r
```

Now that it is not async we can just return the first one that is not None to save time. (As we only return one IP later anyway)


Code after:
    .. warn::
        Set `timeout` to a large number will block the Flow.

    """
    import urllib.request

    def _get_ip(url):
        try:
            req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
            with urllib.request.urlopen(req, timeout=timeout) as fp:
                _ip = fp.read().decode().strip()
                return _ip

        except:
            pass  # intentionally ignored, public ip is not showed

    ip_server_list = [
        'https://api.ipify.org',
        'https://ident.me',
        'https://checkip.amazonaws.com/',
    ]

    for idx, ip in enumerate(ip_server_list):
        r = _get_ip(ip)
        if r:
            return r


def convert_tuple_to_list(d: Dict):
    """
    Convert all the tuple type values from a dict to list.

    :param d: Dict type of data.
    """
    for k, v in d.items():
        if isinstance(v, tuple):
            d[k] = list(v)
        elif isinstance(v, dict):
            convert_tuple_to_list(v)


def is_jupyter() -> bool:  # pragma: no cover
    """
    Check if we're running in a Jupyter notebook, using magic command `get_ipython` that only available in Jupyter.

    :return: True if run in a Jupyter notebook else False.
    """
    try:
        get_ipython  # noqa: F821
    except NameError:
        return False