kopia lustrzana https://github.com/tsileo/little-boxes
More tests and cleanup
rodzic
28dc4a12a2
commit
62698a7724
|
@ -1,5 +1,6 @@
|
||||||
pytest
|
pytest
|
||||||
pytest-cov
|
pytest-cov
|
||||||
|
httpretty
|
||||||
black
|
black
|
||||||
codecov
|
codecov
|
||||||
flake8
|
flake8
|
||||||
|
|
|
@ -12,7 +12,7 @@ from .errors import NotFromOutboxError
|
||||||
|
|
||||||
# from .errors import ActivityNotFoundError
|
# from .errors import ActivityNotFoundError
|
||||||
# from .urlutils import check_url
|
# from .urlutils import check_url
|
||||||
from .utils import parse_collection
|
from .collection import parse_collection
|
||||||
from .backend import Backend
|
from .backend import Backend
|
||||||
|
|
||||||
from typing import List
|
from typing import List
|
||||||
|
|
|
@ -62,6 +62,10 @@ class Backend(abc.ABC):
|
||||||
def inbox_update(self, as_actor: "ap.Person", activity: "ap.Update") -> None:
|
def inbox_update(self, as_actor: "ap.Person", activity: "ap.Update") -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@abc.abstractmethod
|
||||||
|
def outbox_update(self, as_actor: "ap.Person", activity: "ap.Update") -> None:
|
||||||
|
pass
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def inbox_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
def inbox_like(self, as_actor: "ap.Person", activity: "ap.Like") -> None:
|
||||||
pass
|
pass
|
||||||
|
|
|
@ -28,10 +28,16 @@ def is_url_valid(url: str) -> bool:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
try:
|
try:
|
||||||
ip_address = socket.getaddrinfo(parsed.hostname, parsed.port or 80)[0][4][0]
|
ip_address = ipaddress.ip_address(parsed.hostname)
|
||||||
except socket.gaierror:
|
except ValueError:
|
||||||
logger.exception(f"failed to lookup url {url}")
|
try:
|
||||||
return False
|
ip_address = socket.getaddrinfo(parsed.hostname, parsed.port or 80)[0][4][0]
|
||||||
|
logger.debug(f"dns lookup: {parsed.hostname} -> {ip_address}")
|
||||||
|
except socket.gaierror:
|
||||||
|
logger.exception(f"failed to lookup url {url}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
logger.debug(f"{ip_address}")
|
||||||
|
|
||||||
if ipaddress.ip_address(ip_address).is_private:
|
if ipaddress.ip_address(ip_address).is_private:
|
||||||
logger.info(f"rejecting private URL {url}")
|
logger.info(f"rejecting private URL {url}")
|
||||||
|
@ -41,6 +47,7 @@ def is_url_valid(url: str) -> bool:
|
||||||
|
|
||||||
|
|
||||||
def check_url(url: str) -> None:
|
def check_url(url: str) -> None:
|
||||||
|
logger.debug(f"check_url {url}")
|
||||||
if not is_url_valid(url):
|
if not is_url_valid(url):
|
||||||
raise InvalidURLError(f'"{url}" is invalid')
|
raise InvalidURLError(f'"{url}" is invalid')
|
||||||
|
|
||||||
|
|
|
@ -235,10 +235,12 @@ class InMemBackend(Backend):
|
||||||
def outbox_delete(self, as_actor: ap.Person, activity: ap.Delete) -> None:
|
def outbox_delete(self, as_actor: ap.Person, activity: ap.Delete) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@track_call
|
||||||
def inbox_update(self, as_actor: ap.Person, activity: ap.Update) -> None:
|
def inbox_update(self, as_actor: ap.Person, activity: ap.Update) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def outbox_update(self, activity: ap.Update) -> None:
|
@track_call
|
||||||
|
def outbox_update(self, as_actor: ap.Person, activity: ap.Update) -> None:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@track_call
|
@track_call
|
||||||
|
|
Ładowanie…
Reference in New Issue