Porównaj commity

..

4 Commity

Autor SHA1 Wiadomość Data
Sol Lee 0808018da7 Handle the URL with subdirectories 2024-03-25 12:11:38 +00:00
Sol Lee 1039d8dcb9 Mock the response for test_detect_ckan 2024-03-25 08:58:34 +00:00
Sol Lee 7712751c8f Make tests without parameterized information have their own test 2024-03-25 08:47:57 +00:00
Sol Lee 472c5cdc29 Make the splitting logic more straightforward 2024-03-25 02:37:31 +00:00
2 zmienionych plików z 23 dodań i 22 usunięć

Wyświetl plik

@ -1,4 +1,3 @@
import re
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from os import path from os import path
from urllib.parse import urlparse from urllib.parse import urlparse
@ -37,7 +36,6 @@ class CKAN(ContentProvider):
return self.session.get(url, **kwargs) return self.session.get(url, **kwargs)
urlopen = _request urlopen = _request
url_regex = r"/dataset/[a-z0-9_\\-]*$"
def detect(self, source, ref=None, extra_args=None): def detect(self, source, ref=None, extra_args=None):
"""Trigger this provider for things that resolve to a CKAN dataset.""" """Trigger this provider for things that resolve to a CKAN dataset."""
@ -45,14 +43,20 @@ class CKAN(ContentProvider):
if not parsed_url.netloc: if not parsed_url.netloc:
return None return None
url_parts = parsed_url.path.split("/")
if url_parts[-2] == "dataset":
self.dataset_id = url_parts[-1]
else:
return None
api_url_path = "/api/3/action/"
api_url = parsed_url._replace( api_url = parsed_url._replace(
path=re.sub(self.url_regex, "/api/3/action/", parsed_url.path) path="/".join(url_parts[:-2]) + api_url_path
).geturl() ).geturl()
status_show_url = f"{api_url}status_show" status_show_url = f"{api_url}status_show"
resp = self.urlopen(status_show_url) resp = self.urlopen(status_show_url)
if resp.status_code == 200: if resp.status_code == 200:
self.dataset_id = parsed_url.path.rsplit("/", maxsplit=1)[1]
self.version = self._fetch_version(api_url) self.version = self._fetch_version(api_url)
return { return {
"dataset_id": self.dataset_id, "dataset_id": self.dataset_id,

Wyświetl plik

@ -2,29 +2,26 @@ import os
from contextlib import contextmanager from contextlib import contextmanager
from tempfile import NamedTemporaryFile, TemporaryDirectory from tempfile import NamedTemporaryFile, TemporaryDirectory
import pytest
from repo2docker.contentproviders import CKAN from repo2docker.contentproviders import CKAN
test_ckan = CKAN()
test_hosts = [ def test_detect_ckan(requests_mock):
( mock_response = {"result": {"metadata_modified": "2024-02-27T14:15:54.573058"}}
[ requests_mock.get("http://demo.ckan.org/api/3/action/status_show", status_code=200)
"http://demo.ckan.org/dataset/sample-dataset-1", requests_mock.get(
], "http://demo.ckan.org/api/3/action/package_show?id=1234", json=mock_response
{
"dataset_id": "sample-dataset-1",
"api_url": "http://demo.ckan.org/api/3/action/",
"version": "1707387710",
},
) )
]
expected = {
"dataset_id": "1234",
"api_url": "http://demo.ckan.org/api/3/action/",
"version": "1709043354",
}
assert CKAN().detect("http://demo.ckan.org/dataset/1234") == expected
@pytest.mark.parametrize("test_input, expected", test_hosts) def test_detect_not_ckan():
def test_detect_ckan(test_input, expected):
assert CKAN().detect(test_input[0]) == expected
# Don't trigger the CKAN content provider # Don't trigger the CKAN content provider
assert CKAN().detect("/some/path/here") is None assert CKAN().detect("/some/path/here") is None
assert CKAN().detect("https://example.com/path/here") is None assert CKAN().detect("https://example.com/path/here") is None