Skip to content

Commit

Permalink
chore: Move more functions to gallia.net
Browse files Browse the repository at this point in the history
  • Loading branch information
rumpelsepp committed Dec 19, 2024
1 parent 0ec576d commit bb87195
Show file tree
Hide file tree
Showing 5 changed files with 39 additions and 38 deletions.
3 changes: 2 additions & 1 deletion src/gallia/dumpcap.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,9 @@
from urllib.parse import urlparse

from gallia.log import get_logger
from gallia.net import split_host_port
from gallia.transports import TargetURI, TransportScheme
from gallia.utils import auto_int, handle_task_error, set_task_handler_ctx_variable, split_host_port
from gallia.utils import auto_int, handle_task_error, set_task_handler_ctx_variable

logger = get_logger(__name__)

Expand Down
35 changes: 35 additions & 0 deletions src/gallia/net.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@
#
# SPDX-License-Identifier: Apache-2.0

import ipaddress
import subprocess
from urllib.parse import urlparse

import pydantic
from pydantic.networks import IPvAnyAddress
Expand All @@ -13,6 +15,39 @@
logger = get_logger(__name__)


def split_host_port(
hostport: str,
default_port: int | None = None,
) -> tuple[str, int | None]:
"""Splits a combination of ip address/hostname + port into hostname/ip address
and port. The default_port argument can be used to return a port if it is
absent in the hostport argument."""
# Special case: If hostport is an ipv6 then the urlparser does some weird
# things with the colons and tries to parse ports. Catch this case early.
host = ""
port = default_port
try:
# If hostport is a valid ip address (v4 or v6) there
# is no port included
host = str(ipaddress.ip_address(hostport))
except ValueError:
pass

# Only parse if hostport is not a valid ip address.
if host == "":
# urlparse() and urlsplit() insists on absolute URLs starting with "//".
url = urlparse(f"//{hostport}")
host = url.hostname if url.hostname else url.netloc
port = url.port if url.port else default_port
return host, port


def join_host_port(host: str, port: int) -> str:
if ":" in host:
return f"[{host}]:port"
return f"{host}:{port}"


class AddrInfo(pydantic.BaseModel):
family: str
local: IPvAnyAddress
Expand Down
2 changes: 1 addition & 1 deletion src/gallia/transports/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse

from gallia.log import get_logger
from gallia.net import join_host_port
from gallia.transports.schemes import TransportScheme
from gallia.utils import join_host_port

logger = get_logger(__name__)

Expand Down
35 changes: 0 additions & 35 deletions src/gallia/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
import asyncio
import contextvars
import importlib.util
import ipaddress
import logging
import re
import sys
Expand All @@ -16,7 +15,6 @@
from pathlib import Path
from types import ModuleType
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse

import aiofiles

Expand Down Expand Up @@ -45,39 +43,6 @@ def strtobool(val: str) -> bool:
raise ValueError(f"invalid truth value {val!r}")


def split_host_port(
hostport: str,
default_port: int | None = None,
) -> tuple[str, int | None]:
"""Splits a combination of ip address/hostname + port into hostname/ip address
and port. The default_port argument can be used to return a port if it is
absent in the hostport argument."""
# Special case: If hostport is an ipv6 then the urlparser does some weird
# things with the colons and tries to parse ports. Catch this case early.
host = ""
port = default_port
try:
# If hostport is a valid ip address (v4 or v6) there
# is no port included
host = str(ipaddress.ip_address(hostport))
except ValueError:
pass

# Only parse if hostport is not a valid ip address.
if host == "":
# urlparse() and urlsplit() insists on absolute URLs starting with "//".
url = urlparse(f"//{hostport}")
host = url.hostname if url.hostname else url.netloc
port = url.port if url.port else default_port
return host, port


def join_host_port(host: str, port: int) -> str:
if ":" in host:
return f"[{host}]:port"
return f"{host}:{port}"


def camel_to_snake(s: str) -> str:
"""Convert a CamelCase string to a snake_case string."""
# https://stackoverflow.com/a/1176023
Expand Down
2 changes: 1 addition & 1 deletion tests/pytest/test_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,11 @@
import pytest

from gallia.log import setup_logging
from gallia.net import split_host_port
from gallia.services.uds.core.utils import (
address_and_size_length,
uds_memory_parameters,
)
from gallia.utils import split_host_port

setup_logging()

Expand Down

0 comments on commit bb87195

Please sign in to comment.