Changeset View
Changeset View
Standalone View
Standalone View
swh/counters/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 json | |||||
| import re | import re | ||||
| from typing import Any, Dict | from typing import Any, Dict | ||||
| import pytest | import pytest | ||||
| from redis import Redis as RedisClient | from redis import Redis as RedisClient | ||||
| import yaml | import yaml | ||||
| from swh.core.api import RPCServerApp | from swh.core.api import RPCServerApp | ||||
| Show All 13 Lines | |||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_counters_server_config_on_disk( | def swh_counters_server_config_on_disk( | ||||
| tmp_path, monkeypatch, swh_counters_server_config | tmp_path, monkeypatch, swh_counters_server_config | ||||
| ) -> str: | ) -> str: | ||||
| return _environment_config_file(tmp_path, monkeypatch, swh_counters_server_config) | return _environment_config_file(tmp_path, monkeypatch, swh_counters_server_config) | ||||
| @pytest.fixture | |||||
| def history_test_client(tmp_path, monkeypatch): | |||||
| cfg = { | |||||
| "counters": {"cls": "redis", "host": "redis:6379"}, | |||||
| "history": { | |||||
| "cls": "prometheus", | |||||
| "prometheus_host": "prometheus", | |||||
| "prometheus_port": "9090", | |||||
| "live_data_start": "0", | |||||
| "cache_base_directory": "/tmp", | |||||
| }, | |||||
| } | |||||
| _environment_config_file(tmp_path, monkeypatch, cfg) | |||||
| app = make_app_from_configfile() | |||||
| app.config["TESTING"] = True | |||||
| return app.test_client() | |||||
| def write_config_file(tmpdir, config_dict: Dict, name: str = "config.yml") -> str: | def write_config_file(tmpdir, config_dict: Dict, name: str = "config.yml") -> str: | ||||
| """Prepare configuration file in `$tmpdir/name` with content `content`. | """Prepare configuration file in `$tmpdir/name` with content `content`. | ||||
| Args: | Args: | ||||
| tmpdir (LocalPath): root directory | tmpdir (LocalPath): root directory | ||||
| content: Content of the file either as string or as a dict. | content: Content of the file either as string or as a dict. | ||||
| If a dict, converts the dict into a yaml string. | If a dict, converts the dict into a yaml string. | ||||
| name: configuration filename | name: configuration filename | ||||
| ▲ Show 20 Lines • Show All 95 Lines • ▼ Show 20 Lines | for collection in data.keys(): | ||||
| obj_type = f'object_type="{collection}"' | obj_type = f'object_type="{collection}"' | ||||
| assert obj_type in response | assert obj_type in response | ||||
| pattern = r'swh_archive_object_total{col="value", object_type="%s"} (\d+)' % ( | pattern = r'swh_archive_object_total{col="value", object_type="%s"} (\d+)' % ( | ||||
| collection | collection | ||||
| ) | ) | ||||
| m = re.search(pattern, response) | m = re.search(pattern, response) | ||||
| assert data[collection] == int(m.group(1)) | assert data[collection] == int(m.group(1)) | ||||
| def test_server_counters_history(history_test_client, mocker): | |||||
| """Test the counters history file download""" | |||||
| expected_result = {"content": [[1, 1], [2, 2]]} | |||||
| mock = mocker.patch("swh.counters.history.History.get_history") | |||||
| mock.return_value = expected_result | |||||
| r = history_test_client.get("/counters_history/test.json") | |||||
| assert 200 == r.status_code | |||||
| response = r.get_data().decode("utf-8") | |||||
| response_json = json.loads(response) | |||||
| assert response_json == expected_result | |||||
| def test_server_counters_history_file_not_found(history_test_client, mocker): | |||||
| """ensure a 404 is returned when the file doesn't exists""" | |||||
| mock = mocker.patch("swh.counters.history.History.get_history") | |||||
| mock.side_effect = FileNotFoundError | |||||
ardumont: What happens if you are asking crap to the endpoint?
as in `/counters_history/gimme-some… | |||||
Not Done Inline Actions/counters_history/../../../etc/passwd vlorentz: `/counters_history/../../../etc/passwd` | |||||
Not Done Inline Actionsardumont: ...or that ;)
/me notesm down @vlorentz in the naugthy list [1]
[1] https://xkcd.com/838/ | |||||
Done Inline Actions:) For the ../etc/password example, the filename is checked to avoid that when it's called from the RPC, (the _validate_filename method on history,py) for the get method, the .. are computed in the url so `/counters_history/../etc/passwd will be equivalent to /etc/password and return a 404 vsellier: :)
ok for the missing file, I didn't find (I confess I didn't search too long) how to return a… | |||||
Not Done Inline Actionsnotes* and naughty* (damn that's annoying to not be able to amend typos) and great for the validate filename thingy, I had forgotten about it ;) ardumont: `notes*` and `naughty*` (damn that's annoying to not be able to amend typos)
and great for… | |||||
| r = history_test_client.get("/counters_history/fake.json") | |||||
| assert 404 == r.status_code | |||||
What happens if you are asking crap to the endpoint?
as in /counters_history/gimme-some-loving-with-some-inexistent-file ;)
;D (kaboom ;)