Changeset View
Changeset View
Standalone View
Standalone View
swh/search/tests/test_server.py
| # Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import os | import os | ||||
| from typing import Any, Dict | from typing import Any, Dict | ||||
| import pytest | import pytest | ||||
| import yaml | import yaml | ||||
| from swh.core.api import RPCServerApp | from swh.core.api import RPCServerApp | ||||
| from swh.core.config import load_from_envvar | from swh.core.config import load_from_envvar | ||||
| from swh.search.api import server | |||||
| from swh.search.api.server import load_and_check_config, make_app_from_configfile | from swh.search.api.server import load_and_check_config, make_app_from_configfile | ||||
| def teardown_function(): | |||||
| # Ensure there is no configuration loaded from a previous test | |||||
ardumont: til, great. | |||||
| server.api_cfg = None | |||||
| def _write_config_file(tmp_path, monkeypatch, content): | def _write_config_file(tmp_path, monkeypatch, content): | ||||
| conf_path = os.path.join(str(tmp_path), "search.yml") | conf_path = os.path.join(str(tmp_path), "search.yml") | ||||
| with open(conf_path, "w") as f: | with open(conf_path, "w") as f: | ||||
| f.write(yaml.dump(content)) | f.write(yaml.dump(content)) | ||||
| monkeypatch.setenv("SWH_CONFIG_FILENAME", conf_path) | monkeypatch.setenv("SWH_CONFIG_FILENAME", conf_path) | ||||
| return conf_path | return conf_path | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_search_server_config_without_indexes() -> Dict[str, Any]: | def swh_search_server_config_without_indexes() -> Dict[str, Any]: | ||||
| return {"search": {"cls": "elasticsearch", "hosts": ["es1"],}} | return {"search": {"cls": "elasticsearch", "hosts": ["es1"],}} | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_search_server_config_with_indexes( | def swh_search_server_config_with_indexes( | ||||
| swh_search_server_config_without_indexes, | swh_search_server_config_without_indexes, | ||||
| ) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
| return { | return { | ||||
| **swh_search_server_config_without_indexes, | "search": { | ||||
| **{"indexes": {"origin": {"index": "test"}}}, | **{"indexes": {"origin": {"index": "test"}}}, | ||||
| **swh_search_server_config_without_indexes["search"], | |||||
| } | |||||
| } | } | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_search_config_without_indexes( | def swh_search_config_without_indexes( | ||||
| monkeypatch, swh_search_server_config_without_indexes, tmp_path | monkeypatch, swh_search_server_config_without_indexes, tmp_path | ||||
| ): | ): | ||||
| return _write_config_file( | return _write_config_file( | ||||
| ▲ Show 20 Lines • Show All 72 Lines • ▼ Show 20 Lines | ): | ||||
| app2 = make_app_from_configfile() | app2 = make_app_from_configfile() | ||||
| assert app is app2 | assert app is app2 | ||||
| def test_server_make_app_from_config_file_with_indexes(swh_search_config_with_indexes,): | def test_server_make_app_from_config_file_with_indexes(swh_search_config_with_indexes,): | ||||
| app = make_app_from_configfile() | app = make_app_from_configfile() | ||||
| expected_cfg = load_from_envvar() | expected_cfg = load_from_envvar() | ||||
| assert app is not None | assert app is not None | ||||
| assert isinstance(app, RPCServerApp) | assert isinstance(app, RPCServerApp) | ||||
| assert app.config["search"] == expected_cfg["search"] | assert app.config["search"] == expected_cfg["search"] | ||||
| app2 = make_app_from_configfile() | app2 = make_app_from_configfile() | ||||
| assert app is app2 | assert app is app2 | ||||
| def test_server_first_call_initialize_elasticsearch( | |||||
| swh_search_config_with_indexes, mocker | |||||
| ): | |||||
| """Test the initialize method is called during the first and first only | |||||
| request to the server""" | |||||
| mock = mocker.patch("swh.search.elasticsearch.ElasticSearch.initialize") | |||||
| app = make_app_from_configfile() | |||||
| app.config["TESTING"] = True | |||||
| tc = app.test_client() | |||||
| tc.get("/") | |||||
| assert mock.call_count == 1 | |||||
| tc.get("/") | |||||
| assert mock.call_count == 1 | |||||
til, great.