Add an 'external_server' controller

This commit is contained in:
Valentin Lorentz 2021-08-10 18:15:45 +02:00 committed by Val Lorentz
parent c37ed0f218
commit 42e10c3848
3 changed files with 56 additions and 3 deletions

View File

@ -7,12 +7,13 @@ import socket
import subprocess
import tempfile
import time
from typing import IO, Any, Callable, Dict, List, Optional, Set, Type
from typing import IO, Any, Callable, Dict, List, Optional, Set, Tuple, Type
import irctest
from . import authentication, tls
from .client_mock import ClientMock
from .irc_utils.junkdrawer import find_hostname_and_port
from .irc_utils.message_parser import Message
from .runner import NotImplementedByController
@ -187,6 +188,9 @@ class BaseServerController(_BaseController):
extban_mute_char: Optional[str] = None
"""Character used for the 'mute' extban"""
def get_hostname_and_port(self) -> Tuple[str, int]:
return find_hostname_and_port()
def run(
self,
hostname: str,

View File

@ -30,7 +30,6 @@ from .authentication import Authentication
from .basecontrollers import TestCaseControllerConfig
from .exceptions import ConnectionClosed
from .irc_utils import capabilities, message_parser
from .irc_utils.junkdrawer import find_hostname_and_port
from .irc_utils.message_parser import Message
from .irc_utils.sasl import sasl_plain_blob
from .numerics import (
@ -504,7 +503,7 @@ class BaseServerTestCase(
def setUp(self) -> None:
super().setUp()
self.server_support = None
(self.hostname, self.port) = find_hostname_and_port()
(self.hostname, self.port) = self.controller.get_hostname_and_port()
self.controller.run(
self.hostname,
self.port,

View File

@ -0,0 +1,50 @@
import os
from typing import Optional, Set, Tuple, Type
from irctest.basecontrollers import BaseServerController
class ExternalServerController(BaseServerController):
"""Dummy controller that doesn't run a server.
Instead, it allows connecting to servers ran outside irctest."""
software_name = "unknown external server"
supported_sasl_mechanisms = set(
os.environ.get("IRCTEST_SERVER_SASL_MECHS", "").split()
)
def check_is_alive(self) -> None:
pass
def kill_proc(self) -> None:
pass
def wait_for_port(self) -> None:
pass
def get_hostname_and_port(self) -> Tuple[str, int]:
hostname = os.environ.get("IRCTEST_SERVER_HOSTNAME")
port = os.environ.get("IRCTEST_SERVER_PORT")
if not hostname or not port:
raise RuntimeError(
"Please set IRCTEST_SERVER_HOSTNAME and IRCTEST_SERVER_PORT."
)
return (hostname, int(port))
def run(
self,
hostname: str,
port: int,
*,
password: Optional[str],
ssl: bool,
run_services: bool,
valid_metadata_keys: Optional[Set[str]] = None,
invalid_metadata_keys: Optional[Set[str]] = None,
restricted_metadata_keys: Optional[Set[str]] = None,
) -> None:
pass
def get_irctest_controller_class() -> Type[ExternalServerController]:
return ExternalServerController