try:
self.if_name = InterfaceName(parts[0])
except ValueError as e2:
- raise ValueError(
- f"expected IP address or interface name, got '{parts[0]}'.", object_path
- ) from e1 and e2
+ raise ValueError(f"expected IP address or interface name, got '{parts[0]}'.", object_path) from (
+ e1 and e2
+ )
self.port = PortNumber.from_str(parts[1], object_path)
else:
raise ValueError(f"expected '<ip-address|interface-name>@<port>', got '{source_value}'.", object_path)
try:
self.if_name = InterfaceName(parts[0])
except ValueError as e2:
- raise ValueError(
- f"expected IP address or interface name, got '{parts[0]}'.", object_path
- ) from e1 and e2
+ raise ValueError(f"expected IP address or interface name, got '{parts[0]}'.", object_path) from (
+ e1 and e2
+ )
if len(parts) == 2:
self.port = PortNumber.from_str(parts[1], object_path)
else:
await self._collect_already_running_workers()
async def reset_workers_policy_rules(self, _config: KresConfig) -> None:
-
# command all running 'kresd' workers to reset their old policy rules,
# unless the workers have already been started with a new config so reset is not needed
if self._workers_reset_needed and get_registered_workers_kresids():
)
async def set_new_tls_sticket_secret(self, config: KresConfig) -> None:
-
if config.network.tls.sticket_secret or config.network.tls.sticket_secret_file:
logger.debug("User-configured TLS resumption secret found - skipping auto-generation.")
return
elif self._fix_counter.is_too_high():
logger.error(f"Failed to apply config: {e}")
logger.error("There have already been problems recently, refusing to try to fix it.")
- await self.forced_shutdown() # possible improvement - the person who requested this change won't get a response this way
+ await (
+ self.forced_shutdown()
+ ) # possible improvement - the person who requested this change won't get a response this way
else:
logger.error(f"Failed to apply config: {e}")
logger.warning("Reloading system state and trying again.")
logger = logging.getLogger(__name__)
if _prometheus_client:
-
from prometheus_client import exposition # type: ignore
from prometheus_client.bridge.graphite import GraphiteBridge # type: ignore
- from prometheus_client.core import GaugeMetricFamily # type: ignore
- from prometheus_client.core import REGISTRY, CounterMetricFamily, HistogramMetricFamily, Metric
+ from prometheus_client.core import (
+ REGISTRY,
+ CounterMetricFamily,
+ GaugeMetricFamily, # type: ignore
+ HistogramMetricFamily,
+ Metric,
+ )
_graphite_bridge: Optional[GraphiteBridge] = None
raise web.HTTPMovedPermanently("/metrics/json")
async def _handler_metrics_json(self, _request: web.Request) -> web.Response:
-
config = self.config_store.get()
return web.Response(
)
async def _handler_metrics_prometheus(self, _request: web.Request) -> web.Response:
-
metrics_report = await metrics.report_prometheus()
if not metrics_report:
raise web.HTTPNotFound()
from poetry.core.factory import Factory
from poetry.core.masonry.builders.sdist import SdistBuilder
except (ImportError, ModuleNotFoundError):
- from poetry.masonry.builders.sdist import SdistBuilder
from poetry.factory import Factory
+ from poetry.masonry.builders.sdist import SdistBuilder
from poetry.__version__ import __version__
except (ImportError, ModuleNotFoundError) as ee:
raise ImportError(f"install poetry by doing pip install poetry to use this script: {ee}")
('""', r"\"\""),
("''", r"\'\'"),
# fmt: off
- ('\"\"', r'\"\"'),
- ("\'\'", r'\'\''),
+ ('""', r"\"\""),
+ ("''", r"\'\'"),
# fmt: on
],
)
[(5353, True, False), (53, True, True)],
)
def test_forward_invalid(port: int, auth: bool, tls: bool):
-
if not tls:
with raises(DataValidationError):
ForwardSchema(
("\a\b\f\n\r\t\v\\", r"\a\b\f\n\r\t\v\\"),
# fmt: off
("''", r"\'\'"),
- ('""', r'\"\"'),
- ("\'\'", r"\'\'"),
- ('\"\"', r'\"\"'),
- ('\\"\\"', r'\\\"\\\"'),
+ ('""', r"\"\""),
+ ("''", r"\'\'"),
+ ('""', r"\"\""),
+ ('\\"\\"', r"\\\"\\\""),
("\\'\\'", r"\\\'\\\'"),
# fmt: on
],
def test_version():
-
with open("pyproject.toml", "r") as f:
pyproject = toml.load(f)