ability to ping DelBar API
This commit is contained in:
parent
6d650d718c
commit
7affec30ae
3
.gitignore
vendored
3
.gitignore
vendored
@ -2,6 +2,9 @@
|
||||
prototypes/
|
||||
data/
|
||||
|
||||
# credentials
|
||||
CREDENTIALS*
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
|
||||
0
src/delta_barth/api/__init__.py
Normal file
0
src/delta_barth/api/__init__.py
Normal file
38
src/delta_barth/api/common.py
Normal file
38
src/delta_barth/api/common.py
Normal file
@ -0,0 +1,38 @@
|
||||
import re
|
||||
from typing import Final
|
||||
|
||||
import requests
|
||||
from requests import Response
|
||||
|
||||
from delta_barth.errors import UnspecifiedRequestType
|
||||
from delta_barth.types import HttpRequestTypes
|
||||
|
||||
|
||||
def _strip_url_components(string: str) -> str:
|
||||
return re.sub(r"^[ /]+|[ /]+$", "", string)
|
||||
|
||||
|
||||
def combine_route(base_url: str, route: str) -> str:
|
||||
base_url = _strip_url_components(base_url)
|
||||
route = _strip_url_components(route)
|
||||
return "/".join((base_url, route))
|
||||
|
||||
|
||||
def ping(
|
||||
base_url: str,
|
||||
method: HttpRequestTypes,
|
||||
) -> Response:
|
||||
ROUTE: Final[str] = "ping"
|
||||
URL: Final = combine_route(base_url, ROUTE)
|
||||
|
||||
resp: Response
|
||||
if method == HttpRequestTypes.GET:
|
||||
resp = requests.get(URL)
|
||||
elif method == HttpRequestTypes.PUT:
|
||||
resp = requests.put(URL)
|
||||
elif method == HttpRequestTypes.DELETE:
|
||||
resp = requests.delete(URL)
|
||||
else:
|
||||
raise UnspecifiedRequestType(f"Request type {method} not defined for endpoint")
|
||||
|
||||
return resp
|
||||
2
src/delta_barth/errors.py
Normal file
2
src/delta_barth/errors.py
Normal file
@ -0,0 +1,2 @@
|
||||
class UnspecifiedRequestType(Exception):
|
||||
"""exception raised if for a given API endpoint a not defined operation is requested"""
|
||||
@ -5,6 +5,14 @@ from typing import TypeAlias
|
||||
import pandas as pd
|
||||
|
||||
|
||||
# ** API
|
||||
class HttpRequestTypes(enum.StrEnum):
|
||||
GET = enum.auto()
|
||||
POST = enum.auto()
|
||||
PUT = enum.auto()
|
||||
DELETE = enum.auto()
|
||||
|
||||
|
||||
# ** forecasts
|
||||
@dataclass(slots=True)
|
||||
class CustomerDataSalesForecast:
|
||||
|
||||
0
tests/api/__init__.py
Normal file
0
tests/api/__init__.py
Normal file
56
tests/api/test_common.py
Normal file
56
tests/api/test_common.py
Normal file
@ -0,0 +1,56 @@
|
||||
import pytest
|
||||
|
||||
from delta_barth.api import common
|
||||
from delta_barth.errors import UnspecifiedRequestType
|
||||
from delta_barth.types import HttpRequestTypes
|
||||
|
||||
"http://test.com/ "
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
["case", "expect"],
|
||||
[
|
||||
("http://test.com/ ", "http://test.com"),
|
||||
("http://test.com/", "http://test.com"),
|
||||
("http://test.com ", "http://test.com"),
|
||||
("http://test.com// ", "http://test.com"),
|
||||
("http://test.com", "http://test.com"),
|
||||
(" /http://test.com", "http://test.com"),
|
||||
(" //http://test.com", "http://test.com"),
|
||||
("//http://test.com", "http://test.com"),
|
||||
],
|
||||
)
|
||||
def test_strip_url_components(case, expect):
|
||||
res = common._strip_url_components(case)
|
||||
assert res == expect
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
["base", "route", "expect"],
|
||||
[
|
||||
("http://test.com/ ", "ping", "http://test.com/ping"),
|
||||
("http://test.com/", "ping ", "http://test.com/ping"),
|
||||
("http://test.com ", "ping/", "http://test.com/ping"),
|
||||
("http://test.com// ", "ping", "http://test.com/ping"),
|
||||
("http://test.com", "/ping ", "http://test.com/ping"),
|
||||
(" /http://test.com", "/ ping/ ", "http://test.com/ping"),
|
||||
(" //http://test.com", "ping", "http://test.com/ping"),
|
||||
("//http://test.com", "// ping// ", "http://test.com/ping"),
|
||||
],
|
||||
)
|
||||
def test_combine_route(base, route, expect):
|
||||
res = common.combine_route(base, route)
|
||||
assert res == expect
|
||||
|
||||
|
||||
@pytest.mark.api_con_required
|
||||
def test_ping(api_base_url):
|
||||
resp = common.ping(api_base_url, HttpRequestTypes.GET)
|
||||
assert resp.status_code == 204
|
||||
resp = common.ping(api_base_url, HttpRequestTypes.PUT)
|
||||
assert resp.status_code == 204
|
||||
resp = common.ping(api_base_url, HttpRequestTypes.DELETE)
|
||||
assert resp.status_code == 204
|
||||
|
||||
with pytest.raises(UnspecifiedRequestType):
|
||||
resp = common.ping(api_base_url, HttpRequestTypes.POST)
|
||||
@ -1,10 +1,30 @@
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, cast
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def credentials() -> dict[str, str]:
|
||||
pwd = Path.cwd()
|
||||
assert "barth" in pwd.parent.name.lower(), "not in project root directory"
|
||||
creds_pth = pwd / "./CREDENTIALS.toml"
|
||||
assert creds_pth.exists(), "file to credentials data not found"
|
||||
with open(creds_pth, "rb") as file:
|
||||
cfg = tomllib.load(file)
|
||||
|
||||
creds = cast(dict[str, str], cfg["delta-barth-server"]["api"])
|
||||
|
||||
return creds
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def api_base_url(credentials) -> str:
|
||||
return credentials["base_url"]
|
||||
|
||||
|
||||
# TODO: maybe include in main package depending if needed in future
|
||||
def _cvt_str_float(value: str) -> float:
|
||||
import locale
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user